diff --git a/configs/test/swarming/swarming.yaml b/configs/test/swarming/swarming.yaml index 67284933ff3..d94ead9ff41 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 fba3fbd2c51..b23a49624db 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 50a9f8091d5..c13a05f2e6d 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 00000000000..9a77ea713a0 --- /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 1ef4cf827c8..fd61e9ff484 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) diff --git a/test_suites/fixtures/common/__init__.py b/test_suites/fixtures/common/__init__.py new file mode 100644 index 00000000000..f8cb6d9d201 --- /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_errors.py b/test_suites/fixtures/common/wiremock_errors.py new file mode 100644 index 00000000000..bdb860c02f5 --- /dev/null +++ b/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) diff --git a/test_suites/fixtures/gce_metadata_server.py b/test_suites/fixtures/gce_metadata_server.py index 3dd398d535b..eed8ff2c581 100644 --- a/test_suites/fixtures/gce_metadata_server.py +++ b/test_suites/fixtures/gce_metadata_server.py @@ -53,6 +53,7 @@ """ import contextlib +from http import HTTPStatus import json import os import tempfile @@ -66,6 +67,8 @@ from wiremock.constants import Config from wiremock.testing.testcontainer import wiremock_container +from test_suites.fixtures.common import wiremock_errors + _REQUEST_TIMEOUT = 5 _DEFAULT_EXPIRES_IN_SECONDS = 3600 @@ -215,10 +218,23 @@ def _json_value_mapping(path: str, class MetadataEmulatorClient: """Talks to a running WireMock metadata emulator.""" - def __init__(self, hostport: str, config: dict): + def __init__( + self, + hostport: str, + config: dict, + 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._error_injector = ( + error_injector or wiremock_errors.WireMockErrorInjector( + 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. @@ -250,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() @@ -313,7 +321,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 +344,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 +410,38 @@ def get(self, path: str) -> str: response.raise_for_status() return response.text + def inject_error( + 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: + 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_errors() is called. + delay_seconds: How long the emulator should stall before responding. + method: HTTP method to match (defaults to HttpMethods.GET). + """ + self._error_injector.inject_error( + path=path, + status=status, + times=times, + delay_seconds=delay_seconds, + method=method, + ) + + def clear_errors(self) -> None: + """Removes every injected error and runtime metadata override.""" + self._runtime_instance_attributes.clear() + self._error_injector.clear_errors() + 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 8f320291370..ff39bc47068 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_errors() + self.uworker.clear_errors() + 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,48 @@ def test_a_second_emulator_cannot_take_a_held_address(self): with gce_metadata_server.untrusted_as_default(): pass + 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') + 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_errors(self): + """Verifies that get() retries itself, and only fails past its budget.""" + budget = compute_metadata._RETRIES # pylint: disable=protected-access + + 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_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 + + def test_set_adds_and_overrides_metadata_across_scopes(self): + """Verifies runtime metadata overrides in project and instance scopes.""" + 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( + 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(key='preempted', value='TRUE') + assert compute_metadata.get('instance/preempted') == 'TRUE' + class TestGceMetadataEmulatorUntrustedDefault: """Integration tests for a test class that runs as the untrusted worker."""