fix: Show user-friendly messages for GCP resource exhaustion errors

Fixes #901

When GCP fails to create a VM (e.g., due to zone resource exhaustion),
the error was logged but the test status showed a raw error dict that
was not user-friendly.

Changes:
- Added parse_gcp_error() helper function to extract meaningful messages
  from GCP API error responses
- Added GCP_ERROR_MESSAGES dict mapping known error codes to user-friendly
  messages (ZONE_RESOURCE_POOL_EXHAUSTED, QUOTA_EXCEEDED, TIMEOUT, etc.)
- Updated start_test() to use parse_gcp_error() when VM creation fails
- For unknown errors, shows the error code and a truncated message

Now users see messages like:
"GCP resources temporarily unavailable in the configured zone.
The test will be retried automatically when resources become available."

Instead of raw dicts like:
"{'errors': [{'code': 'ZONE_RESOURCE_POOL_EXHAUSTED', ...}]}"

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: cfsmp3

mod_ci/controllers.py

@@ -122,6 +122,67 @@ def safe_db_commit(db, operation_description: str = "database operation") -> boo
         return False
 
 
+# User-friendly messages for known GCP error codes
+GCP_ERROR_MESSAGES = {
+    'ZONE_RESOURCE_POOL_EXHAUSTED': (
+        "GCP resources temporarily unavailable in the configured zone. "
+        "The test will be retried automatically when resources become available."
+    ),
+    'QUOTA_EXCEEDED': (
+        "GCP quota limit reached. Please wait for other tests to complete "
+        "or contact the administrator."
+    ),
+    'RESOURCE_NOT_FOUND': "Required GCP resource not found. Please contact the administrator.",
+    'RESOURCE_ALREADY_EXISTS': "A VM with this name already exists. Please contact the administrator.",
+    'TIMEOUT': "GCP operation timed out. The test will be retried automatically.",
+}
+
+
+def parse_gcp_error(result: Dict) -> str:
+    """
+    Parse a GCP API error response and return a user-friendly message.
+
+    GCP errors have the structure:
+    {
+        'error': {
+            'errors': [{'code': 'ERROR_CODE', 'message': '...'}]
+        }
+    }
+
+    :param result: The GCP API response dictionary
+    :return: A user-friendly error message
+    """
+    if not isinstance(result, dict):
+        return f"Unknown error: {result}"
+
+    error = result.get('error')
+    if error is None:
+        return "Unknown error (no error details provided)"
+
+    if not isinstance(error, dict):
+        return f"Unknown error: {error}"
+
+    errors = error.get('errors', [])
+    if not errors:
+        return f"Unknown error: {error}"
+
+    # Get the first error (usually the most relevant)
+    first_error = errors[0] if isinstance(errors, list) and len(errors) > 0 else {}
+    error_code = first_error.get('code', 'UNKNOWN')
+    error_message = first_error.get('message', 'No details provided')
+
+    # Check if we have a user-friendly message for this error code
+    if error_code in GCP_ERROR_MESSAGES:
+        return GCP_ERROR_MESSAGES[error_code]
+
+    # For unknown errors, return a cleaned-up version of the original message
+    # Truncate very long messages
+    if len(error_message) > 200:
+        error_message = error_message[:200] + "..."
+
+    return f"{error_code}: {error_message}"
+
+
 mod_ci = Blueprint('ci', __name__)
 
 
