[Better Tests] Add error injection and runtime overrides for GCE metadata fake values

configs/test/swarming/swarming.yaml

@@ -89,7 +89,7 @@ mapping:
         value: 'VALUE1'
       - key: 'ENV_VAR2'
         value: 'VALUE2'
-swarming_server: 'server-name'
+swarming_server: 'http://127.0.0.1:9014'
 swarming_pool: 'pool-name'
 swarming_realm: 'realm-name'
 logs_project_id: 'project_id'

src/clusterfuzz/_internal/swarming/__init__.py

@@ -26,7 +26,6 @@
 from clusterfuzz._internal.config import local_config
 from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.google_cloud_utils import compute_metadata
-from clusterfuzz._internal.google_cloud_utils import credentials
 from clusterfuzz._internal.metrics import logs
 from clusterfuzz._internal.protos import swarming_pb2
 from clusterfuzz._internal.system import environment

src/clusterfuzz/_internal/swarming/api.py

@@ -14,33 +14,20 @@
 """Swarming pRPC API client."""
 
 from typing import Optional
+import urllib.parse
 
 from google.auth import exceptions as auth_exceptions
 from google.protobuf import json_format
-from google.protobuf.timestamp_pb2 import \
-    Timestamp  # pylint: disable=no-name-in-module
 from requests.exceptions import HTTPError
 
 from clusterfuzz._internal.base import utils
 from clusterfuzz._internal.config.local_config import SwarmingConfig
 from clusterfuzz._internal.google_cloud_utils import credentials
 from clusterfuzz._internal.metrics import logs
 from clusterfuzz._internal.protos import swarming_pb2
+from clusterfuzz._internal.swarming import constants
 from clusterfuzz._internal.swarming import get_swarming_config
 
-# TODO(b/516627559): Move scopes to config file
-_SWARMING_SCOPES = [
-    'https://www.googleapis.com/auth/cloud-platform',
-    'https://www.googleapis.com/auth/userinfo.email'
-]
-
-_COUNT_TASKS_ENDPOINT = 'swarming.v2.Tasks/CountTasks'
-_NEW_TASK_ENDPOINT = 'swarming.v2.Tasks/NewTask'
-
-_MIN_TASK_START_TIME = "2026-06-01T00:00:00Z"
-_MIN_TASK_START_TIME_PROTO = json_format.Parse(f'"{_MIN_TASK_START_TIME}"',
-                                               Timestamp())
-
 
 class SwarmingApiError(Exception):
   """Exception raised for errors in the Swarming API."""
@@ -54,7 +41,13 @@ class SwarmingApi:
 
   def __init__(self, config: SwarmingConfig):
     self._config = config
-    self._base_url = f"https://{self._config.get('swarming_server')}/prpc/"
+    server = self._config.get('swarming_server')
+    parsed = urllib.parse.urlsplit(server) if isinstance(server, str) else None
+    if (not parsed or parsed.scheme not in ('http', 'https') or
+        not parsed.netloc):
+      raise ValueError(f'Invalid swarming_server URL: {server!r}. '
+                       'Must be a valid http:// or https:// URL.')
+    self._base_url = server.rstrip('/')
 
   @staticmethod
   def create() -> Optional['SwarmingApi']:
@@ -73,7 +66,7 @@ def _get_token(self) -> str:
     """Gets a valid token for the Swarming API.  Returns "" if it fails."""
     try:
       creds = credentials.get_scoped_service_account_credentials(
-          _SWARMING_SCOPES)
+          constants.SWARMING_SCOPES)
       if not creds:
         logs.error('[Swarming] Failed to get credentials. None found.')
         return ""
@@ -129,8 +122,8 @@ def _make_request(self, endpoint: str, body: str) -> str | None:
       return None
 
     # Strip XSSI prefix if present.
-    if response.startswith(")]}'\n"):
-      response = response[len(")]}'\n"):]
+    if response.startswith(constants.XSSI_PREFIX):
+      response = response[len(constants.XSSI_PREFIX):]
 
     return response
 
