From bceafb1c0cd6e4a8bfb8c6fbd66409eb082e8dda Mon Sep 17 00:00:00 2001 From: Ivan Barba Date: Wed, 30 Sep 2026 19:15:30 +0000 Subject: [PATCH 1/5] [Better Tests] Add fault injection and runtime overrides to metadata emulator --- test_suites/fixtures/gce_metadata_server.py | 91 ++++++++++++++++++- .../fixtures/gce_metadata_server_test.py | 47 ++++++++++ 2 files changed, 137 insertions(+), 1 deletion(-) diff --git a/test_suites/fixtures/gce_metadata_server.py b/test_suites/fixtures/gce_metadata_server.py index 3dd398d535..654977bc75 100644 --- a/test_suites/fixtures/gce_metadata_server.py +++ b/test_suites/fixtures/gce_metadata_server.py @@ -56,6 +56,7 @@ import json import os import tempfile +import uuid import requests from wiremock.client import HttpMethods @@ -212,6 +213,39 @@ def _json_value_mapping(path: str, ) +def _fault_mapping(path: str, + status_code: int, + 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 fault for path.""" + response_kwargs = { + 'status': status_code, + 'body': f'injected fault for {path}\n', + 'headers': _TEXT_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={ + 'fault': True, + 'fault_path': path, + }, + request=MappingRequest( + method=HttpMethods.GET, + url_path=f'/computeMetadata/v1/{path}', + headers=_REQUIRED_HEADER_MATCHER, + ), + response=MappingResponse(**response_kwargs), + ) + + class MetadataEmulatorClient: """Talks to a running WireMock metadata emulator.""" @@ -219,6 +253,7 @@ def __init__(self, hostport: str, config: dict): self.hostport = hostport self.admin_url = f'http://{hostport}/__admin' self._baseline_instance_attributes = set() + self._runtime_instance_attributes = set() # Loopback traffic must not be handed to an ambient http_proxy, which # would answer with its own error instead of reaching the emulator. @@ -313,7 +348,10 @@ def set_project_attribute(self, _text_value_mapping( f'project/attributes/{key}', value, persistent=persistent)) - if key not in self._baseline_instance_attributes: + overridden_on_instance = ( + key in self._baseline_instance_attributes or + key in self._runtime_instance_attributes) + if not overridden_on_instance: self._create_mapping( _text_value_mapping( f'instance/attributes/{key}', value, persistent=persistent)) @@ -333,6 +371,8 @@ def set_instance_attribute(self, """Sets 'instance/attributes/', shadowing any project attribute.""" if persistent: self._baseline_instance_attributes.add(key) + else: + self._runtime_instance_attributes.add(key) self._create_mapping( _text_value_mapping( f'instance/attributes/{key}', value, persistent=persistent)) @@ -397,6 +437,55 @@ def get(self, path: str) -> str: response.raise_for_status() return response.text + def _remove_existing_faults_for_path(self, path: str) -> None: + """Removes any active fault mappings targeting path.""" + for mapping in self._all_mappings(): + metadata = mapping.metadata or {} + if metadata.get('fault') and metadata.get('fault_path') == path: + self._delete_mapping(mapping.id) + + def inject_fault(self, + path: str, + status: int = 500, + times: int = 1, + delay_seconds: float = 0) -> None: + """Makes the emulator fail a metadata path. + + Args: + path: A metadata path without the /computeMetadata/v1/ prefix. + status: The HTTP status to return. + times: How many requests to affect. 0 means indefinitely until + clear_faults() is called. + delay_seconds: How long the emulator should stall before responding. + """ + status_code = status or 500 + 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_faults_for_path(path) + + if times == 0: + self._create_mapping(_fault_mapping(path, status_code, delay_ms)) + return + + scenario_name = f'fault-{path}-{uuid.uuid4().hex}' + for i in range(times): + self._create_mapping( + _fault_mapping( + path, + status_code, + 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_faults(self) -> None: + """Removes every injected fault and runtime metadata override.""" + self._runtime_instance_attributes.clear() + self._reset_mappings() + def bootstrap() -> None: """Points this process at the address the default emulator will bind. diff --git a/test_suites/fixtures/gce_metadata_server_test.py b/test_suites/fixtures/gce_metadata_server_test.py index 8f32029137..c14894d75a 100644 --- a/test_suites/fixtures/gce_metadata_server_test.py +++ b/test_suites/fixtures/gce_metadata_server_test.py @@ -49,6 +49,12 @@ def _emulators(cls): cls.uworker = uworker yield + @pytest.fixture(autouse=True) + def clean_up_after_each_test(self): + yield + self.tworker.clear_faults() + self.uworker.clear_faults() + def test_default_credentials_come_from_the_trusted_emulator(self): """Verifies that google.auth.default() resolves Compute Engine credentials from the tworker emulator, not uworker.""" @@ -170,6 +176,47 @@ def test_a_second_emulator_cannot_take_a_held_address(self): with gce_metadata_server.untrusted_as_default(): pass + def test_a_fault_stops_after_the_configured_request_count(self): + """Verifies that a fault budgeted for one request breaks only that one.""" + self.tworker.inject_fault('instance/zone', status=500, times=1) + + with pytest.raises(requests.exceptions.HTTPError) as caught: + self.tworker.get('instance/zone') + assert caught.value.response.status_code == 500 + + assert self.tworker.get('instance/zone') == ( + 'projects/1234567890/zones/us-central1-f') + + def test_compute_metadata_get_retries_itself_past_transient_faults(self): + """Verifies that get() retries itself, and only fails past its budget.""" + budget = compute_metadata._RETRIES # pylint: disable=protected-access + + self.tworker.inject_fault('instance/zone', status=500, times=budget) + assert compute_metadata.get('instance/zone') == ( + 'projects/1234567890/zones/us-central1-f') + + self.tworker.inject_fault('instance/zone', status=500, times=budget + 1) + with pytest.raises(requests.exceptions.HTTPError) as caught: + compute_metadata.get('instance/zone') + assert caught.value.response.status_code == 500 + + def test_set_adds_and_overrides_metadata_across_scopes(self): + """Verifies runtime metadata overrides in project and instance scopes.""" + self.tworker.set_project_attribute('custom-attr', 'from-project') + assert compute_metadata.get('project/attributes/custom-attr') == ( + 'from-project') + assert compute_metadata.get('instance/attributes/custom-attr') == ( + 'from-project') + + self.tworker.set_instance_attribute('custom-attr', 'from-instance') + assert compute_metadata.get('instance/attributes/custom-attr') == ( + 'from-instance') + assert compute_metadata.get('project/attributes/custom-attr') == ( + 'from-project') + + self.tworker.set_instance_metadata('preempted', 'TRUE') + assert compute_metadata.get('instance/preempted') == 'TRUE' + class TestGceMetadataEmulatorUntrustedDefault: """Integration tests for a test class that runs as the untrusted worker.""" From 7ddb1345b1a46aa4c0d063924788826036794e9b Mon Sep 17 00:00:00 2001 From: Ivan Barba Date: Wed, 30 Sep 2026 20:54:51 +0000 Subject: [PATCH 2/5] [Better tests] Makes args clearer by passing them as named args in tests --- test_suites/fixtures/gce_metadata_server_test.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/test_suites/fixtures/gce_metadata_server_test.py b/test_suites/fixtures/gce_metadata_server_test.py index c14894d75a..416f96d3b7 100644 --- a/test_suites/fixtures/gce_metadata_server_test.py +++ b/test_suites/fixtures/gce_metadata_server_test.py @@ -202,19 +202,20 @@ def test_compute_metadata_get_retries_itself_past_transient_faults(self): def test_set_adds_and_overrides_metadata_across_scopes(self): """Verifies runtime metadata overrides in project and instance scopes.""" - self.tworker.set_project_attribute('custom-attr', 'from-project') + self.tworker.set_project_attribute(key='custom-attr', value='from-project') assert compute_metadata.get('project/attributes/custom-attr') == ( 'from-project') assert compute_metadata.get('instance/attributes/custom-attr') == ( 'from-project') - self.tworker.set_instance_attribute('custom-attr', 'from-instance') + self.tworker.set_instance_attribute( + key='custom-attr', value='from-instance') assert compute_metadata.get('instance/attributes/custom-attr') == ( 'from-instance') assert compute_metadata.get('project/attributes/custom-attr') == ( 'from-project') - self.tworker.set_instance_metadata('preempted', 'TRUE') + self.tworker.set_instance_metadata(key='preempted', value='TRUE') assert compute_metadata.get('instance/preempted') == 'TRUE' From 6b6ff020978dd297fc17c03fda8b4da5b5fac644 Mon Sep 17 00:00:00 2001 From: Ivan Barba Date: Fri, 2 Oct 2026 20:44:27 +0000 Subject: [PATCH 3/5] [Better Tests] Extract shared WireMock fault injector with composition interface --- test_suites/fixtures/common/__init__.py | 13 ++ .../fixtures/common/wiremock_faults.py | 173 ++++++++++++++++++ test_suites/fixtures/gce_metadata_server.py | 110 ++++------- 3 files changed, 219 insertions(+), 77 deletions(-) create mode 100644 test_suites/fixtures/common/__init__.py create mode 100644 test_suites/fixtures/common/wiremock_faults.py diff --git a/test_suites/fixtures/common/__init__.py b/test_suites/fixtures/common/__init__.py new file mode 100644 index 0000000000..f8cb6d9d20 --- /dev/null +++ b/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. diff --git a/test_suites/fixtures/common/wiremock_faults.py b/test_suites/fixtures/common/wiremock_faults.py new file mode 100644 index 0000000000..db0a5560d8 --- /dev/null +++ b/test_suites/fixtures/common/wiremock_faults.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 fault 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 FaultInjector(abc.ABC): + """Interface for injecting and clearing HTTP faults in service emulators.""" + + @abc.abstractmethod + def inject_fault( + 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_faults(self) -> None: + """Removes every injected fault.""" + + +class WireMockFaultInjector(FaultInjector): + """Injects and clears HTTP faults 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 _fault_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 fault 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 fault 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={ + 'fault': True, + 'fault_path': path, + }, + request=MappingRequest(**request_kwargs), + response=MappingResponse(**response_kwargs), + ) + + def _remove_existing_faults_for_path(self, path: str) -> None: + """Removes any active fault mappings targeting path.""" + for mapping in self._all_mappings(): + metadata = mapping.metadata or {} + if metadata.get('fault') and metadata.get('fault_path') == path: + self._delete_mapping(mapping.id) + + def inject_fault( + 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_faults() 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_faults_for_path(path) + + if times == 0: + self._create_mapping( + self._fault_mapping( + path, status_code, method=method, delay_ms=delay_ms)) + return + + scenario_name = f'fault-{path}-{uuid.uuid4().hex}' + for i in range(times): + self._create_mapping( + self._fault_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_faults(self) -> None: + """Removes every injected fault mapping.""" + for mapping in self._all_mappings(): + metadata = mapping.metadata or {} + if metadata.get('fault'): + self._delete_mapping(mapping.id) diff --git a/test_suites/fixtures/gce_metadata_server.py b/test_suites/fixtures/gce_metadata_server.py index 654977bc75..6984a95805 100644 --- a/test_suites/fixtures/gce_metadata_server.py +++ b/test_suites/fixtures/gce_metadata_server.py @@ -53,10 +53,10 @@ """ import contextlib +from http import HTTPStatus import json import os import tempfile -import uuid import requests from wiremock.client import HttpMethods @@ -67,6 +67,8 @@ from wiremock.constants import Config from wiremock.testing.testcontainer import wiremock_container +from test_suites.fixtures.common import wiremock_faults + _REQUEST_TIMEOUT = 5 _DEFAULT_EXPIRES_IN_SECONDS = 3600 @@ -213,47 +215,26 @@ def _json_value_mapping(path: str, ) -def _fault_mapping(path: str, - status_code: int, - 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 fault for path.""" - response_kwargs = { - 'status': status_code, - 'body': f'injected fault for {path}\n', - 'headers': _TEXT_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={ - 'fault': True, - 'fault_path': path, - }, - request=MappingRequest( - method=HttpMethods.GET, - url_path=f'/computeMetadata/v1/{path}', - headers=_REQUIRED_HEADER_MATCHER, - ), - response=MappingResponse(**response_kwargs), - ) - - class MetadataEmulatorClient: """Talks to a running WireMock metadata emulator.""" - def __init__(self, hostport: str, config: dict): + def __init__( + self, + hostport: str, + config: dict, + fault_injector: wiremock_faults.FaultInjector | None = None, + ): self.hostport = hostport self.admin_url = f'http://{hostport}/__admin' self._baseline_instance_attributes = set() self._runtime_instance_attributes = set() + self._fault_injector = ( + fault_injector or wiremock_faults.WireMockFaultInjector( + admin_url=self.admin_url, + path_prefix='/computeMetadata/v1/', + request_headers=_REQUIRED_HEADER_MATCHER, + response_headers=_TEXT_HEADERS, + )) # Loopback traffic must not be handed to an ambient http_proxy, which # would answer with its own error instead of reaching the emulator. @@ -285,14 +266,6 @@ 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 _reset_mappings(self) -> None: Config.base_url = self.admin_url Mappings.reset_mappings() @@ -437,18 +410,14 @@ def get(self, path: str) -> str: response.raise_for_status() return response.text - def _remove_existing_faults_for_path(self, path: str) -> None: - """Removes any active fault mappings targeting path.""" - for mapping in self._all_mappings(): - metadata = mapping.metadata or {} - if metadata.get('fault') and metadata.get('fault_path') == path: - self._delete_mapping(mapping.id) - - def inject_fault(self, - path: str, - status: int = 500, - times: int = 1, - delay_seconds: float = 0) -> None: + def inject_fault( + self, + path: str, + status: int = HTTPStatus.INTERNAL_SERVER_ERROR, + times: int = 1, + delay_seconds: float = 0, + method: str = HttpMethods.GET, + ) -> None: """Makes the emulator fail a metadata path. Args: @@ -457,33 +426,20 @@ def inject_fault(self, times: How many requests to affect. 0 means indefinitely until clear_faults() is called. delay_seconds: How long the emulator should stall before responding. + method: HTTP method to match (defaults to HttpMethods.GET). """ - status_code = status or 500 - 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_faults_for_path(path) - - if times == 0: - self._create_mapping(_fault_mapping(path, status_code, delay_ms)) - return - - scenario_name = f'fault-{path}-{uuid.uuid4().hex}' - for i in range(times): - self._create_mapping( - _fault_mapping( - path, - status_code, - delay_ms=delay_ms, - scenario_name=scenario_name, - required_state='Started' if i == 0 else f'step_{i}', - new_state=f'step_{i + 1}', - )) + self._fault_injector.inject_fault( + path=path, + status=status, + times=times, + delay_seconds=delay_seconds, + method=method, + ) def clear_faults(self) -> None: """Removes every injected fault and runtime metadata override.""" self._runtime_instance_attributes.clear() + self._fault_injector.clear_faults() self._reset_mappings() From 35ed0cb870afd8c3070f34c7f273cf083a338031 Mon Sep 17 00:00:00 2001 From: Ivan Barba Date: Tue, 6 Oct 2026 13:53:08 -0600 Subject: [PATCH 4/5] [Swarming][Refactor] Swarming constants and changes server URL format (#5517) Bug: b/555379018 ## Overview The following Pr stack focuses in the creation of a swarming API fake to be used by the integration tests suite. For this purpose, we perform a small refactor in the swarming package, focused on making some constants public so that they can be reused in our fake api, as well as some general adjustments in the API initialization so that we can give it http paths. ## Changes - Creates `constants.py` which consists of swarming related constants - Now the swarming server url must be given including `https://` or `https://` this way we give the api the flexiblity of using non secure urls such as ones coming from localhost. - There are following changes in the swarming config in the [config repo](https://clusterfuzz-config-472119376969.us-central1.sourcemanager.dev/clusterfuzz-testing/clusterfuzz-config/pulls/387) ## Tests - Added UT for the new validation in the API --- configs/test/swarming/swarming.yaml | 2 +- .../_internal/swarming/__init__.py | 1 - src/clusterfuzz/_internal/swarming/api.py | 39 ++++++++----------- .../_internal/swarming/constants.py | 33 ++++++++++++++++ .../_internal/tests/core/swarming/api_test.py | 30 +++++++++++++- 5 files changed, 79 insertions(+), 26 deletions(-) create mode 100644 src/clusterfuzz/_internal/swarming/constants.py diff --git a/configs/test/swarming/swarming.yaml b/configs/test/swarming/swarming.yaml index 67284933ff..d94ead9ff4 100644 --- a/configs/test/swarming/swarming.yaml +++ b/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' diff --git a/src/clusterfuzz/_internal/swarming/__init__.py b/src/clusterfuzz/_internal/swarming/__init__.py index fba3fbd2c5..b23a49624d 100644 --- a/src/clusterfuzz/_internal/swarming/__init__.py +++ b/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 diff --git a/src/clusterfuzz/_internal/swarming/api.py b/src/clusterfuzz/_internal/swarming/api.py index 50a9f8091d..c13a05f2e6 100644 --- a/src/clusterfuzz/_internal/swarming/api.py +++ b/src/clusterfuzz/_internal/swarming/api.py @@ -14,11 +14,10 @@ """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 @@ -26,21 +25,9 @@ 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 diff --git a/src/clusterfuzz/_internal/swarming/constants.py b/src/clusterfuzz/_internal/swarming/constants.py new file mode 100644 index 0000000000..9a77ea713a --- /dev/null +++ b/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()) diff --git a/src/clusterfuzz/_internal/tests/core/swarming/api_test.py b/src/clusterfuzz/_internal/tests/core/swarming/api_test.py index 1ef4cf827c..fd61e9ff48 100644 --- a/src/clusterfuzz/_internal/tests/core/swarming/api_test.py +++ b/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) From 3547bc87080ba04d2486b0ee99189f79a74ccaff Mon Sep 17 00:00:00 2001 From: Ivan Barba Date: Tue, 6 Oct 2026 22:03:32 +0000 Subject: [PATCH 5/5] [Better Tests] Rename WireMock fault injection to error injection 'Fault' was an unclear and overly specific term. 'Error' is the more general and widely understood word for what the injector does: make the emulator answer with an HTTP error status (optionally after a delay). - wiremock_faults.py -> wiremock_errors.py - FaultInjector / WireMockFaultInjector -> ErrorInjector / WireMockErrorInjector - inject_fault / clear_faults -> inject_error / clear_errors - fault_injector ctor arg -> error_injector - WireMock mapping metadata keys 'fault' / 'fault_path' -> 'error' / 'error_path' --- ...{wiremock_faults.py => wiremock_errors.py} | 50 +++++++++---------- test_suites/fixtures/gce_metadata_server.py | 20 ++++---- .../fixtures/gce_metadata_server_test.py | 16 +++--- 3 files changed, 43 insertions(+), 43 deletions(-) rename test_suites/fixtures/common/{wiremock_faults.py => wiremock_errors.py} (80%) diff --git a/test_suites/fixtures/common/wiremock_faults.py b/test_suites/fixtures/common/wiremock_errors.py similarity index 80% rename from test_suites/fixtures/common/wiremock_faults.py rename to test_suites/fixtures/common/wiremock_errors.py index db0a5560d8..bdb860c02f 100644 --- a/test_suites/fixtures/common/wiremock_faults.py +++ b/test_suites/fixtures/common/wiremock_errors.py @@ -11,7 +11,7 @@ # 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 fault injection for service emulators.""" +"""Shared WireMock error injection for service emulators.""" import abc from http import HTTPStatus @@ -29,11 +29,11 @@ } -class FaultInjector(abc.ABC): - """Interface for injecting and clearing HTTP faults in service emulators.""" +class ErrorInjector(abc.ABC): + """Interface for injecting and clearing HTTP errors in service emulators.""" @abc.abstractmethod - def inject_fault( + def inject_error( self, path: str, status: int = HTTPStatus.INTERNAL_SERVER_ERROR, @@ -44,12 +44,12 @@ def inject_fault( """Makes the emulator fail requests to path.""" @abc.abstractmethod - def clear_faults(self) -> None: - """Removes every injected fault.""" + def clear_errors(self) -> None: + """Removes every injected error.""" -class WireMockFaultInjector(FaultInjector): - """Injects and clears HTTP faults in a running WireMock container.""" +class WireMockErrorInjector(ErrorInjector): + """Injects and clears HTTP errors in a running WireMock container.""" def __init__( self, @@ -75,7 +75,7 @@ def _delete_mapping(self, mapping_id: str) -> None: Config.base_url = self._admin_url Mappings.delete_mapping(mapping_id) - def _fault_mapping( + def _error_mapping( self, path: str, status_code: int, @@ -85,7 +85,7 @@ def _fault_mapping( required_state: str | None = None, new_state: str | None = None, ) -> Mapping: - """Returns a non-persistent Mapping that injects an HTTP fault for path.""" + """Returns a non-persistent Mapping that injects an HTTP error for path.""" request_kwargs: dict = { 'method': method, 'url_path': f'{self._path_prefix}{path}', @@ -95,7 +95,7 @@ def _fault_mapping( response_kwargs: dict = { 'status': status_code, - 'body': f'injected fault for {path}\n', + 'body': f'injected error for {path}\n', 'headers': self._response_headers, } if delay_ms is not None: @@ -107,21 +107,21 @@ def _fault_mapping( required_scenario_state=required_state, new_scenario_state=new_state, metadata={ - 'fault': True, - 'fault_path': path, + 'error': True, + 'error_path': path, }, request=MappingRequest(**request_kwargs), response=MappingResponse(**response_kwargs), ) - def _remove_existing_faults_for_path(self, path: str) -> None: - """Removes any active fault mappings targeting path.""" + 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('fault') and metadata.get('fault_path') == path: + if metadata.get('error') and metadata.get('error_path') == path: self._delete_mapping(mapping.id) - def inject_fault( + def inject_error( self, path: str, status: int = HTTPStatus.INTERNAL_SERVER_ERROR, @@ -135,7 +135,7 @@ def inject_fault( 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_faults() is called. + clear_errors() is called. delay_seconds: How long the emulator should stall before responding. method: HTTP method to match (defaults to HttpMethods.ANY). """ @@ -144,18 +144,18 @@ def inject_fault( if times < 0: raise ValueError(f'times must be >= 0, got {times}') - self._remove_existing_faults_for_path(path) + self._remove_existing_errors_for_path(path) if times == 0: self._create_mapping( - self._fault_mapping( + self._error_mapping( path, status_code, method=method, delay_ms=delay_ms)) return - scenario_name = f'fault-{path}-{uuid.uuid4().hex}' + scenario_name = f'error-{path}-{uuid.uuid4().hex}' for i in range(times): self._create_mapping( - self._fault_mapping( + self._error_mapping( path, status_code, method=method, @@ -165,9 +165,9 @@ def inject_fault( new_state=f'step_{i + 1}', )) - def clear_faults(self) -> None: - """Removes every injected fault mapping.""" + def clear_errors(self) -> None: + """Removes every injected error mapping.""" for mapping in self._all_mappings(): metadata = mapping.metadata or {} - if metadata.get('fault'): + if metadata.get('error'): self._delete_mapping(mapping.id) diff --git a/test_suites/fixtures/gce_metadata_server.py b/test_suites/fixtures/gce_metadata_server.py index 6984a95805..eed8ff2c58 100644 --- a/test_suites/fixtures/gce_metadata_server.py +++ b/test_suites/fixtures/gce_metadata_server.py @@ -67,7 +67,7 @@ from wiremock.constants import Config from wiremock.testing.testcontainer import wiremock_container -from test_suites.fixtures.common import wiremock_faults +from test_suites.fixtures.common import wiremock_errors _REQUEST_TIMEOUT = 5 _DEFAULT_EXPIRES_IN_SECONDS = 3600 @@ -222,14 +222,14 @@ def __init__( self, hostport: str, config: dict, - fault_injector: wiremock_faults.FaultInjector | None = None, + error_injector: wiremock_errors.ErrorInjector | None = None, ): self.hostport = hostport self.admin_url = f'http://{hostport}/__admin' self._baseline_instance_attributes = set() self._runtime_instance_attributes = set() - self._fault_injector = ( - fault_injector or wiremock_faults.WireMockFaultInjector( + self._error_injector = ( + error_injector or wiremock_errors.WireMockErrorInjector( admin_url=self.admin_url, path_prefix='/computeMetadata/v1/', request_headers=_REQUIRED_HEADER_MATCHER, @@ -410,7 +410,7 @@ def get(self, path: str) -> str: response.raise_for_status() return response.text - def inject_fault( + def inject_error( self, path: str, status: int = HTTPStatus.INTERNAL_SERVER_ERROR, @@ -424,11 +424,11 @@ def inject_fault( path: A metadata path without the /computeMetadata/v1/ prefix. status: The HTTP status to return. times: How many requests to affect. 0 means indefinitely until - clear_faults() is called. + clear_errors() is called. delay_seconds: How long the emulator should stall before responding. method: HTTP method to match (defaults to HttpMethods.GET). """ - self._fault_injector.inject_fault( + self._error_injector.inject_error( path=path, status=status, times=times, @@ -436,10 +436,10 @@ def inject_fault( method=method, ) - def clear_faults(self) -> None: - """Removes every injected fault and runtime metadata override.""" + def clear_errors(self) -> None: + """Removes every injected error and runtime metadata override.""" self._runtime_instance_attributes.clear() - self._fault_injector.clear_faults() + self._error_injector.clear_errors() self._reset_mappings() diff --git a/test_suites/fixtures/gce_metadata_server_test.py b/test_suites/fixtures/gce_metadata_server_test.py index 416f96d3b7..ff39bc4706 100644 --- a/test_suites/fixtures/gce_metadata_server_test.py +++ b/test_suites/fixtures/gce_metadata_server_test.py @@ -52,8 +52,8 @@ def _emulators(cls): @pytest.fixture(autouse=True) def clean_up_after_each_test(self): yield - self.tworker.clear_faults() - self.uworker.clear_faults() + self.tworker.clear_errors() + self.uworker.clear_errors() def test_default_credentials_come_from_the_trusted_emulator(self): """Verifies that google.auth.default() resolves Compute Engine credentials @@ -176,9 +176,9 @@ def test_a_second_emulator_cannot_take_a_held_address(self): with gce_metadata_server.untrusted_as_default(): pass - def test_a_fault_stops_after_the_configured_request_count(self): - """Verifies that a fault budgeted for one request breaks only that one.""" - self.tworker.inject_fault('instance/zone', status=500, times=1) + def test_an_error_stops_after_the_configured_request_count(self): + """Verifies that an error budgeted for one request breaks only that one.""" + self.tworker.inject_error('instance/zone', status=500, times=1) with pytest.raises(requests.exceptions.HTTPError) as caught: self.tworker.get('instance/zone') @@ -187,15 +187,15 @@ def test_a_fault_stops_after_the_configured_request_count(self): assert self.tworker.get('instance/zone') == ( 'projects/1234567890/zones/us-central1-f') - def test_compute_metadata_get_retries_itself_past_transient_faults(self): + def test_compute_metadata_get_retries_itself_past_transient_errors(self): """Verifies that get() retries itself, and only fails past its budget.""" budget = compute_metadata._RETRIES # pylint: disable=protected-access - self.tworker.inject_fault('instance/zone', status=500, times=budget) + self.tworker.inject_error('instance/zone', status=500, times=budget) assert compute_metadata.get('instance/zone') == ( 'projects/1234567890/zones/us-central1-f') - self.tworker.inject_fault('instance/zone', status=500, times=budget + 1) + self.tworker.inject_error('instance/zone', status=500, times=budget + 1) with pytest.raises(requests.exceptions.HTTPError) as caught: compute_metadata.get('instance/zone') assert caught.value.response.status_code == 500