@@ -782,9 +843,9 @@ def start_test(compute, app, db, repository: Repository.Repository, test, bot_to
         if not safe_db_commit(db, f"recording GCP instance for test {test.id}"):
             log.error(f"Failed to record GCP instance for test {test.id}, but VM was created")
     else:
-        error_msg = result.get('error', 'Unknown error') if isinstance(result, dict) else str(result)
+        error_msg = parse_gcp_error(result)
         log.error(f"Error creating test instance for test {test.id}, result: {result}")
-        mark_test_failed(db, test, repository, f"Failed to create VM: {error_msg}")
+        mark_test_failed(db, test, repository, error_msg)
 
 
 def create_instance(compute, project, zone, test, reportURL) -> Dict:

tests/test_ci/test_controllers.py

@@ -1,4 +1,5 @@
 import json
+import unittest
 from importlib import reload
 from unittest import mock
 from unittest.mock import MagicMock
@@ -3092,3 +3093,132 @@ def test_mark_test_failed_includes_target_url(self, mock_progress, mock_update_g
         call_args = mock_update_github.call_args
         self.assertEqual(len(call_args[0]), 5)  # 5 positional args including target_url
         self.assertIn("456", call_args[0][4])  # target_url contains test ID
+
+
+class TestParseGcpError(unittest.TestCase):
+    """Tests for the parse_gcp_error helper function."""
+
+    def test_parse_gcp_error_zone_resource_exhausted(self):
+        """Test that ZONE_RESOURCE_POOL_EXHAUSTED returns user-friendly message."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {
+            'status': 'DONE',
+            'error': {
+                'errors': [{
+                    'code': 'ZONE_RESOURCE_POOL_EXHAUSTED',
+                    'message': "The zone 'projects/test/zones/us-central1-a' does not "
+                               "have enough resources available to fulfill the request."
+                }]
+            }
+        }
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("GCP resources temporarily unavailable", error_msg)
+        self.assertIn("retried automatically", error_msg)
+        # Should NOT contain raw technical details
+        self.assertNotIn("us-central1-a", error_msg)
+
+    def test_parse_gcp_error_quota_exceeded(self):
+        """Test that QUOTA_EXCEEDED returns user-friendly message."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {
+            'error': {
+                'errors': [{
+                    'code': 'QUOTA_EXCEEDED',
+                    'message': 'Quota exceeded for resource.'
+                }]
+            }
+        }
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("quota limit reached", error_msg)
+
+    def test_parse_gcp_error_timeout(self):
+        """Test that TIMEOUT returns user-friendly message."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {
+            'status': 'TIMEOUT',
+            'error': {
+                'errors': [{
+                    'code': 'TIMEOUT',
+                    'message': 'Operation timed out after 1800 seconds'
+                }]
+            }
+        }
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("timed out", error_msg)
+        self.assertIn("retried automatically", error_msg)
+
+    def test_parse_gcp_error_unknown_code(self):
+        """Test that unknown error codes return the code and message."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {
+            'error': {
+                'errors': [{
+                    'code': 'SOME_NEW_ERROR',
+                    'message': 'Something unexpected happened.'
+                }]
+            }
+        }
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("SOME_NEW_ERROR", error_msg)
+        self.assertIn("Something unexpected happened", error_msg)
+
+    def test_parse_gcp_error_long_message_truncated(self):
+        """Test that very long error messages are truncated."""
+        from mod_ci.controllers import parse_gcp_error
+
+        long_message = "A" * 300  # 300 characters
+        result = {
+            'error': {
+                'errors': [{
+                    'code': 'UNKNOWN_ERROR',
+                    'message': long_message
+                }]
+            }
+        }
+
+        error_msg = parse_gcp_error(result)
+        self.assertLess(len(error_msg), 250)  # Should be truncated
+        self.assertIn("...", error_msg)
+
+    def test_parse_gcp_error_no_error_key(self):
+        """Test handling when 'error' key is missing."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {'status': 'DONE'}
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("Unknown error", error_msg)
+
+    def test_parse_gcp_error_empty_errors_list(self):
+        """Test handling when 'errors' list is empty."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {'error': {'errors': []}}
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("Unknown error", error_msg)
+
+    def test_parse_gcp_error_not_a_dict(self):
+        """Test handling when result is not a dictionary."""
+        from mod_ci.controllers import parse_gcp_error
+
+        error_msg = parse_gcp_error("some string error")
+        self.assertIn("Unknown error", error_msg)
+        self.assertIn("some string error", error_msg)
+
+    def test_parse_gcp_error_error_not_a_dict(self):
+        """Test handling when 'error' value is not a dictionary."""
+        from mod_ci.controllers import parse_gcp_error
+
+        result = {'error': 'just a string'}
+
+        error_msg = parse_gcp_error(result)
+        self.assertIn("Unknown error", error_msg)