From a6b8fee0a33d820917af4b6c612abd6194a3f1a4 Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:20:44 -0400 Subject: [PATCH] TEST: Make macOS unit-test gate resilient to runner stalls The macOS build_and_test leg failed 80 times in two days; 71 were single tests whose short watchdogs fired while the 3-vCPU runner stalled for 20-68s, not genuine hangs. - Add TEST_WATCHDOG_SECONDS (300s) in tests/unit/async_utils.py as the shared deadlock backstop and use it for positive test watchdogs, analytics query budgets, and the pytest-timeout markers that capped those waits. - Make the discover-capabilities timeout variant deterministic: expire the probe deadline through the real wait_for cancellation path once the request is pending, instead of a 5s wall-clock deadline that a stalled runner could exhaust before the request started. - Reuse the shared analytics control in the eval-identity tests. - Run 3 xdist workers on macOS via a PYTEST_WORKERS Makefile override. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/workflows/build_and_test.yml | 3 + Makefile | 14 ++- doc/contributing/4_running_tests.md | 7 +- tests/unit/analytics/conftest.py | 8 +- tests/unit/async_utils.py | 6 +- tests/unit/backend/conftest.py | 8 +- .../unit/backend/test_analytics_lifecycle.py | 17 +-- tests/unit/backend/test_analytics_routes.py | 16 ++- .../unit/backend/test_conversation_editor.py | 3 +- .../unit/backend/test_message_send_service.py | 22 ++-- tests/unit/backend/test_scenario_resume.py | 28 ++--- .../test_seeded_converter_determinism.py | 3 +- .../test_scoring_expectation_transport.py | 9 +- .../test_interface_observations.py | 11 +- tests/unit/memory/test_attack_analytics.py | 9 +- .../test_attack_analytics_eval_identity.py | 7 +- tests/unit/memory/test_sqlite_cancellation.py | 24 ++-- .../target/test_github_copilot_target.py | 104 +++++++++--------- .../test_discover_target_capabilities.py | 41 +++++-- tests/unit/scenario/core/test_scenario.py | 5 +- .../core/test_scenario_partial_results.py | 18 +-- 21 files changed, 206 insertions(+), 157 deletions(-) diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 3a53942384..b49a5f4994 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -196,6 +196,9 @@ jobs: fi - name: Run unit tests + # The macOS runner has 3 vCPUs; a fourth worker oversubscribes it and stretches timing-sensitive tests. + env: + PYTEST_WORKERS: ${{ runner.os == 'macOS' && '3' || '4' }} run: make unit-test-junit - name: Upload JUnit results diff --git a/Makefile b/Makefile index 03d9b54f2e..e365869176 100644 --- a/Makefile +++ b/Makefile @@ -8,6 +8,7 @@ INTEGRATION_TESTS:=tests/integration PARTNER_INTEGRATION_TESTS:=tests/partner_integration END_TO_END_TESTS:=tests/end_to_end JUNIT_XML?=junit/test-results.xml +PYTEST_WORKERS?=4 DIFF_COVER_BASE?=origin/main all: pre-commit @@ -38,21 +39,22 @@ docs-api: uv run python -m build_scripts.pydoc2json pyrit --submodules -o doc/_api/pyrit_all.json uv run python -m build_scripts.gen_api_md -# Because of import time, "auto" seemed to actually go slower than just using 4 processes +# Because of import time, "auto" seemed to actually go slower than just using 4 processes. +# Override PYTEST_WORKERS on hosts with fewer cores (CI uses 3 on the 3-vCPU macOS runner). unit-test: - $(CMD) pytest -n 4 --dist=loadfile $(UNIT_TESTS) + $(CMD) pytest -n $(PYTEST_WORKERS) --dist=loadfile $(UNIT_TESTS) unit-test-junit: - $(CMD) pytest -n 4 --dist=loadfile $(UNIT_TESTS) --junitxml=$(JUNIT_XML) --durations=25 + $(CMD) pytest -n $(PYTEST_WORKERS) --dist=loadfile $(UNIT_TESTS) --junitxml=$(JUNIT_XML) --durations=25 unit-test-cov-html: - $(CMD) pytest -n 4 --dist=loadfile --cov=$(PYMODULE) --cov-fail-under=78 $(UNIT_TESTS) --cov-report html + $(CMD) pytest -n $(PYTEST_WORKERS) --dist=loadfile --cov=$(PYMODULE) --cov-fail-under=78 $(UNIT_TESTS) --cov-report html unit-test-cov-xml: - $(CMD) pytest -n 4 --dist=loadfile --cov=$(PYMODULE) --cov-fail-under=78 $(UNIT_TESTS) --cov-report xml --cov-report term --junitxml=$(JUNIT_XML) --durations=25 + $(CMD) pytest -n $(PYTEST_WORKERS) --dist=loadfile --cov=$(PYMODULE) --cov-fail-under=78 $(UNIT_TESTS) --cov-report xml --cov-report term --junitxml=$(JUNIT_XML) --durations=25 diff-cover: - $(CMD) pytest -n 4 --dist=loadfile --cov=$(PYMODULE) --cov-fail-under=78 $(UNIT_TESTS) --cov-report xml + $(CMD) pytest -n $(PYTEST_WORKERS) --dist=loadfile --cov=$(PYMODULE) --cov-fail-under=78 $(UNIT_TESTS) --cov-report xml uv run python -m diff_cover.diff_cover_tool coverage.xml --compare-branch="$(DIFF_COVER_BASE)" --diff-range-notation=.. --fail-under=90 unit-test-diff-cover: diff --git a/doc/contributing/4_running_tests.md b/doc/contributing/4_running_tests.md index a056340145..dc586dfd0e 100644 --- a/doc/contributing/4_running_tests.md +++ b/doc/contributing/4_running_tests.md @@ -71,9 +71,10 @@ The helper's scenarios can also run directly with For a narrower run, invoke `pytest` directly. You can invoke pytest if it's in your path or via python; either `pytest` or `python -m pytest`. For the following examples, we will use `pytest`. -Analytics correctness tests use a shared, finite 120-second query budget to allow for -SQL compilation and scheduling delays on busy CI workers. Deadline and cancellation -tests use explicit controls and keep their separate timing assertions. +Analytics correctness tests and unit-test watchdogs share the finite +`TEST_WATCHDOG_SECONDS` (300-second) backstop in `tests/unit/async_utils.py`. It only +catches deadlocks; shorter waits fail healthy tests on busy CI workers. Deadline and +cancellation tests use explicit controls and keep their separate timing assertions. * To run every test under a subdirectory: diff --git a/tests/unit/analytics/conftest.py b/tests/unit/analytics/conftest.py index 97fd7b531f..8eacb157ad 100644 --- a/tests/unit/analytics/conftest.py +++ b/tests/unit/analytics/conftest.py @@ -12,6 +12,7 @@ from pyrit.analytics import AttackResultAnalytics from pyrit.analytics._execution import AnalyticsExecution from pyrit.memory import SQLiteMemory +from unit.async_utils import TEST_WATCHDOG_SECONDS @pytest.fixture @@ -23,7 +24,12 @@ async def analytics(sqlite_instance: SQLiteMemory) -> AsyncGenerator[AttackResul Default budgets and expiration/cleanup are tested independently in test_execution.py. Admission capacities and native database work are unchanged. """ - execution = partial(AnalyticsExecution, queue_timeout=30.0, report_timeout=30.0, quick_timeout=30.0) + execution = partial( + AnalyticsExecution, + queue_timeout=TEST_WATCHDOG_SECONDS, + report_timeout=TEST_WATCHDOG_SECONDS, + quick_timeout=TEST_WATCHDOG_SECONDS, + ) with patch("pyrit.analytics.attack_result_analytics.AnalyticsExecution", side_effect=execution): async with AttackResultAnalytics(memory=sqlite_instance) as analytics: yield analytics diff --git a/tests/unit/async_utils.py b/tests/unit/async_utils.py index 77f47db423..802ad59885 100644 --- a/tests/unit/async_utils.py +++ b/tests/unit/async_utils.py @@ -6,13 +6,17 @@ T = TypeVar("T") +# Deadlock backstop for correctness tests, not a latency budget. Loaded CI runners (notably the +# 3-vCPU macOS images) stall single tests for over a minute, so shorter waits fail healthy tests. +TEST_WATCHDOG_SECONDS = 300.0 + def get_defined_tasks(*tasks: asyncio.Task[T] | None) -> list[asyncio.Task[T]]: """Retain owned tasks even when a test fails before creating all of them.""" return [task for task in tasks if task is not None] -async def wait_for_completion_async(*, future: asyncio.Future[T], timeout: float = 300) -> T: +async def wait_for_completion_async(*, future: asyncio.Future[T], timeout: float = TEST_WATCHDOG_SECONDS) -> T: """Bound a test wait without injecting cancellation into the operation under test. The default is a deadlock backstop rather than a latency assertion, so it stays finite to keep a diff --git a/tests/unit/backend/conftest.py b/tests/unit/backend/conftest.py index d37c53770a..6d0ce4bde3 100644 --- a/tests/unit/backend/conftest.py +++ b/tests/unit/backend/conftest.py @@ -19,12 +19,18 @@ from pyrit.backend.services.message_send_service import get_message_send_service from pyrit.memory import SQLiteMemory from test_support.fixtures import authentication_policy, compatibility_headers, compatibility_id # noqa: F401 +from unit.async_utils import TEST_WATCHDOG_SECONDS @pytest.fixture async def analytics_service(sqlite_instance: SQLiteMemory) -> AsyncGenerator[AttackResultAnalytics, None]: """Use real isolated memory with finite correctness budgets, not production latency assertions.""" - execution = partial(AnalyticsExecution, queue_timeout=30.0, report_timeout=30.0, quick_timeout=30.0) + execution = partial( + AnalyticsExecution, + queue_timeout=TEST_WATCHDOG_SECONDS, + report_timeout=TEST_WATCHDOG_SECONDS, + quick_timeout=TEST_WATCHDOG_SECONDS, + ) assert get_analytics_service.cache_info().currsize == 0 with patch("pyrit.analytics.attack_result_analytics.AnalyticsExecution", side_effect=execution): try: diff --git a/tests/unit/backend/test_analytics_lifecycle.py b/tests/unit/backend/test_analytics_lifecycle.py index 20cd31beee..27ae200cd5 100644 --- a/tests/unit/backend/test_analytics_lifecycle.py +++ b/tests/unit/backend/test_analytics_lifecycle.py @@ -34,6 +34,7 @@ from pyrit.memory import CentralMemory, SQLiteMemory from pyrit.memory.memory_models import Base from pyrit.setup.configuration_loader import ConfigurationLoader +from unit.async_utils import TEST_WATCHDOG_SECONDS @pytest.fixture @@ -163,7 +164,7 @@ async def test_http_timeout_keeps_native_cleanup_busy_until_runtime_drain_async( cleanup = held_native_cleanup async def expire_async(tasks: set[asyncio.Task[Any]], *, timeout: float) -> tuple[set[Any], set[Any]]: - assert 0 < timeout <= 30 + assert 0 < timeout <= TEST_WATCHDOG_SECONDS await expire.wait() return set(), tasks @@ -186,10 +187,10 @@ async def close_async() -> None: request = asyncio.create_task(client.post("/api/analytics/attacks/results", json={})) shutdown: asyncio.Task[None] | None = None try: - await asyncio.wait_for(cleanup.entered.wait(), 30) + await asyncio.wait_for(cleanup.entered.wait(), TEST_WATCHDOG_SECONDS) assert has_active_analytics_work() expire.set() - response = await asyncio.wait_for(request, 10) + response = await asyncio.wait_for(request, TEST_WATCHDOG_SECONDS) assert response.status_code == 504 assert not analytics_runtime.operations assert analytics_service.has_active_work @@ -197,7 +198,7 @@ async def close_async() -> None: assert analytics_runtime.outcome == "busy" assert analytics_runtime.generation == "original" shutdown = asyncio.create_task(analytics_runtime.shutdown_async()) - await asyncio.wait_for(closing.wait(), 10) + await asyncio.wait_for(closing.wait(), TEST_WATCHDOG_SECONDS) assert not shutdown.done() assert not analytics_service.is_closed assert get_analytics_service() is analytics_service @@ -212,9 +213,9 @@ async def close_async() -> None: cleanup.release.set() if cancel_shutdown: with pytest.raises(asyncio.CancelledError): - await asyncio.wait_for(shutdown, 30) + await asyncio.wait_for(shutdown, TEST_WATCHDOG_SECONDS) else: - await asyncio.wait_for(shutdown, 30) + await asyncio.wait_for(shutdown, TEST_WATCHDOG_SECONDS) assert cleanup.closed.is_set() assert analytics_service.is_closed assert not has_active_analytics_work() @@ -233,7 +234,7 @@ async def test_caller_and_repeated_close_cancellation_retain_native_owner_async( request = asyncio.create_task(analytics_service.results_async()) close: asyncio.Task[None] | None = None try: - await asyncio.wait_for(cleanup.entered.wait(), 30) + await asyncio.wait_for(cleanup.entered.wait(), TEST_WATCHDOG_SECONDS) request.cancel() with pytest.raises(asyncio.CancelledError): await request @@ -247,7 +248,7 @@ async def test_caller_and_repeated_close_cancellation_retain_native_owner_async( assert get_analytics_service() is analytics_service cleanup.release.set() with pytest.raises(asyncio.CancelledError): - await asyncio.wait_for(close, 30) + await asyncio.wait_for(close, TEST_WATCHDOG_SECONDS) assert cleanup.closed.is_set() assert analytics_service.is_closed assert get_analytics_service.cache_info().currsize == 0 diff --git a/tests/unit/backend/test_analytics_routes.py b/tests/unit/backend/test_analytics_routes.py index 02e1a766c5..6309ad18ee 100644 --- a/tests/unit/backend/test_analytics_routes.py +++ b/tests/unit/backend/test_analytics_routes.py @@ -43,6 +43,7 @@ AttackAnalyticsResultsQuery, AttackOutcome, ) +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.memory.test_attack_analytics import make_result @@ -123,7 +124,12 @@ def create() -> AttackResultAnalytics: assert get_analytics_service.cache_info().currsize == 0 user = AuthenticatedUser(oid="test", name="Test", email="test@example.invalid", groups=["test-group"]) - budgets = partial(AnalyticsExecution, report_timeout=30.0, quick_timeout=30.0, queue_timeout=30.0) + budgets = partial( + AnalyticsExecution, + report_timeout=TEST_WATCHDOG_SECONDS, + quick_timeout=TEST_WATCHDOG_SECONDS, + queue_timeout=TEST_WATCHDOG_SECONDS, + ) with ( patch("pyrit.backend.services.analytics_service.AttackResultAnalytics", side_effect=create), patch("pyrit.analytics.attack_result_analytics.AnalyticsExecution", side_effect=budgets), @@ -412,13 +418,13 @@ async def blocked_async(**kwargs: object) -> Any: asyncio.create_task(analytics_client.post("/api/analytics/attacks/results", json={})) for _ in range(12) ] try: - await asyncio.wait_for(admitted.wait(), 10) + await asyncio.wait_for(admitted.wait(), TEST_WATCHDOG_SECONDS) rejected = await analytics_client.post("/api/analytics/attacks/results", json={}) assert rejected.status_code == 503 assert rejected.headers["Retry-After"] == "1" assert analytics_service.has_active_work release.set() - responses = await asyncio.wait_for(asyncio.gather(*requests), 30) + responses = await asyncio.wait_for(asyncio.gather(*requests), TEST_WATCHDOG_SECONDS) assert all(response.status_code == 200 for response in responses) assert read.await_count == 12 finally: @@ -466,10 +472,10 @@ async def expire_async(tasks: set[Any], *, timeout: float) -> tuple[set[Any], se clock.monotonic.return_value = 100.0 request = asyncio.create_task(analytics_client.post(f"/api/analytics/attacks/{endpoint}", json=body)) try: - await asyncio.wait_for(entered.wait(), 10) + await asyncio.wait_for(entered.wait(), TEST_WATCHDOG_SECONDS) assert observed == [deadline] expire.set() - response = await asyncio.wait_for(request, 10) + response = await asyncio.wait_for(request, TEST_WATCHDOG_SECONDS) assert response.status_code == response.json()["status"] == 504 assert "Retry-After" not in response.headers assert analytics_service.has_active_work diff --git a/tests/unit/backend/test_conversation_editor.py b/tests/unit/backend/test_conversation_editor.py index 14db53dacc..86a92dc990 100644 --- a/tests/unit/backend/test_conversation_editor.py +++ b/tests/unit/backend/test_conversation_editor.py @@ -46,6 +46,7 @@ from pyrit.models.target.target_capabilities import TargetCapabilities from pyrit.prompt_target import OpenAIResponseTarget from pyrit.score.scorer_prompt_validator import ScorerPromptValidator +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.backend.mocks import _settle_send_async, message_send_lifecycle_async from unit.mocks import MockPromptTarget, get_mock_target_identifier, run_memory_session_async @@ -1305,7 +1306,7 @@ async def interrupted_write_async(*args: Any, **kwargs: Any) -> None: with patch.object(storage, "write_file_async", side_effect=interrupted_write_async): task = asyncio.create_task(AttackService().save_conversation_async(request=request)) - await asyncio.wait_for(started.wait(), timeout=5) + await asyncio.wait_for(started.wait(), timeout=TEST_WATCHDOG_SECONDS) if cancel: task.cancel() finish.set() diff --git a/tests/unit/backend/test_message_send_service.py b/tests/unit/backend/test_message_send_service.py index 53e58b799e..127b7d0ef4 100644 --- a/tests/unit/backend/test_message_send_service.py +++ b/tests/unit/backend/test_message_send_service.py @@ -60,7 +60,7 @@ PromptDataType, ) from pyrit.prompt_normalizer import ConverterConfiguration, PromptNormalizer -from unit.async_utils import wait_for_completion_async +from unit.async_utils import TEST_WATCHDOG_SECONDS, wait_for_completion_async from unit.backend.mocks import ( _make_matching_target_mock, _settle_send_async, @@ -1614,7 +1614,7 @@ async def test_converter_merge_preserves_sibling_children_hash(self, message_sen @pytest.mark.usefixtures("patch_central_database") -@pytest.mark.timeout(10) +@pytest.mark.timeout(TEST_WATCHDOG_SECONDS) class TestConcurrentMessages: def test_default_service_instances_share_scheduler(self, mock_memory: MagicMock) -> None: first = MessageSendService() @@ -1766,7 +1766,7 @@ async def hold_async(*, conversation_id: str, **_: Any) -> None: await first_started.wait() await asyncio.wait_for( service.add_message_async(attack_result_id="attack", request=_request(conversation_id="second")), - timeout=3, + timeout=TEST_WATCHDOG_SECONDS, ) assert order == ["main", "second"] assert not active.done() @@ -1774,7 +1774,6 @@ async def hold_async(*, conversation_id: str, **_: Any) -> None: release_first.set() await active - @pytest.mark.timeout(90) async def test_target_pacing_does_not_block_another_target_async( self, *, @@ -1824,12 +1823,12 @@ async def update_attack_result_async(*, attack_result_id: str, update_fields: di ) tasks = [active] try: - await asyncio.wait_for(waiting.wait(), timeout=30) + await asyncio.wait_for(waiting.wait(), timeout=TEST_WATCHDOG_SECONDS) independent = asyncio.create_task( service.add_message_async(attack_result_id=other_attack.attack_result_id, request=other_request) ) tasks.append(independent) - await asyncio.wait_for(metadata_started.wait(), timeout=30) + await asyncio.wait_for(metadata_started.wait(), timeout=TEST_WATCHDOG_SECONDS) assert delays == [2.0, 1.0] assert target.prompt_sent == [] assert other_target.prompt_sent == ["Hello"] @@ -1921,7 +1920,8 @@ async def observe_guard_async(converter: Converter) -> AsyncGenerator[None, None ) await shared_waiting.wait() await asyncio.wait_for( - peer.add_message_async(attack_result_id=ar.attack_result_id, request=requests[2]), timeout=3 + peer.add_message_async(attack_result_id=ar.attack_result_id, request=requests[2]), + timeout=TEST_WATCHDOG_SECONDS, ) assert calls == 1 assert not any(task.done() for task in tasks) @@ -1969,7 +1969,7 @@ async def test_cancellation_joins_memory_write_before_releasing_ownership_async( async def hold_write_async(**_: Any) -> None: started.set() - await asyncio.wait_for(release.wait(), timeout=5) + await asyncio.wait_for(release.wait(), timeout=TEST_WATCHDOG_SECONDS) finished.set() with patch.object(mock_memory, write_method, side_effect=hold_write_async): @@ -2148,7 +2148,7 @@ async def delay_first_write_async(*, attack_result_id: str, update_fields: dict[ writes.append(update_fields) if len(writes) == 1: first_write_started.set() - await asyncio.wait_for(release_first_write.wait(), timeout=5) + await asyncio.wait_for(release_first_write.wait(), timeout=TEST_WATCHDOG_SECONDS) await update(attack_result_id=attack_result_id, update_fields=update_fields) with ( @@ -3360,7 +3360,7 @@ async def hold_write_async(**_: Any) -> None: ).state == MessageSendState.INTERRUPTED -@pytest.mark.timeout(20) +@pytest.mark.timeout(TEST_WATCHDOG_SECONDS) @pytest.mark.usefixtures("patch_central_database") class TestRepeatedMessageSend: async def test_missing_validated_target_fails_before_branch_registration_async( @@ -3720,7 +3720,7 @@ async def finalize_async(**kwargs: Any) -> None: request=_submission(conversation_id=attack.conversation_id, submission_id="too-early"), ) release_metadata.set() - async with asyncio.timeout(5): + async with asyncio.timeout(TEST_WATCHDOG_SECONDS): while during.conversations[0].state != MessageSendState.COMPLETED: during = await service.get_status_async( attack_result_id=attack.attack_result_id, send_id=status.send_id, wait_ms=1 diff --git a/tests/unit/backend/test_scenario_resume.py b/tests/unit/backend/test_scenario_resume.py index 37c6549e45..04244b78cb 100644 --- a/tests/unit/backend/test_scenario_resume.py +++ b/tests/unit/backend/test_scenario_resume.py @@ -56,6 +56,7 @@ from pyrit.scenario.scenarios.airt.rapid_response import RapidResponse, _build_rapid_response_technique from pyrit.scenario.scenarios.foundry.red_team_agent import RedTeamAgent from pyrit.score import SubStringScorer +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.mocks import MockPromptTarget _SCENARIO_NAME = "offline.resume" @@ -140,12 +141,12 @@ async def resume_environment( await service.shutdown_async() -async def _wait_for_idle_async(service: ScenarioRunService, *, timeout: float = 10) -> None: +async def _wait_for_idle_async(service: ScenarioRunService) -> None: async def wait_async() -> None: while service.get_queue_snapshot().active is not None: await asyncio.sleep(0.01) - await asyncio.wait_for(wait_async(), timeout=timeout) + await asyncio.wait_for(wait_async(), timeout=TEST_WATCHDOG_SECONDS) async def test_real_matrix_launch_keeps_heartbeat_alive_during_disk_reads_async( @@ -179,7 +180,7 @@ def load_examples() -> list[dict[str, str]]: assert threading.get_ident() != backend_thread assert get_default_adversarial_target() is target reads += 1 - if not release.wait(5): + if not release.wait(TEST_WATCHDOG_SECONDS): raise TimeoutError("Many-shot disk read was not released.") return original_load() @@ -227,11 +228,11 @@ async def read_seeds_async(**kwargs: Any) -> Sequence[Seed[Any]]: ) ) try: - await asyncio.wait_for(entered.wait(), 5) - await asyncio.wait_for(heartbeat.wait(), 1) + await asyncio.wait_for(entered.wait(), TEST_WATCHDOG_SECONDS) + await asyncio.wait_for(heartbeat.wait(), TEST_WATCHDOG_SECONDS) assert not launch.done() release.set() - response = await asyncio.wait_for(launch, 10) + response = await asyncio.wait_for(launch, TEST_WATCHDOG_SECONDS) await _wait_for_idle_async(service) stored = await sqlite_instance.get_scenario_result_header_async( scenario_result_id=response.scenario_result_id @@ -282,7 +283,7 @@ def load_template(file: str | pathlib.Path) -> dict[str, Any]: assert threading.get_ident() != backend_thread assert get_default_adversarial_target() is target reads += 1 - if not release.wait(5): + if not release.wait(TEST_WATCHDOG_SECONDS): raise TimeoutError("Foundry disk read was not released.") return original_load(file) @@ -343,11 +344,11 @@ async def send_async(*, normalized_conversation: list[Message]) -> list[Message] ) ) try: - await asyncio.wait_for(entered.wait(), 5) - await asyncio.wait_for(heartbeat.wait(), 1) + await asyncio.wait_for(entered.wait(), TEST_WATCHDOG_SECONDS) + await asyncio.wait_for(heartbeat.wait(), TEST_WATCHDOG_SECONDS) assert not launch.done() release.set() - response = await asyncio.wait_for(launch, 10) + response = await asyncio.wait_for(launch, TEST_WATCHDOG_SECONDS) await _wait_for_idle_async(service) stored = await sqlite_instance.get_scenario_result_header_async( scenario_result_id=response.scenario_result_id @@ -537,8 +538,7 @@ async def run_async(self: _OfflineResumeScenario) -> None: include_baseline=False, ) ) - # This checks real SQLite completion, not latency under parallel CI disk load. - await _wait_for_idle_async(service, timeout=60) + await _wait_for_idle_async(service) dispose.assert_not_awaited() assert len(engines) == 2 and engines[0] is engines[1] assert sqlite_instance._get_async_engine() is engines[0] @@ -789,7 +789,7 @@ async def fail_launch_async(*, request: RunScenarioRequest) -> None: with patch.object(service, "_start_run_locked_async", side_effect=fail_launch_async) as start: first = asyncio.create_task(service.resume_run_async(scenario_result_id=str(stored.id))) - await asyncio.wait_for(entered.wait(), timeout=5) + await asyncio.wait_for(entered.wait(), timeout=TEST_WATCHDOG_SECONDS) with pytest.raises(ScenarioRunConflictError, match="already being resumed"): await service.resume_run_async(scenario_result_id=str(stored.id)) with pytest.raises(ScenarioRunConflictError, match="already being resumed"): @@ -866,7 +866,7 @@ async def hold_run_async(*, scenario_result_id: str) -> None: active = await service.start_run_async( request=RunScenarioRequest(scenario_name=_SCENARIO_NAME, target_name=_TARGET_NAME, max_concurrency=1) ) - await asyncio.wait_for(entered.wait(), timeout=5) + await asyncio.wait_for(entered.wait(), timeout=TEST_WATCHDOG_SECONDS) resumed = await service.resume_run_async(scenario_result_id=run_id) assert resumed.status == ScenarioRunState.QUEUED assert resumed.queue_position == 1 diff --git a/tests/unit/converter/test_seeded_converter_determinism.py b/tests/unit/converter/test_seeded_converter_determinism.py index 7606ca64e3..2f807359d5 100644 --- a/tests/unit/converter/test_seeded_converter_determinism.py +++ b/tests/unit/converter/test_seeded_converter_determinism.py @@ -7,6 +7,7 @@ from collections.abc import Callable import pytest +from unit.async_utils import TEST_WATCHDOG_SECONDS from pyrit.common.random_context import configure_random_seed from pyrit.converter import ( @@ -133,7 +134,7 @@ async def convert_async(self, *, prompt: str, input_type: PromptDataType = "text barrier = threading.Barrier(2) def draw() -> tuple[random.Random, float]: - barrier.wait(timeout=5) + barrier.wait(timeout=TEST_WATCHDOG_SECONDS) rng = self._get_random_generator(stream="values") return (rng, rng.random()) diff --git a/tests/unit/executor/attack/test_scoring_expectation_transport.py b/tests/unit/executor/attack/test_scoring_expectation_transport.py index d968dd6677..7fe2b90f35 100644 --- a/tests/unit/executor/attack/test_scoring_expectation_transport.py +++ b/tests/unit/executor/attack/test_scoring_expectation_transport.py @@ -12,7 +12,7 @@ import pytest from aiosqlite import Cursor from sqlalchemy import event -from unit.async_utils import wait_for_completion_async +from unit.async_utils import TEST_WATCHDOG_SECONDS, wait_for_completion_async from unit.mocks import MockPromptTarget, store_message_async from pyrit.exceptions import ( @@ -291,7 +291,7 @@ async def test_concurrent_rows_keep_effective_expectations_isolated_async(self, else: execution = executor.execute_attack_async(objectives=objectives, **kwargs) - batch = await asyncio.wait_for(execution, timeout=10) + batch = await asyncio.wait_for(execution, timeout=TEST_WATCHDOG_SECONDS) expected = [ supplied.model_copy(update={"objective": objectives[0]}), @@ -638,10 +638,9 @@ async def delayed_fetchall_async(cursor: Cursor) -> Iterable[Any]: attack.execute_async(objective="attack objective", expectation=_expectation()) ) try: - # Time cancellation cleanup separately from attack startup and target persistence. - await asyncio.wait_for(validation_started.wait(), timeout=30) + await asyncio.wait_for(validation_started.wait(), timeout=TEST_WATCHDOG_SECONDS) with pytest.raises(RuntimeError, match="Strategy execution failed for auxiliary_scorer") as raised: - await wait_for_completion_async(future=task, timeout=5) + await wait_for_completion_async(future=task) assert validation_cancelled.is_set() assert not release_read.is_set() finally: diff --git a/tests/unit/memory/memory_interface/test_interface_observations.py b/tests/unit/memory/memory_interface/test_interface_observations.py index 77abf077b7..7a2f73b419 100644 --- a/tests/unit/memory/memory_interface/test_interface_observations.py +++ b/tests/unit/memory/memory_interface/test_interface_observations.py @@ -13,6 +13,7 @@ from sqlalchemy.dialects import mssql from sqlalchemy.exc import OperationalError, SQLAlchemyError from sqlalchemy.orm import Session +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.mocks import run_memory_session_async from pyrit.memory import MemoryInterface, SQLiteMemory @@ -639,11 +640,13 @@ async def test_concurrent_orm_score_deletions_clean_final_observation(file_memor async def delete_async(score_id: uuid.UUID) -> None: async with await file_memory.get_session_async() as session: entry = await session.get(ScoreEntry, score_id) - await asyncio.wait_for(ready.wait(), timeout=5) + await asyncio.wait_for(ready.wait(), timeout=TEST_WATCHDOG_SECONDS) await session.delete(entry) await session.commit() - await asyncio.wait_for(asyncio.gather(*(delete_async(value.id) for value in (score, replay))), timeout=10) + await asyncio.wait_for( + asyncio.gather(*(delete_async(value.id) for value in (score, replay))), timeout=TEST_WATCHDOG_SECONDS + ) assert file_memory._query_entries(ScoreEntry) == [] assert (await file_memory.get_observations_async(observation_ids=[observation.id])) == [] assert file_memory._query_entries(ObservationMessagePieceEntry) == [] @@ -691,10 +694,10 @@ async def compete_async() -> None: event.listen(engine, "before_cursor_execute", _before_write, named=True) task = asyncio.create_task(compete_async()) try: - await asyncio.wait_for(competing_write.wait(), timeout=5) + await asyncio.wait_for(competing_write.wait(), timeout=TEST_WATCHDOG_SECONDS) assert not task.done() await session.commit() - await asyncio.wait_for(task, timeout=10) + await asyncio.wait_for(task, timeout=TEST_WATCHDOG_SECONDS) finally: await session.rollback() if not task.done(): diff --git a/tests/unit/memory/test_attack_analytics.py b/tests/unit/memory/test_attack_analytics.py index 220f3cf58f..ced72c6729 100644 --- a/tests/unit/memory/test_attack_analytics.py +++ b/tests/unit/memory/test_attack_analytics.py @@ -57,6 +57,7 @@ TargetIdentifier, ) from pyrit.prompt_normalizer import ConverterConfiguration +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.mocks import MockPromptTarget if TYPE_CHECKING: @@ -106,17 +107,17 @@ def make_result( def control() -> QueryControl: """Bound correctness queries without imposing a short CI performance limit.""" - return QueryControl(deadline=time.monotonic() + 120) + return QueryControl(deadline=time.monotonic() + TEST_WATCHDOG_SECONDS) def test_correctness_query_control_retains_a_finite_deadline() -> None: with patch.object(time, "monotonic", return_value=100): query_control = control() - assert query_control.deadline == 220 + assert query_control.deadline == 100 + TEST_WATCHDOG_SECONDS with patch.object(time, "monotonic", return_value=135): query_control.check() - assert query_control.remaining == 85 - with patch.object(time, "monotonic", return_value=220): + assert query_control.remaining == TEST_WATCHDOG_SECONDS - 35 + with patch.object(time, "monotonic", return_value=100 + TEST_WATCHDOG_SECONDS): with pytest.raises(AnalyticsTimeoutException): query_control.check() diff --git a/tests/unit/memory/test_attack_analytics_eval_identity.py b/tests/unit/memory/test_attack_analytics_eval_identity.py index 555981a515..b5199d4d6d 100644 --- a/tests/unit/memory/test_attack_analytics_eval_identity.py +++ b/tests/unit/memory/test_attack_analytics_eval_identity.py @@ -5,7 +5,6 @@ from __future__ import annotations -import time import uuid from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING @@ -25,7 +24,6 @@ AttackTechniqueIdentifierEntry, TargetIdentifierEntry, ) -from pyrit.memory.query_control import QueryControl from pyrit.models import ( AtomicAttackIdentifier, AttackAnalyticsDimension, @@ -43,6 +41,7 @@ TargetIdentifier, ) from pyrit.prompt_target import GandalfLevel, GandalfTarget, HTTPTarget +from unit.memory.test_attack_analytics import control as _control if TYPE_CHECKING: from pyrit.memory import SQLiteMemory @@ -77,10 +76,6 @@ def _result(*, index: int, target: TargetIdentifier, outcome: AttackOutcome, ope ) -def _control() -> QueryControl: - return QueryControl(deadline=time.monotonic() + 20) - - def test_target_evaluation_v1_is_frozen_to_the_current_marker_rules() -> None: first = _target(deployment="deploy-a", endpoint="https://one.example") second = _target(deployment="deploy-b", endpoint="https://two.example") diff --git a/tests/unit/memory/test_sqlite_cancellation.py b/tests/unit/memory/test_sqlite_cancellation.py index f97fe6eed2..5d1dbfbd12 100644 --- a/tests/unit/memory/test_sqlite_cancellation.py +++ b/tests/unit/memory/test_sqlite_cancellation.py @@ -22,7 +22,7 @@ from pyrit.memory import sqlite_memory as sqlite_memory_module from pyrit.memory.sqlite_memory import _CursorClosingSQLiteConnection from pyrit.models import AttackOutcome, AttackResult, MessagePiece, MessageScorable, Score -from unit.async_utils import wait_for_completion_async +from unit.async_utils import TEST_WATCHDOG_SECONDS, wait_for_completion_async from unit.mocks import get_mock_scorer_identifier @@ -125,9 +125,9 @@ async def delayed_connection_close_async(connection: Connection) -> None: ): task = asyncio.create_task(sqlite_memory_async.add_scores_to_memory_async(scores=[persistable_score_async])) try: - await asyncio.wait_for(started.wait(), timeout=30) + await asyncio.wait_for(started.wait(), timeout=TEST_WATCHDOG_SECONDS) task.cancel("cancel score transaction") - await asyncio.wait_for(closing_connection.wait(), timeout=30) + await asyncio.wait_for(closing_connection.wait(), timeout=TEST_WATCHDOG_SECONDS) for _ in range(additional_cancellations): task.cancel("cancel connection cleanup again") await asyncio.sleep(0) @@ -181,9 +181,9 @@ async def write_async() -> None: with patch.object(Connection, "rollback", new=delayed_rollback_async): task = asyncio.create_task(write_async()) try: - await asyncio.wait_for(inserted.wait(), timeout=30) + await asyncio.wait_for(inserted.wait(), timeout=TEST_WATCHDOG_SECONDS) task.cancel("cancel transaction body") - await asyncio.wait_for(rolling_back.wait(), timeout=30) + await asyncio.wait_for(rolling_back.wait(), timeout=TEST_WATCHDOG_SECONDS) for _ in range(additional_cancellations): task.cancel("cancel rollback again") await asyncio.sleep(0) @@ -258,10 +258,10 @@ async def write_async() -> None: patch.object(Connection, "rollback", new=failed_rollback_async), patch.object(Connection, "close", new=observed_close_async), ): - await asyncio.wait_for(inserted.wait(), timeout=30) + await asyncio.wait_for(inserted.wait(), timeout=TEST_WATCHDOG_SECONDS) if cancel_in_body: task.cancel(cancellation_message) - await asyncio.wait_for(rolling_back.wait(), timeout=30) + await asyncio.wait_for(rolling_back.wait(), timeout=TEST_WATCHDOG_SECONDS) if cancel_in_body is False: task.cancel(cancellation_message) await asyncio.sleep(0) @@ -302,7 +302,7 @@ async def test_cancelled_database_worker_finishes_before_return_async( def wait_in_database() -> int: started.set() - if not release.wait(timeout=60): + if not release.wait(timeout=TEST_WATCHDOG_SECONDS): raise RuntimeError("Database worker was not released") finished.set() return 1 @@ -331,9 +331,9 @@ async def read_async() -> None: with patch.object(sqlite_memory_module, "_finish_sqlite_cleanup_async", new=observed_cleanup_async): task = asyncio.create_task(read_async()) try: - assert await asyncio.to_thread(started.wait, 30) + assert await asyncio.to_thread(started.wait, TEST_WATCHDOG_SECONDS) task.cancel("cancel database worker") - await asyncio.wait_for(closing.wait(), timeout=30) + await asyncio.wait_for(closing.wait(), timeout=TEST_WATCHDOG_SECONDS) task.cancel("cancel worker cleanup again") await asyncio.sleep(0) assert not task.done() @@ -406,9 +406,9 @@ async def delayed_session_close_async(session: AsyncSession) -> None: ): task = asyncio.create_task(sqlite_memory_async.add_scores_to_memory_async(scores=[persistable_score_async])) try: - await asyncio.wait_for(started.wait(), timeout=30) + await asyncio.wait_for(started.wait(), timeout=TEST_WATCHDOG_SECONDS) task.cancel("cancel failed cleanup") - await asyncio.wait_for(closing_cleanup.wait(), timeout=30) + await asyncio.wait_for(closing_cleanup.wait(), timeout=TEST_WATCHDOG_SECONDS) for _ in range(additional_cancellations): task.cancel("cancel cleanup again") await asyncio.sleep(0) diff --git a/tests/unit/prompt_target/target/test_github_copilot_target.py b/tests/unit/prompt_target/target/test_github_copilot_target.py index f85a3a45a0..240b9d98c2 100644 --- a/tests/unit/prompt_target/target/test_github_copilot_target.py +++ b/tests/unit/prompt_target/target/test_github_copilot_target.py @@ -15,7 +15,7 @@ from uuid import UUID, uuid4 import pytest -from unit.async_utils import get_defined_tasks, wait_for_completion_async +from unit.async_utils import TEST_WATCHDOG_SECONDS, get_defined_tasks, wait_for_completion_async from unit.mocks import store_message_async from pyrit.models import Message, MessagePiece, MessageScorable, ScoringExpectation @@ -388,7 +388,7 @@ async def send_a_async(*_args: Any, **_kwargs: Any) -> Any: ) second_task: asyncio.Task[list[Message]] | None = None try: - await asyncio.wait_for(first_send_started.wait(), timeout=2.0) + await asyncio.wait_for(first_send_started.wait(), timeout=TEST_WATCHDOG_SECONDS) second_task = asyncio.create_task( target.send_prompt_async( message=_user_message( @@ -398,7 +398,7 @@ async def send_a_async(*_args: Any, **_kwargs: Any) -> Any: ) ) ) - second_response = await asyncio.wait_for(second_task, timeout=2.0) + second_response = await asyncio.wait_for(second_task, timeout=TEST_WATCHDOG_SECONDS) assert not release_first_send.is_set() assert (second_response[0].get_piece().conversation_id, second_response[0].get_piece().converted_value) == ( "conversation-b", @@ -406,7 +406,7 @@ async def send_a_async(*_args: Any, **_kwargs: Any) -> Any: ) release_first_send.set() - first_response = await asyncio.wait_for(first_task, timeout=2.0) + first_response = await asyncio.wait_for(first_task, timeout=TEST_WATCHDOG_SECONDS) assert (first_response[0].get_piece().conversation_id, first_response[0].get_piece().converted_value) == ( "conversation-a", "A", @@ -417,7 +417,7 @@ async def send_a_async(*_args: Any, **_kwargs: Any) -> Any: release_first_send.set() await _cancel_tasks_async(first_task, second_task) with suppress(Exception): - await asyncio.wait_for(target.cleanup_target_async(), timeout=2.0) + await asyncio.wait_for(target.cleanup_target_async(), timeout=TEST_WATCHDOG_SECONDS) sdk.CopilotClient.assert_called_once_with(github_token=None, working_directory=None) client.start.assert_awaited_once() @@ -547,12 +547,12 @@ async def request_async(method: str, params: dict[str, Any]) -> None: ) reset_task = asyncio.create_task(target.reset_conversation_async(conversation_id="conversation-a")) try: - await asyncio.wait_for(release_started.wait(), timeout=2.0) + await asyncio.wait_for(release_started.wait(), timeout=TEST_WATCHDOG_SECONDS) first_failure: BaseException if cancel_release: reset_task.cancel("cancel selected reset") with pytest.raises(asyncio.CancelledError) as cancellation: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert reset_task.cancelled() assert cancellation.value is release_cancellation assert cancellation.value.args == ("cancel selected reset",) @@ -560,7 +560,7 @@ async def request_async(method: str, params: dict[str, Any]) -> None: else: finish_release.set() with pytest.raises(RuntimeError) as first_reset: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert first_reset.value is release_error first_failure = first_reset.value @@ -603,7 +603,7 @@ async def request_async(method: str, params: dict[str, Any]) -> None: finish_release.set() await _cancel_tasks_async(reset_task) with suppress(Exception): - await asyncio.wait_for(target.cleanup_target_async(), timeout=2.0) + await asyncio.wait_for(target.cleanup_target_async(), timeout=TEST_WATCHDOG_SECONDS) @pytest.mark.usefixtures("patch_central_database") @@ -630,14 +630,14 @@ async def disconnect_async() -> None: cleanup_task = asyncio.create_task(target.cleanup_target_async()) reset_task: asyncio.Task[None] | None = None try: - await asyncio.wait_for(disconnect_started.wait(), timeout=2.0) + await asyncio.wait_for(disconnect_started.wait(), timeout=TEST_WATCHDOG_SECONDS) reset_task = asyncio.create_task(target.reset_conversation_async(conversation_id="conversation-a")) await asyncio.sleep(0) assert not reset_task.done() release_disconnect.set() - await asyncio.wait_for(cleanup_task, timeout=2.0) - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) session.disconnect.assert_awaited_once() client.delete_session.assert_not_awaited() client.stop.assert_awaited_once() @@ -676,14 +676,14 @@ async def create_session_async(*_args: Any, **_kwargs: Any) -> Any: ) reset_task: asyncio.Task[None] | None = None try: - await asyncio.wait_for(create_started.wait(), timeout=2.0) + await asyncio.wait_for(create_started.wait(), timeout=TEST_WATCHDOG_SECONDS) reset_task = asyncio.create_task(target.reset_conversation_async(conversation_id=conversation_id)) await asyncio.sleep(0) assert not reset_task.done() release_creation.set() - first_response = await asyncio.wait_for(first_task, timeout=2.0) - await asyncio.wait_for(reset_task, timeout=2.0) + first_response = await asyncio.wait_for(first_task, timeout=TEST_WATCHDOG_SECONDS) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert first_response[0].get_piece().converted_value == "FIRST" session.send_and_wait.assert_awaited_once_with("first prepared", timeout=60.0) client.create_session.assert_awaited_once() @@ -696,7 +696,7 @@ async def create_session_async(*_args: Any, **_kwargs: Any) -> Any: release_creation.set() await _cancel_tasks_async(first_task, reset_task) with suppress(Exception): - await asyncio.wait_for(target.cleanup_target_async(), timeout=2.0) + await asyncio.wait_for(target.cleanup_target_async(), timeout=TEST_WATCHDOG_SECONDS) @pytest.mark.usefixtures("patch_central_database") @@ -740,7 +740,7 @@ async def disconnect_b_async() -> None: cleanup_task = asyncio.create_task(target.cleanup_target_async()) reset_task: asyncio.Task[None] | None = None try: - await asyncio.wait_for(disconnect_started.wait(), timeout=2.0) + await asyncio.wait_for(disconnect_started.wait(), timeout=TEST_WATCHDOG_SECONDS) if not reset_before_cleanup: session_a.disconnect.assert_awaited_once() reset_task = asyncio.create_task(target.reset_conversation_async(conversation_id="conversation-a")) @@ -752,7 +752,7 @@ async def disconnect_b_async() -> None: release_disconnect.set() with pytest.raises(RuntimeError, match="B disconnect failed"): - await asyncio.wait_for(cleanup_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) session_b.disconnect.assert_awaited_once() client.stop.assert_awaited_once() finally: @@ -839,41 +839,41 @@ async def disconnect_b_async() -> None: cleanup_task = asyncio.create_task(target.cleanup_target_async()) reset_task: asyncio.Task[None] | None = None try: - await asyncio.wait_for(a_release_started.wait(), timeout=2.0) + await asyncio.wait_for(a_release_started.wait(), timeout=TEST_WATCHDOG_SECONDS) reset_task = asyncio.create_task(target.reset_conversation_async(conversation_id="conversation-a")) await asyncio.sleep(0) assert not reset_task.done() release_a.set() - await asyncio.wait_for(a_release_finished.wait(), timeout=2.0) - await asyncio.wait_for(b_release_started.wait(), timeout=2.0) + await asyncio.wait_for(a_release_finished.wait(), timeout=TEST_WATCHDOG_SECONDS) + await asyncio.wait_for(b_release_started.wait(), timeout=TEST_WATCHDOG_SECONDS) assert a_release_finished.is_set() if failure_kind == "caller-cancel": reset_task.cancel() with pytest.raises(asyncio.CancelledError): - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert reset_task.cancelled() assert not cleanup_task.done() client.stop.assert_not_awaited() release_b.set() - await asyncio.wait_for(cleanup_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) else: release_b.set() if failure_kind == "error": expected_cleanup_failure = a_failure if failed_conversation == "conversation-a" else b_failure with pytest.raises(RuntimeError) as cleanup_error: - await asyncio.wait_for(cleanup_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) assert cleanup_error.value is expected_cleanup_failure if failed_conversation == "conversation-a": with pytest.raises(RuntimeError) as reset_error: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert reset_error.value is a_failure else: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) elif failure_kind == "release-cancel-stop-error": with pytest.raises(BaseExceptionGroup) as cleanup_error: - await asyncio.wait_for(cleanup_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) expected_release_failure = ( selected_cancellation if failed_conversation == "conversation-a" else unrelated_cancellation ) @@ -882,18 +882,18 @@ async def disconnect_b_async() -> None: assert cleanup_error.value.exceptions[1] is stop_failure if failed_conversation == "conversation-a": with pytest.raises(BaseExceptionGroup) as reset_error: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert reset_error.value is cleanup_error.value else: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) else: with pytest.raises(asyncio.CancelledError): - await asyncio.wait_for(cleanup_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) if failed_conversation == "conversation-a": with pytest.raises(asyncio.CancelledError): - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) else: - await asyncio.wait_for(reset_task, timeout=2.0) + await asyncio.wait_for(reset_task, timeout=TEST_WATCHDOG_SECONDS) assert client.create_session.await_count == 2 session_a.send_and_wait.assert_awaited_once() @@ -1203,10 +1203,10 @@ async def send_and_wait_async(*_args: Any, **_kwargs: Any) -> None: ) try: - await asyncio.wait_for(send_started.wait(), timeout=2.0) + await asyncio.wait_for(send_started.wait(), timeout=TEST_WATCHDOG_SECONDS) send_task.cancel() with pytest.raises(asyncio.CancelledError) as exc_info: - await asyncio.wait_for(send_task, timeout=2.0) + await asyncio.wait_for(send_task, timeout=TEST_WATCHDOG_SECONDS) assert exc_info.value.__cause__ is release_error client.delete_session.assert_awaited_once_with(session.session_id) finally: @@ -1245,7 +1245,7 @@ async def send_and_wait_async(*_args: Any, **_kwargs: Any) -> Any: queued_task: asyncio.Task[list[Message]] | None = None cleanup_task: asyncio.Task[None] | None = None try: - await asyncio.wait_for(first_send_started.wait(), timeout=2.0) + await asyncio.wait_for(first_send_started.wait(), timeout=TEST_WATCHDOG_SECONDS) queued_task = asyncio.create_task( target.send_prompt_async( message=_user_message(conversation_id=conversation_id, original_value="queued"), @@ -1263,18 +1263,18 @@ async def send_and_wait_async(*_args: Any, **_kwargs: Any) -> Any: original_value="fresh", ) ), - timeout=2.0, + timeout=TEST_WATCHDOG_SECONDS, ) _assert_no_resource_release(client=client) release_first_send.set() - first_response = await asyncio.wait_for(first_task, timeout=2.0) + first_response = await asyncio.wait_for(first_task, timeout=TEST_WATCHDOG_SECONDS) assert first_response[0].get_piece().converted_value == "FIRST" assert queued_task is not None with pytest.raises(RuntimeError, match="cleaned up"): - await asyncio.wait_for(queued_task, timeout=2.0) + await asyncio.wait_for(queued_task, timeout=TEST_WATCHDOG_SECONDS) assert cleanup_task is not None - await asyncio.wait_for(cleanup_task, timeout=2.0) + await asyncio.wait_for(cleanup_task, timeout=TEST_WATCHDOG_SECONDS) finally: release_first_send.set() await _cancel_tasks_async(first_task, queued_task, cleanup_task) @@ -1424,12 +1424,12 @@ async def start_async() -> None: ) ) try: - await asyncio.wait_for(startup_entered.wait(), timeout=2.0) + await asyncio.wait_for(startup_entered.wait(), timeout=TEST_WATCHDOG_SECONDS) with pytest.raises(RuntimeError, match="new target"): target.set_model_name(model_name="gpt-5.4") release_startup.set() - response = await asyncio.wait_for(first_task, timeout=2.0) + response = await asyncio.wait_for(first_task, timeout=TEST_WATCHDOG_SECONDS) assert response[0].get_piece().converted_value == "HELLO" assert client.create_session.await_args.kwargs["model"] == "gpt-5-mini" with pytest.raises(RuntimeError, match="new target"): @@ -1440,7 +1440,7 @@ async def start_async() -> None: release_startup.set() await _cancel_tasks_async(first_task) with suppress(Exception): - await asyncio.wait_for(target.cleanup_target_async(), timeout=2.0) + await asyncio.wait_for(target.cleanup_target_async(), timeout=TEST_WATCHDOG_SECONDS) @pytest.mark.usefixtures("patch_central_database") @@ -1576,10 +1576,10 @@ async def stall_send_async(*_args: Any, **_kwargs: Any) -> None: request.side_effect = stall_send_async send_task = asyncio.create_task(_send_normalized_async(target=target, original_value="Reply exactly HELLO.")) try: - await asyncio.wait_for(send_entered.wait(), timeout=10.0) + await asyncio.wait_for(send_entered.wait(), timeout=TEST_WATCHDOG_SECONDS) with pytest.raises(Exception, match="Error sending prompt with conversation ID:") as exc_info: # A bare watchdog TimeoutError must not satisfy the normalizer-wrapped failure. - await asyncio.wait_for(send_task, timeout=10.0) + await asyncio.wait_for(send_task, timeout=TEST_WATCHDOG_SECONDS) assert isinstance(exc_info.value.__cause__, TimeoutError) request.assert_awaited_once() assert request.await_args.args[0] == "session.send" @@ -2170,11 +2170,11 @@ async def stop_async() -> None: target.send_prompt_async(message=_user_message(conversation_id="failed-startup", original_value="hello")) ) try: - await asyncio.wait_for(stop_started.wait(), timeout=2.0) + await asyncio.wait_for(stop_started.wait(), timeout=TEST_WATCHDOG_SECONDS) if stop_failure == "caller-cancel": send_task.cancel("caller cancelled during startup cleanup") with pytest.raises(asyncio.CancelledError) as cancellation: - await asyncio.wait_for(send_task, timeout=2.0) + await asyncio.wait_for(send_task, timeout=TEST_WATCHDOG_SECONDS) assert send_task.cancelled() assert cancellation.value is caller_cancellation assert cancellation.value.args == ("caller cancelled during startup cleanup",) @@ -2182,7 +2182,7 @@ async def stop_async() -> None: else: finish_stop.set() with pytest.raises(BaseExceptionGroup) as failure: - await asyncio.wait_for(send_task, timeout=2.0) + await asyncio.wait_for(send_task, timeout=TEST_WATCHDOG_SECONDS) assert failure.value.exceptions == (startup_error, cleanup_error) assert failure.value.__cause__ is startup_error assert not send_task.cancelled() @@ -2217,7 +2217,7 @@ async def stop_async() -> None: finish_stop.set() await _cancel_tasks_async(send_task) with suppress(Exception): - await asyncio.wait_for(target.cleanup_target_async(), timeout=2.0) + await asyncio.wait_for(target.cleanup_target_async(), timeout=TEST_WATCHDOG_SECONDS) @pytest.mark.usefixtures("patch_central_database", "mock_copilot_startup_io") @@ -2249,7 +2249,7 @@ async def stop_async() -> None: target = GitHubCopilotTarget(model_name="gpt-5-mini") request_task = asyncio.create_task(_send_normalized_async(target=target, original_value="Reply exactly HELLO.")) try: - await asyncio.wait_for(allocated.wait(), timeout=2.0) + await asyncio.wait_for(allocated.wait(), timeout=TEST_WATCHDOG_SECONDS) request_task.cancel() with pytest.raises(asyncio.CancelledError) as exc_info: await request_task @@ -2389,7 +2389,7 @@ def construct_client(*, github_token: str | None, working_directory: str | None) try: loop.call_soon_threadsafe(constructor_entered.set) # The bound lets a blocked event loop escape without satisfying the responsiveness assertion. - released_while_constructing = release.wait(timeout=5.0) + released_while_constructing = release.wait(timeout=TEST_WATCHDOG_SECONDS) return client finally: loop.call_soon_threadsafe(constructor_finished.set) @@ -2410,11 +2410,11 @@ async def release_constructor_async() -> None: ) release_task = asyncio.create_task(release_constructor_async()) try: - await asyncio.wait_for(asyncio.gather(request_task, release_task), timeout=10.0) + await asyncio.wait_for(asyncio.gather(request_task, release_task), timeout=TEST_WATCHDOG_SECONDS) finally: release.set() await _cancel_tasks_async(request_task, release_task) - await asyncio.wait_for(constructor_finished.wait(), timeout=5.0) + await asyncio.wait_for(constructor_finished.wait(), timeout=TEST_WATCHDOG_SECONDS) await target.cleanup_target_async() client.stop.assert_awaited_once() diff --git a/tests/unit/prompt_target/test_discover_target_capabilities.py b/tests/unit/prompt_target/test_discover_target_capabilities.py index 1d17a17290..e90e9898d4 100644 --- a/tests/unit/prompt_target/test_discover_target_capabilities.py +++ b/tests/unit/prompt_target/test_discover_target_capabilities.py @@ -4,6 +4,7 @@ import asyncio import json import logging +from collections.abc import Awaitable from copy import deepcopy from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -12,6 +13,7 @@ import pytest from mcp.types import Tool as MCPToolDefinition from openai.types.chat import ChatCompletion +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.mocks import MockPromptTarget from pyrit.models import Message, MessagePiece, PromptDataType, RequestTraceContext @@ -792,15 +794,27 @@ async def direct_async() -> str: original_value_data_type="function_call", ).to_message() + requested = asyncio.Event() + async def finish_async(**kwargs: object) -> Message: if outcome == "error": raise ValueError("Probe request failed") if outcome == "cancel": raise asyncio.CancelledError if outcome == "timeout": + requested.set() await asyncio.Event().wait() return response + original_wait_for = asyncio.wait_for + + async def expire_once_requested_async(awaitable: Awaitable[list[Message]], *, timeout: float) -> list[Message]: + # Expire the probe deadline once the request is pending. A wall-clock deadline can + # expire during the preceding memory access on a stalled runner, so the request never starts. + probe = asyncio.ensure_future(awaitable) + await original_wait_for(requested.wait(), timeout=TEST_WATCHDOG_SECONDS) + return await original_wait_for(probe, timeout=0) + with ( patch.object(provider, "execution_scope_async") as scope, patch.object(provider, "get_tools_async", new_callable=AsyncMock) as discover, @@ -816,13 +830,16 @@ async def probe_async() -> None: capabilities=[CapabilityName.JSON_OUTPUT], test_modalities=set(), retries=0, - # The deadline includes async memory access before the mocked request. - per_probe_timeout_s=5.0, + per_probe_timeout_s=TEST_WATCHDOG_SECONDS, ) if outcome == "cancel": with pytest.raises(asyncio.CancelledError): await probe_async() + elif outcome == "timeout": + with patch.object(asyncio, "wait_for", side_effect=expire_once_requested_async) as deadline: + await probe_async() + deadline.assert_awaited_once() else: await probe_async() send.assert_awaited_once() @@ -1188,7 +1205,7 @@ async def test_returns_target_capabilities_assembled_from_probes(self) -> None: target._configuration = TargetConfiguration(capabilities=declared) target._send_prompt_to_target_async = AsyncMock(return_value=_ok_response()) # type: ignore[method-assign] - result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=5.0) + result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=TEST_WATCHDOG_SECONDS) assert isinstance(result, TargetCapabilities) # Single-piece probes that don't touch memory always succeed when @@ -1286,7 +1303,7 @@ async def test_discover_target_capabilities_async_forwards_test_modalities(self, target=target, test_modalities={extra_combo}, test_assets={"image_path": image_asset}, - per_probe_timeout_s=2.0, + per_probe_timeout_s=TEST_WATCHDOG_SECONDS, ) # The undeclared combination is in the result only if test_modalities was forwarded. @@ -1306,7 +1323,7 @@ async def test_discover_target_capabilities_async_preserves_declared_modalities_ target=target, test_modalities={probed_combo}, test_assets={"image_path": image_asset}, - per_probe_timeout_s=2.0, + per_probe_timeout_s=TEST_WATCHDOG_SECONDS, ) assert result.input_modalities == frozenset({declared_combo, probed_combo}) @@ -1319,7 +1336,7 @@ async def test_discover_target_capabilities_async_forwards_capabilities(self) -> await discover_target_capabilities_async( target=target, capabilities={CapabilityName.JSON_OUTPUT}, - per_probe_timeout_s=2.0, + per_probe_timeout_s=TEST_WATCHDOG_SECONDS, ) # Only the JSON_OUTPUT probe (1 send) and the modality probe(s) should run; @@ -1346,7 +1363,7 @@ async def test_discover_target_capabilities_async_preserves_declared_when_capabi result = await discover_target_capabilities_async( target=target, capabilities={CapabilityName.JSON_OUTPUT}, - per_probe_timeout_s=2.0, + per_probe_timeout_s=TEST_WATCHDOG_SECONDS, ) # The probed capability reflects the queried result. @@ -1375,7 +1392,7 @@ async def selective_send(*, normalized_conversation: list[Message]) -> list[Mess target._send_prompt_to_target_async = AsyncMock(side_effect=selective_send) # type: ignore[method-assign] - result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=2.0) + result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=TEST_WATCHDOG_SECONDS) assert result.supports_multi_turn is False assert result.supports_editable_history is False @@ -1393,7 +1410,7 @@ async def test_discover_target_capabilities_async_accepts_single_pass_iterable(s result = await discover_target_capabilities_async( target=target, capabilities=gen, - per_probe_timeout_s=2.0, + per_probe_timeout_s=TEST_WATCHDOG_SECONDS, ) assert result.supports_json_output is True @@ -1408,7 +1425,9 @@ async def test_discover_target_capabilities_async_apply_installs_capabilities_on assert target.capabilities.supports_multi_turn is False - result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=2.0, apply=True) + result = await discover_target_capabilities_async( + target=target, per_probe_timeout_s=TEST_WATCHDOG_SECONDS, apply=True + ) assert target.capabilities == result assert target.capabilities.supports_multi_turn is True @@ -1420,7 +1439,7 @@ async def test_discover_target_capabilities_async_apply_defaults_to_false(self) target._configuration = TargetConfiguration(capabilities=declared) target._send_prompt_to_target_async = AsyncMock(return_value=_ok_response()) # type: ignore[method-assign] - result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=2.0) + result = await discover_target_capabilities_async(target=target, per_probe_timeout_s=TEST_WATCHDOG_SECONDS) # Result reflects probe; target capabilities remain at declared values. assert result.supports_multi_turn is True diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index 2177225bab..6b5bd49136 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -10,6 +10,7 @@ from unittest.mock import ANY, AsyncMock, MagicMock, PropertyMock, patch import pytest +from unit.async_utils import TEST_WATCHDOG_SECONDS from unit.mocks import MockPromptTarget, make_scenario_identifier, make_scenario_result from pyrit.analytics import compute_scenario_statistics @@ -2152,7 +2153,7 @@ async def run_async(*args, **kwargs): if in_flight >= 3: started.set() try: - await asyncio.wait_for(started.wait(), timeout=2.0) + await asyncio.wait_for(started.wait(), timeout=TEST_WATCHDOG_SECONDS) finally: async with lock: in_flight -= 1 @@ -2285,7 +2286,7 @@ async def sibling_run_async(**_kwargs): await scenario.initialize_async() with pytest.raises(asyncio.CancelledError, match="atomic attack cancelled"): - await asyncio.wait_for(scenario.run_async(), timeout=5) + await asyncio.wait_for(scenario.run_async(), timeout=TEST_WATCHDOG_SECONDS) mock_atomic_attacks[2].run_async.assert_not_called() assert not scenario._active_atomic_groups diff --git a/tests/unit/scenario/core/test_scenario_partial_results.py b/tests/unit/scenario/core/test_scenario_partial_results.py index 2047b001bb..704eeae3d9 100644 --- a/tests/unit/scenario/core/test_scenario_partial_results.py +++ b/tests/unit/scenario/core/test_scenario_partial_results.py @@ -8,7 +8,7 @@ from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch import pytest -from unit.async_utils import wait_for_completion_async +from unit.async_utils import TEST_WATCHDOG_SECONDS, wait_for_completion_async from unit.mocks import MockPromptTarget from pyrit.exceptions import ScenarioPartialFailureException @@ -605,7 +605,7 @@ async def run_async(**_kwargs: object) -> AttackExecutorResult[AttackResult]: parent = asyncio.create_task(scenario.run_async()) try: - await asyncio.wait_for(all_started.wait(), timeout=30) + await asyncio.wait_for(all_started.wait(), timeout=TEST_WATCHDOG_SECONDS) release_workers.set() parent.cancel("stop scenario") with pytest.raises(asyncio.CancelledError, match="stop scenario"): @@ -677,10 +677,10 @@ async def sibling_run_async(**_kwargs: object) -> AttackExecutorResult[AttackRes parent = asyncio.create_task(scenario.run_async()) try: - await asyncio.wait_for(all_started.wait(), timeout=30) + await asyncio.wait_for(all_started.wait(), timeout=TEST_WATCHDOG_SECONDS) release_sibling.set() parent.cancel("stop scenario") - await asyncio.wait_for(cleanup_started.wait(), timeout=30) + await asyncio.wait_for(cleanup_started.wait(), timeout=TEST_WATCHDOG_SECONDS) assert not parent.done() assert worker_tasks[0].cancelling() == 1 [stored] = await scenario._memory.get_scenario_results_async( @@ -747,7 +747,7 @@ async def cancel_then_resume_async() -> ScenarioResult: parent = asyncio.create_task(cancel_then_resume_async()) try: - await asyncio.wait_for(started.wait(), timeout=30) + await asyncio.wait_for(started.wait(), timeout=TEST_WATCHDOG_SECONDS) parent.cancel("stop first run") result = await wait_for_completion_async(future=parent) assert result.scenario_run_state is ScenarioRunState.COMPLETED @@ -791,7 +791,7 @@ async def sibling_run_async(**_kwargs): task = asyncio.create_task(scenario.run_async()) try: - await asyncio.wait_for(cleanup_started.wait(), timeout=30) + await asyncio.wait_for(cleanup_started.wait(), timeout=TEST_WATCHDOG_SECONDS) task.cancel("caller cancelled during cleanup") await asyncio.sleep(0) allow_cleanup.set() @@ -858,10 +858,10 @@ async def observe_fast_worker_async(**kwargs): ): parent = asyncio.create_task(scenario.run_async()) try: - await asyncio.wait_for(all_started.wait(), timeout=30) + await asyncio.wait_for(all_started.wait(), timeout=TEST_WATCHDOG_SECONDS) parent.cancel("stop scenario") - await asyncio.wait_for(cleanup_started.wait(), timeout=30) - await asyncio.wait_for(fast_worker_finished.wait(), timeout=30) + await asyncio.wait_for(cleanup_started.wait(), timeout=TEST_WATCHDOG_SECONDS) + await asyncio.wait_for(fast_worker_finished.wait(), timeout=TEST_WATCHDOG_SECONDS) assert sends["slow"].cancelling() == 1 assert not cleanup_finished.is_set() assert not parent.done()