@@ -152,7 +145,8 @@ def push_task(
     message_body = json_format.MessageToJson(task_request)
 
     try:
-      raw_response = self._make_request(_NEW_TASK_ENDPOINT, message_body)
+      raw_response = self._make_request(constants.NEW_TASK_ENDPOINT,
+                                        message_body)
     except HTTPError as e:
       raise SwarmingApiError(f'HTTP error calling push_task: {e}') from e
 
@@ -185,11 +179,12 @@ def count_tasks(
       SwarmingApiError: If the pRPC request fails or response parsing fails.
     """
     if not count_request.HasField('start'):
-      count_request.start.CopyFrom(_MIN_TASK_START_TIME_PROTO)
+      count_request.start.CopyFrom(constants.MIN_TASK_START_TIME_PROTO)
     message_body = json_format.MessageToJson(count_request)
 
     try:
-      response_str = self._make_request(_COUNT_TASKS_ENDPOINT, message_body)
+      response_str = self._make_request(constants.COUNT_TASKS_ENDPOINT,
+                                        message_body)
     except HTTPError as e:
       raise SwarmingApiError(f"HTTP error calling count_tasks: {e}") from e
 

src/clusterfuzz/_internal/swarming/constants.py

@@ -0,0 +1,33 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Constants for Swarming API interaction."""
+
+from google.protobuf import json_format
+from google.protobuf.timestamp_pb2 import \
+    Timestamp  # pylint: disable=no-name-in-module
+
+# TODO(b/516627559): Move scopes to config file
+SWARMING_SCOPES = [
+    'https://www.googleapis.com/auth/cloud-platform',
+    'https://www.googleapis.com/auth/userinfo.email',
+]
+
+COUNT_TASKS_ENDPOINT = '/prpc/swarming.v2.Tasks/CountTasks'
+NEW_TASK_ENDPOINT = '/prpc/swarming.v2.Tasks/NewTask'
+
+XSSI_PREFIX = ")]}'\n"
+
+MIN_TASK_START_TIME = '2026-06-01T00:00:00Z'
+MIN_TASK_START_TIME_PROTO = json_format.Parse(f'"{MIN_TASK_START_TIME}"',
+                                              Timestamp())

src/clusterfuzz/_internal/tests/core/swarming/api_test.py

@@ -73,7 +73,7 @@ def test_push_task(self):
         'Content-Type': 'application/json',
         'Authorization': 'Bearer fake_token'
     }
-    expected_url = 'https://server-name/prpc/swarming.v2.Tasks/NewTask'
+    expected_url = 'http://127.0.0.1:9014/prpc/swarming.v2.Tasks/NewTask'
     self.mock.post_url.assert_called_with(
         url=expected_url,
         data=json_format.MessageToJson(task_request),
@@ -92,7 +92,7 @@ def test_count_tasks(self):
         'Content-Type': 'application/json',
         'Authorization': 'Bearer fake_token'
     }
-    expected_url = 'https://server-name/prpc/swarming.v2.Tasks/CountTasks'
+    expected_url = 'http://127.0.0.1:9014/prpc/swarming.v2.Tasks/CountTasks'
 
     expected_request = swarming_pb2.TasksCountRequest(tags=['tag1'])
     json_format.Parse('"2026-06-01T00:00:00Z"', expected_request.start)
@@ -210,3 +210,29 @@ def test_push_task_parse_error(self):
     self.mock.post_url.return_value = 'invalid json'
     with self.assertRaises(SwarmingApiError):
       self.api.push_task(swarming_pb2.NewTaskRequest())
+
+  def test_init_accepts_http_and_https_urls(self):
+    """Tests that SwarmingApi accepts valid http:// and https:// URLs."""
+    for server, expected_base_url in [
+        ('http://127.0.0.1:9014', 'http://127.0.0.1:9014'),
+        ('https://server-name/', 'https://server-name'),
+    ]:
+      mock_config = mock.MagicMock()
+      mock_config.get.return_value = server
+      api = SwarmingApi(mock_config)
+      self.assertEqual(api._base_url, expected_base_url)  # pylint: disable=protected-access
+
+  def test_init_rejects_malformed_swarming_server(self):
+    """Tests that SwarmingApi raises ValueError when swarming_server is malformed."""
+    for invalid_server in [
+        None,
+        '',
+        'server-name',
+        'ftp://server-name',
+        'http://',
+        'https://',
+    ]:
+      mock_config = mock.MagicMock()
+      mock_config.get.return_value = invalid_server
+      with self.assertRaises(ValueError):
+        SwarmingApi(mock_config)

test_suites/fixtures/common/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

test_suites/fixtures/common/wiremock_errors.py

@@ -0,0 +1,173 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Shared WireMock error injection for service emulators."""
+
+import abc
+from http import HTTPStatus
+import uuid
+
+from wiremock.client import HttpMethods
+from wiremock.client import Mapping
+from wiremock.client import MappingRequest
+from wiremock.client import MappingResponse
+from wiremock.client import Mappings
+from wiremock.constants import Config
+
+_DEFAULT_RESPONSE_HEADERS = {
+    'Content-Type': 'text/plain; charset=utf-8',
+}
+
+
+class ErrorInjector(abc.ABC):
+  """Interface for injecting and clearing HTTP errors in service emulators."""
+
+  @abc.abstractmethod
+  def inject_error(
+      self,
+      path: str,
+      status: int = HTTPStatus.INTERNAL_SERVER_ERROR,
+      times: int = 1,
+      delay_seconds: float = 0,
+      method: str = HttpMethods.ANY,
+  ) -> None:
+    """Makes the emulator fail requests to path."""
+
+  @abc.abstractmethod
+  def clear_errors(self) -> None:
+    """Removes every injected error."""
+
+
+class WireMockErrorInjector(ErrorInjector):
+  """Injects and clears HTTP errors in a running WireMock container."""
+
+  def __init__(
+      self,
+      admin_url: str,
+      path_prefix: str = '',
+      request_headers: dict | None = None,
+      response_headers: dict | None = None,
+  ):
+    self._admin_url = admin_url
+    self._path_prefix = path_prefix
+    self._request_headers = request_headers
+    self._response_headers = response_headers or _DEFAULT_RESPONSE_HEADERS
+
+  def _create_mapping(self, mapping: Mapping) -> None:
+    Config.base_url = self._admin_url
+    Mappings.create_mapping(mapping)
+
+  def _all_mappings(self) -> list[Mapping]:
+    Config.base_url = self._admin_url
+    return Mappings.retrieve_all_mappings().mappings
+
+  def _delete_mapping(self, mapping_id: str) -> None:
+    Config.base_url = self._admin_url
+    Mappings.delete_mapping(mapping_id)
+
+  def _error_mapping(
+      self,
+      path: str,
+      status_code: int,
+      method: str = HttpMethods.ANY,
+      delay_ms: int | None = None,
+      scenario_name: str | None = None,
+      required_state: str | None = None,
+      new_state: str | None = None,
+  ) -> Mapping:
+    """Returns a non-persistent Mapping that injects an HTTP error for path."""
+    request_kwargs: dict = {
+        'method': method,
+        'url_path': f'{self._path_prefix}{path}',
+    }
+    if self._request_headers is not None:
+      request_kwargs['headers'] = self._request_headers
+
+    response_kwargs: dict = {
+        'status': status_code,
+        'body': f'injected error for {path}\n',
+        'headers': self._response_headers,
+    }
+    if delay_ms is not None:
+      response_kwargs['fixed_delay_milliseconds'] = delay_ms
+
+    return Mapping(
+        persistent=False,
+        scenario_name=scenario_name,
+        required_scenario_state=required_state,
+        new_scenario_state=new_state,
+        metadata={
+            'error': True,
+            'error_path': path,
+        },
+        request=MappingRequest(**request_kwargs),
+        response=MappingResponse(**response_kwargs),
+    )
+
+  def _remove_existing_errors_for_path(self, path: str) -> None:
+    """Removes any active error mappings targeting path."""
+    for mapping in self._all_mappings():
+      metadata = mapping.metadata or {}
+      if metadata.get('error') and metadata.get('error_path') == path:
+        self._delete_mapping(mapping.id)
+
+  def inject_error(
+      self,
+      path: str,
+      status: int = HTTPStatus.INTERNAL_SERVER_ERROR,
+      times: int = 1,
+      delay_seconds: float = 0,
+      method: str = HttpMethods.ANY,
+  ) -> None:
+    """Makes the emulator fail requests to path.
+
+    Args:
+      path: Endpoint path (relative to path_prefix if configured).
+      status: The HTTP status to return.
+      times: How many requests to affect. 0 means indefinitely until
+        clear_errors() is called.
+      delay_seconds: How long the emulator should stall before responding.
+      method: HTTP method to match (defaults to HttpMethods.ANY).
+    """
+    status_code = status or HTTPStatus.INTERNAL_SERVER_ERROR
+    delay_ms = int(delay_seconds * 1000) if delay_seconds > 0 else None
+    if times < 0:
+      raise ValueError(f'times must be >= 0, got {times}')
+
+    self._remove_existing_errors_for_path(path)
+
+    if times == 0:
+      self._create_mapping(
+          self._error_mapping(
+              path, status_code, method=method, delay_ms=delay_ms))
+      return
+
+    scenario_name = f'error-{path}-{uuid.uuid4().hex}'
+    for i in range(times):
+      self._create_mapping(
+          self._error_mapping(
+              path,
+              status_code,
+              method=method,
+              delay_ms=delay_ms,
+              scenario_name=scenario_name,
+              required_state='Started' if i == 0 else f'step_{i}',
+              new_state=f'step_{i + 1}',
+          ))
+
+  def clear_errors(self) -> None:
+    """Removes every injected error mapping."""
+    for mapping in self._all_mappings():
+      metadata = mapping.metadata or {}
+      if metadata.get('error'):
+        self._delete_mapping(mapping.id)