diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 041345b8..3a95026e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,11 +59,15 @@ jobs: if-no-files-found: ignore mutation: - name: Mutation Gate + name: Mutation Gate (${{ matrix.shard }}) runs-on: ubuntu-latest - timeout-minutes: 40 + timeout-minutes: 90 permissions: contents: read + strategy: + fail-fast: false + matrix: + shard: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22] steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 with: @@ -76,15 +80,16 @@ jobs: with: version: "0.12.17" - run: uv sync --locked - - name: Mutate changed and critical runtime modules + - name: Mutate all handwritten runtime modules in this shard env: - MUTATION_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.event.before }} - run: uv run --locked poe mutation + MUTATION_SHARD_INDEX: ${{ matrix.shard }} + MUTATION_SHARD_COUNT: 23 + run: uv run --locked poe mutation-full - name: Preserve mutation outcomes if: ${{ always() }} uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: - name: mutation-outcomes + name: mutation-outcomes-${{ matrix.shard }} path: reports/mutation.json if-no-files-found: error @@ -94,7 +99,23 @@ jobs: needs: [test, mutation] runs-on: ubuntu-latest timeout-minutes: 5 + permissions: + contents: read steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 + with: + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.12" + - uses: actions/download-artifact@37930b1c2abaa49bbe596cd826c3c89aef350131 # v7.0.0 + with: + pattern: mutation-outcomes-* + path: reports/shards + - name: Require every runtime module in one mutation shard + run: | + git ls-files -z 'src/volcano_sdk/*.py' > reports/shards/source.bin + python -m scripts.check_mutation_shards reports/shards reports/shards/source.bin 23 - name: Require every mandatory job to succeed env: RESULTS: ${{ toJSON(needs.*.result) }} diff --git a/maintainers/mutation-testing.md b/maintainers/mutation-testing.md index db16609c..42c13552 100644 --- a/maintainers/mutation-testing.md +++ b/maintainers/mutation-testing.md @@ -1,21 +1,29 @@ # Mutation testing -`uv run --locked poe quality` runs all native checks and mutates every changed -handwritten runtime module plus the lock acquisition, guard, renewal, and worker -modules. CI uses the same `checks` and `mutation` tasks in separate jobs, then -requires both through `Quality Gate`. The weekly `poe mutation-full` task audits -every handwritten runtime module without a debt baseline. +`uv run --locked poe quality` runs all native checks and mutates every handwritten +runtime module. CI runs `checks` and divides `mutation-full` across eight jobs; +the required `Quality Gate` checks every job and the complete runtime inventory. +The weekly full audit also runs `mutation-full` without sharding. `poe mutation` +remains a faster local diagnostic for changed modules and the lock runtime. Mutmut's [native configuration](https://mutmut.readthedocs.io/en/latest/) lives in `pyproject.toml`; it excludes only the generated OpenAPI client. Mutmut can -select modules by name but has no Git-changed-module option and returns success -when mutants survive. `scripts/mutation.sh` selects module names from Git and -`scripts/mutation_results.py` reads only those modules' native metadata. The -report at `reports/mutation.json` distinguishes killed, statically invalid, +select modules by name but cannot divide a run across CI jobs or verify that +their combined results cover every tracked source module. GitHub's matrix +reports job success without checking that source inventory. +`scripts/mutation.sh` assigns Git-tracked runtime modules to shards, +`scripts/mutation_results.py` reads each shard's native metadata, and +`scripts/check_mutation_shards.py` checks that all 23 reports cover each +handwritten module exactly once. The export-only `__init__.py` shares a shard +with a mutatable module. `reports/mutation.json` distinguishes killed, +statically invalid, surviving, uncovered, timed-out, crashed, interrupted, and missing results. A pytest internal error is a harness crash, not a killed mutant. -The pinned Pyrefly check rejects type-invalid realtime and auth mutants before pytest; +The pinned Pyrefly check rejects type-invalid mutants in every handwritten +runtime module before pytest; the report counts these as `type_checked`, separately from test-killed mutants. +The runner checks the unmutated source first, so an existing type error cannot +make every mutant appear invalid. Surviving, uncovered, timed-out, crashed, and incomplete mutants still fail. Mutmut passes pytest `-x` so a selected test's first assertion failure kills the mutant before a later selected test can hang on the same defect. Mutants that diff --git a/maintainers/quality-policy.lock.json b/maintainers/quality-policy.lock.json index 54d94fc7..22f43e6a 100644 --- a/maintainers/quality-policy.lock.json +++ b/maintainers/quality-policy.lock.json @@ -299,16 +299,9 @@ "pyrefly", "check", "--output-format=json", - "src/volcano_sdk/realtime.py", - "src/volcano_sdk/auth.py", - "src/volcano_sdk/storage.py", - "src/volcano_sdk/functions.py", - "src/volcano_sdk/durable_authoring.py", - "src/volcano_sdk/database.py", - "src/volcano_sdk/_function_resolution.py", - "src/volcano_sdk/_transport.py", - "src/volcano_sdk/_session.py", - "src/volcano_sdk/_session_operations.py" + "--project-excludes", + "src/volcano_sdk/_generated/**", + "src/volcano_sdk/*.py" ] }, "mypy": { @@ -400,7 +393,10 @@ "generated": "python -m scripts.check_openapi", "lint": "ruff check --config pyproject.toml .", "mutation": "bash scripts/mutation.sh", - "mutation-full": "MUTATION_FULL=1 bash scripts/mutation.sh", + "mutation-full": { + "interpreter": "bash", + "shell": "MUTATION_FULL=1 bash scripts/mutation.sh" + }, "mypy": "mypy --config-file pyproject.toml", "package-check": { "interpreter": "bash", @@ -412,7 +408,7 @@ "policy": "bash scripts/check_quality_policy.sh", "quality": [ "checks", - "mutation" + "mutation-full" ], "test": { "interpreter": "bash", diff --git a/pyproject.toml b/pyproject.toml index d012d250..bdcb43c7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -458,16 +458,8 @@ partial_branches = [] source_paths = ["src/volcano_sdk"] type_check_command = [ "pyrefly", "check", "--output-format=json", - "src/volcano_sdk/realtime.py", - "src/volcano_sdk/auth.py", - "src/volcano_sdk/storage.py", - "src/volcano_sdk/functions.py", - "src/volcano_sdk/durable_authoring.py", - "src/volcano_sdk/database.py", - "src/volcano_sdk/_function_resolution.py", - "src/volcano_sdk/_transport.py", - "src/volcano_sdk/_session.py", - "src/volcano_sdk/_session_operations.py", + "--project-excludes", "src/volcano_sdk/_generated/**", + "src/volcano_sdk/*.py", ] cache_invalidation_files = ["tests/**/*.py"] pytest_add_cli_args_test_selection = ["tests/unit"] @@ -486,7 +478,7 @@ also_copy = [ ] [tool.poe.tasks] -quality = ["checks", "mutation"] +quality = ["checks", "mutation-full"] checks = ["policy", "audit", "generated", "lint", "format-check", "types", "test", "coverage", "contract-check", "package-check", "package-extras"] policy = "bash scripts/check_quality_policy.sh" build = "uv build --no-sources --require-hashes" @@ -498,7 +490,7 @@ types = ["mypy", "basedpyright"] mypy = "mypy --config-file pyproject.toml" basedpyright = "basedpyright --project pyproject.toml" mutation = "bash scripts/mutation.sh" -mutation-full = "MUTATION_FULL=1 bash scripts/mutation.sh" +mutation-full = { shell = "MUTATION_FULL=1 bash scripts/mutation.sh", interpreter = "bash" } [tool.poe.tasks.test] interpreter = "bash" diff --git a/scripts/check_mutation_shards.py b/scripts/check_mutation_shards.py new file mode 100644 index 00000000..4a74c95b --- /dev/null +++ b/scripts/check_mutation_shards.py @@ -0,0 +1,118 @@ +"""Require every runtime module in exactly one successful mutation shard.""" + +from __future__ import annotations + +import json +import os +import sys +from collections import Counter +from pathlib import Path +from typing import cast + + +class MutationShardError(Exception): + """A required mutation shard or runtime module is missing or failed.""" + + +def source_modules(path: Path) -> set[str]: + """Read the tracked handwritten runtime inventory. + + Returns: + Runtime module paths. + + """ + return { + os.fsdecode(raw) + for raw in path.read_bytes().split(b"\0") + if raw and not raw.startswith(b"src/volcano_sdk/_generated/") + } + + +def valid_outcomes(raw: object) -> bool: + """Accept only nonempty native killed or statically invalid outcomes. + + Returns: + Whether the result counts are complete and successful. + + """ + if not isinstance(raw, dict): + return False + counts = cast("dict[object, object]", raw) + if not counts or any( + name not in {"killed", "type_checked"} + or not isinstance(count, int) + or isinstance(count, bool) + or count < 0 + for name, count in counts.items() + ): + return False + validated_counts = cast("dict[str, int]", counts) + return sum(validated_counts.values()) > 0 + + +def shard_modules(path: Path) -> list[str]: + """Validate one native Mutmut result report. + + Returns: + Paths audited by this shard. + + Raises: + MutationShardError: The report is incomplete or contains failed mutants. + + """ + raw = cast("object", json.loads(path.read_text(encoding="utf-8"))) + if not isinstance(raw, dict): + msg = f"Invalid mutation report: {path}" + raise MutationShardError(msg) + report = cast("dict[str, object]", raw) + modules = report.get("modules") + outcomes = report.get("outcomes") + failures = report.get("failures") + if ( + not isinstance(modules, list) + or not modules + or not isinstance(failures, list) + or failures + ): + msg = f"Incomplete mutation report: {path}" + raise MutationShardError(msg) + typed_modules = cast("list[object]", modules) + if not all(isinstance(module, str) for module in typed_modules): + msg = f"Invalid mutation modules: {path}" + raise MutationShardError(msg) + if not valid_outcomes(outcomes): + msg = f"Failed mutation outcomes: {path}" + raise MutationShardError(msg) + return cast("list[str]", modules) + + +def main(root: Path, source_path: Path, count: int) -> int: + """Check shard presence, successful results, and complete source coverage. + + Returns: + Zero only when every runtime module appears exactly once. + + Raises: + MutationShardError: A shard or source module is missing or invalid. + + """ + expected = source_modules(source_path) + if not expected or count < 1: + msg = "Empty source inventory or invalid shard count" + raise MutationShardError(msg) + actual: Counter[str] = Counter() + for index in range(count): + report = root / f"mutation-outcomes-{index}" / "mutation.json" + actual.update(shard_modules(report)) + extra = sorted(set(actual) - expected) + missing = sorted(expected - set(actual)) + duplicate = sorted(name for name, occurrences in actual.items() if occurrences != 1) + if extra or missing or duplicate: + msg = f"Mutation shard inventory differs: {extra=}, {missing=}, {duplicate=}" + raise MutationShardError(msg) + print(f"Mutation shards covered {len(expected)} handwritten runtime modules") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(Path(sys.argv[1]), Path(sys.argv[2]), int(sys.argv[3]))) diff --git a/scripts/check_quality_policy.py b/scripts/check_quality_policy.py index 784e2d51..9623b622 100644 --- a/scripts/check_quality_policy.py +++ b/scripts/check_quality_policy.py @@ -18,7 +18,7 @@ from collections.abc import Iterable GENERATED = "src/volcano_sdk/_generated" -LOCK_SHA256 = "489b1a41632f67d528dbb10267bb8921c38c78726c430d990b4ee99ebdd9236a" +LOCK_SHA256 = "64c4fcc180777984cca637b2cbe41684b984ab8e33884da9aa6d0def1ddd048e" TYPE_FIXTURES = { "tests/typing/contract_steps.py", "tests/typing/durable_callbacks.py", diff --git a/scripts/mutation.sh b/scripts/mutation.sh index d05e3ae7..532cba71 100644 --- a/scripts/mutation.sh +++ b/scripts/mutation.sh @@ -4,6 +4,9 @@ set -euo pipefail # Forked macOS workers must not query SystemConfiguration through urllib/httpx. export NO_PROXY='*' no_proxy='*' +# An invalid baseline must not turn every mutant into a type-checked result. +pyrefly check --output-format=json --project-excludes 'src/volcano_sdk/_generated/**' 'src/volcano_sdk/*.py' + # Refresh mutmut's test-to-mutant map so newly added tests are selected. rm -rf -- mutants @@ -51,11 +54,30 @@ else done < reports/mutation-changed.bin fi +if [[ -n ${MUTATION_SHARD_INDEX:-} ]]; then + if [[ ${MUTATION_FULL:-0} != 1 || ! ${MUTATION_SHARD_INDEX} =~ ^[0-9]+$ || ! ${MUTATION_SHARD_COUNT:-} =~ ^[0-9]+$ || ${MUTATION_SHARD_COUNT:-0} -eq 0 || ${MUTATION_SHARD_INDEX} -ge ${MUTATION_SHARD_COUNT} ]]; then + echo 'Invalid full-mutation shard configuration' >&2 + exit 1 + fi + all_modules=("${modules[@]}") + modules=() + for index in "${!all_modules[@]}"; do + if (( index % MUTATION_SHARD_COUNT == MUTATION_SHARD_INDEX )); then + modules+=("${all_modules[index]}") + fi + done +fi + +if (( ${#modules[@]} == 0 )); then + echo 'No runtime modules selected for mutation' >&2 + exit 1 +fi + for path in "${modules[@]}"; do printf '%s\0' "$path" >> "$targets" done -if [[ ${MUTATION_FULL:-0} == 1 ]]; then +if [[ ${MUTATION_FULL:-0} == 1 && -z ${MUTATION_SHARD_INDEX:-} ]]; then if ! mutmut run --max-children 1; then printf '%s\0' "full mutation run" >> "$failed" fi diff --git a/src/volcano_sdk/_realtime_fetch_worker.py b/src/volcano_sdk/_realtime_fetch_worker.py index 421e73f5..6ebdb9bc 100644 --- a/src/volcano_sdk/_realtime_fetch_worker.py +++ b/src/volcano_sdk/_realtime_fetch_worker.py @@ -145,8 +145,7 @@ async def close(self) -> None: task = self._task if self._closed and (task is None or task.cancelled()): return - if not self._closed: - self._closed = True + self._closed = True self._raise_worker_failure() if task is not None and self._stop_task is None: self._stop_task = asyncio.create_task(self._queue.put(_STOP_WORKER)) @@ -157,15 +156,13 @@ async def close(self) -> None: async def abort(self) -> None: """Discard obsolete jobs and stop without waiting for row fetches.""" self._closed = True - task = self._task - stop_task = self._stop_task - if stop_task is not None and not stop_task.done(): - _ = stop_task.cancel() - if task is not None and not task.done(): - _ = task.cancel() pending = tuple( - candidate for candidate in (task, stop_task) if candidate is not None + candidate + for candidate in (self._stop_task, self._task) + if candidate is not None ) + for candidate in pending: + _ = candidate.cancel() if pending: _ = await asyncio.gather(*pending, return_exceptions=True) self._discard_pending() @@ -182,41 +179,56 @@ async def _put_while_running( ) -> None: put_task = asyncio.create_task(self._queue.put(job)) try: - completed, _pending = await asyncio.wait( - (task, put_task), - return_when=asyncio.FIRST_COMPLETED, - ) + await self._await_put_or_worker(task, put_task) finally: if not put_task.done(): _ = put_task.cancel() _ = await asyncio.gather(put_task, return_exceptions=True) - if task in completed: + if task.done(): if task.cancelled(): raise RuntimeError(_WORKER_CLOSED) task.result() await put_task + @staticmethod + async def _await_put_or_worker( + task: asyncio.Task[None], + put_task: asyncio.Task[None], + ) -> None: + ready: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def notify(_completed: asyncio.Task[None]) -> None: + if not ready.done(): + ready.set_result(None) + + task.add_done_callback(notify) + put_task.add_done_callback(notify) + try: + await ready + finally: + # The worker may outlive this enqueue; the put task cannot. + _ = task.remove_done_callback(notify) + def _discard_pending(self) -> None: while not self._queue.empty(): _ = self._queue.get_nowait() self._queue.task_done() async def _run(self) -> None: - pending: PostgresFetchJob[FallbackT] | _StopWorker | None = None + item = await self._queue.get() while True: - item = pending if pending is not None else await self._queue.get() - pending = None batch: list[PostgresFetchJob[FallbackT]] = [] try: if isinstance(item, _StopWorker): return batch.append(item) - pending = await self._collect_batch(batch) + next_item = await self._collect_batch(batch) await self._fetch_and_deliver(batch) finally: self._queue.task_done() for _job in batch[1:]: self._queue.task_done() + item = next_item if next_item is not None else await self._queue.get() async def _collect_batch( self, @@ -244,11 +256,8 @@ async def _next_before( self, deadline: float, ) -> PostgresFetchJob[FallbackT] | _StopWorker | None: - remaining = deadline - asyncio.get_running_loop().time() - if remaining <= 0: - return None try: - async with asyncio.timeout(remaining): + async with asyncio.timeout_at(deadline): return await self._queue.get() except TimeoutError: return None @@ -295,7 +304,8 @@ async def _fetch_and_deliver( return if len(result) != len(jobs): raise RuntimeError(_INVALID_RESULT_COUNT) - for job, record in zip(jobs, result, strict=True): + for index, job in enumerate(jobs): + record = result[index] await self._deliver(PostgresFetchOutcome(job=job, record=record)) async def _deliver_failure( diff --git a/src/volcano_sdk/connection_string.py b/src/volcano_sdk/connection_string.py index b69ad544..e2764767 100644 --- a/src/volcano_sdk/connection_string.py +++ b/src/volcano_sdk/connection_string.py @@ -1,7 +1,7 @@ """Postgres connection helpers for Volcano functions.""" import re -from urllib.parse import quote, unquote +from urllib.parse import quote, unquote, urlencode _FULL_ACCESS_APP_NAME = "volcano_full_access" _USER_ACCESS_APP_NAME = "volcano_user_access" @@ -40,9 +40,11 @@ def database_connection_string( target, query = _connection_parts(base_connection_string) parameters = _query_parameters(query) - application_name = quote(_database_application_name(user_id), safe="") - parameters.append(f"application_name={application_name}") - return f"{target}?{'&'.join(parameters)}" + application = urlencode( + {"application_name": _database_application_name(user_id)}, quote_via=quote + ) + separator = "&" if parameters else "" + return f"{target}?{parameters}{separator}{application}" def _connection_parts(value: str) -> tuple[str, str]: @@ -50,32 +52,23 @@ def _connection_parts(value: str) -> tuple[str, str]: if prefix is None or _INVALID_PERCENT_ENCODING.search(value): raise ValueError(_INVALID_ERROR) - authority_end = value.find("/", prefix.end()) - possible_userinfo_end = value.find("@", prefix.end()) - userinfo_end = ( - possible_userinfo_end - if possible_userinfo_end != -1 - and (authority_end == -1 or possible_userinfo_end < authority_end) - else -1 + authority, _, _ = value[prefix.end() :].partition("/") + userinfo, separator, _ = authority.partition("@") + search_from = ( + prefix.end() + len(userinfo) + len(separator) if separator else prefix.end() ) - query_start = value.find("?", max(prefix.end(), userinfo_end + 1)) - if query_start == -1: + before_query, marker, query = value[search_from:].partition("?") + if not marker: return value, "" - return value[:query_start], value[query_start + 1 :] + return value[: search_from + len(before_query)], query -def _query_parameters(query: str) -> list[str]: - if not query: - return [] - parameters = [ +def _query_parameters(query: str) -> str: + return "&".join( parameter for parameter in query.split("&") if unquote(parameter.partition("=")[0]) != "application_name" - ] - for index in range(len(parameters) - 1, -1, -1): - if parameters[index]: - return parameters[: index + 1] - return [] + ).rstrip("&") def _database_application_name(user_id: str | None) -> str: diff --git a/tests/unit/test_client_session_boundaries.py b/tests/unit/test_client_session_boundaries.py index 41bf0808..77dd4406 100644 --- a/tests/unit/test_client_session_boundaries.py +++ b/tests/unit/test_client_session_boundaries.py @@ -96,6 +96,7 @@ def test_callback_dispatch_state_has_boolean_ownership_and_empty_failure() -> No assert client._dispatching_auth_notifications is False +@pytest.mark.order(0) def test_unsubscribe_releases_callback_ownership() -> None: client = VolcanoClient(anon_key="anon") diff --git a/tests/unit/test_connection_string.py b/tests/unit/test_connection_string.py index 3ad56efc..adc78878 100644 --- a/tests/unit/test_connection_string.py +++ b/tests/unit/test_connection_string.py @@ -27,6 +27,18 @@ def test_database_connection_string_preserves_unrelated_query_encoding() -> None ) +def test_database_connection_string_keeps_multiple_unrelated_parameters() -> None: + base = ( + "postgresql://db.example.com/app?" + "sslmode=require&connect_timeout=5&application_name=old" + ) + + assert database_connection_string(base) == ( + "postgresql://db.example.com/app?" + "sslmode=require&connect_timeout=5&application_name=volcano_full_access" + ) + + def test_database_connection_string_selects_encoded_user_access() -> None: assert database_connection_string( "postgres://user:password@db.example.com/app", @@ -77,6 +89,27 @@ def test_database_connection_string_drops_trailing_query_separator() -> None: ) +def test_database_connection_string_preserves_trailing_x_in_query_value() -> None: + assert database_connection_string("postgresql://host/db?label=X") == ( + "postgresql://host/db?label=X&application_name=volcano_full_access" + ) + + +def test_database_connection_string_drops_empty_query_after_replaced_name() -> None: + assert ( + database_connection_string( + "postgresql://host/db?label=one&&application_name=old&" + ) + == "postgresql://host/db?label=one&application_name=volcano_full_access" + ) + + +def test_database_connection_string_replaces_name_without_a_path() -> None: + assert database_connection_string("postgres://host?application_name=old") == ( + "postgres://host?application_name=volcano_full_access" + ) + + def test_database_connection_string_drops_repeated_trailing_separators() -> None: assert database_connection_string("postgresql://host/db?sslmode=require&&&") == ( "postgresql://host/db?sslmode=require&application_name=volcano_full_access" diff --git a/tests/unit/test_database_refresh.py b/tests/unit/test_database_refresh.py index d409bad0..544cbb0c 100644 --- a/tests/unit/test_database_refresh.py +++ b/tests/unit/test_database_refresh.py @@ -356,12 +356,16 @@ def on_auth_change(event: str, _session: Session | None) -> None: def test_refresh_listener_can_wait_for_another_refresh_thread() -> None: completed: list[bool] = [] workers: list[Thread] = [] + notifications: list[str] = [] client = make_client(handle_expired_read) def on_refresh(event: str, _session: Session | None) -> None: if event != "TOKEN_REFRESHED": return + notifications.append(event) + if len(notifications) > 1: + return subscription.unsubscribe() worker = Thread(target=client.auth.refresh_session) workers.append(worker) @@ -374,6 +378,7 @@ def on_refresh(event: str, _session: Session | None) -> None: for worker in workers: worker.join(timeout=5) assert completed == [True] + assert notifications == ["TOKEN_REFRESHED"] @pytest.mark.parametrize("replace_at", ["replay", "failed-refresh-listener"]) diff --git a/tests/unit/test_mutation_results.py b/tests/unit/test_mutation_results.py index 626a5c08..4236be56 100644 --- a/tests/unit/test_mutation_results.py +++ b/tests/unit/test_mutation_results.py @@ -38,6 +38,7 @@ def test_scoped_mutation_excludes_prefix_sibling_module( esac """, "mutmut": "#!/bin/sh\nprintf '%s\\n' \"$@\" > mutation-args.txt\n", + "pyrefly": "#!/bin/sh\nexit 0\n", "python": "#!/bin/sh\nexit 0\n", } for name, content in stubs.items(): @@ -45,6 +46,8 @@ def test_scoped_mutation_excludes_prefix_sibling_module( _ = stub.write_text(content, encoding="utf-8") stub.chmod(0o755) monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}") + for name in ("MUTATION_FULL", "MUTATION_SHARD_INDEX", "MUTATION_SHARD_COUNT"): + monkeypatch.delenv(name, raising=False) result = subprocess.run( ["/bin/bash", "scripts/mutation.sh"], diff --git a/tests/unit/test_mutation_shards.py b/tests/unit/test_mutation_shards.py new file mode 100644 index 00000000..21a2d0b4 --- /dev/null +++ b/tests/unit/test_mutation_shards.py @@ -0,0 +1,81 @@ +"""A full mutation gate must account for every tracked runtime module.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest + +from scripts.check_mutation_shards import MutationShardError, main + +if TYPE_CHECKING: + from pathlib import Path + + +def write_report( + root: Path, index: int, module: str, *, outcomes: dict[str, int] | None = None +) -> None: + """Create one native-result-shaped shard fixture.""" + report = root / f"mutation-outcomes-{index}" / "mutation.json" + report.parent.mkdir(parents=True) + _ = report.write_text( + json.dumps( + { + "modules": [module], + "outcomes": outcomes or {"killed": 1}, + "unmutatable_modules": [], + "failures": [], + } + ), + encoding="utf-8", + ) + + +def test_full_mutation_shards_cover_each_runtime_module_once(tmp_path: Path) -> None: + source = tmp_path / "source.bin" + _ = source.write_bytes( + b"src/volcano_sdk/auth.py\0src/volcano_sdk/realtime.py\0" + b"src/volcano_sdk/_generated/client.py\0" + ) + write_report(tmp_path, 0, "src/volcano_sdk/auth.py") + write_report(tmp_path, 1, "src/volcano_sdk/realtime.py") + + assert main(tmp_path, source, 2) == 0 + + +@pytest.mark.parametrize( + ("failure", "message"), + [ + ("missing", "mutation.json"), + ("uncovered", "Failed mutation outcomes"), + ("survived", "Failed mutation outcomes"), + ("timed_out", "Failed mutation outcomes"), + ("zero", "Failed mutation outcomes"), + ("duplicate", "inventory differs"), + ("omitted", "inventory differs"), + ], +) +def test_incomplete_or_failed_mutation_shards_fail( + tmp_path: Path, failure: str, message: str +) -> None: + source = tmp_path / "source.bin" + _ = source.write_bytes(b"src/volcano_sdk/auth.py\0src/volcano_sdk/realtime.py\0") + write_report(tmp_path, 0, "src/volcano_sdk/auth.py") + if failure not in {"missing", "omitted"}: + module = ( + "src/volcano_sdk/auth.py" + if failure == "duplicate" + else "src/volcano_sdk/realtime.py" + ) + outcomes = None + if failure == "zero": + outcomes = {"killed": 0} + elif failure in {"uncovered", "survived", "timed_out"}: + outcomes = {failure: 1} + write_report(tmp_path, 1, module, outcomes=outcomes) + elif failure == "omitted": + write_report(tmp_path, 1, "src/volcano_sdk/other.py") + + with pytest.raises((MutationShardError, FileNotFoundError), match=message): + _ = main(tmp_path, source, 2) diff --git a/tests/unit/test_realtime_delivery_boundaries.py b/tests/unit/test_realtime_delivery_boundaries.py index 098de980..612e5e13 100644 --- a/tests/unit/test_realtime_delivery_boundaries.py +++ b/tests/unit/test_realtime_delivery_boundaries.py @@ -391,7 +391,7 @@ async def test_closed_postgres_worker_reports_only_current_delivery_failures( enqueue = worker.enqueue async def stop_before_enqueue(job: PostgresFetchJob[_PostgresDelivery]) -> None: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) if unsubscribe: await channel.unsubscribe() await enqueue(job) diff --git a/tests/unit/test_realtime_fetch_lifecycle.py b/tests/unit/test_realtime_fetch_lifecycle.py index 9f06716f..a27ea584 100644 --- a/tests/unit/test_realtime_fetch_lifecycle.py +++ b/tests/unit/test_realtime_fetch_lifecycle.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import gc +import weakref from typing import TYPE_CHECKING import pytest @@ -9,9 +11,17 @@ OutcomeRecorder, RecordingBatchFetch, fetch_job, + passthrough_job, +) +from typing_extensions import override + +from volcano_sdk._realtime_fetch_worker import ( + PostgresFetchJob, + PostgresFetchOutcome, + PostgresFetchWorker, + _StopWorker, + _wait_for_close, ) - -from volcano_sdk._realtime_fetch_worker import PostgresFetchOutcome, PostgresFetchWorker if TYPE_CHECKING: from volcano_sdk.realtime import _PostgresFetchRequest @@ -23,6 +33,73 @@ async def cancel_operation(task: asyncio.Task[None] | None) -> None: _ = await asyncio.gather(task, return_exceptions=True) +class CancellationAwareQueue(asyncio.Queue[PostgresFetchJob[str] | _StopWorker]): + def __init__(self) -> None: + super().__init__(maxsize=1) + self.cancelled = asyncio.Event() + self.release = asyncio.Event() + self.pending_put: asyncio.Task[None] | None = None + + @override + async def put(self, item: PostgresFetchJob[str] | _StopWorker) -> None: + self.pending_put = asyncio.current_task() + try: + await super().put(item) + except asyncio.CancelledError: + self.cancelled.set() + _ = await self.release.wait() + raise + + +@pytest.mark.order(0) +async def test_successful_batch_delivers_each_result() -> None: + fetch = RecordingBatchFetch() + deliver = OutcomeRecorder() + worker = PostgresFetchWorker(fetch, deliver, queue_limit=2) + jobs = [fetch_job(1), fetch_job(2)] + + await asyncio.wait_for(worker._fetch_and_deliver(jobs), timeout=1) + + assert fetch.calls == [(1, 2)] + assert deliver.items == [ + PostgresFetchOutcome(job=jobs[0], record={"id": 1}), + PostgresFetchOutcome(job=jobs[1], record={"id": 2}), + ] + + +async def test_failed_close_waits_for_stop_task_cleanup() -> None: + failure = RuntimeError("worker failed") + cancellation_seen = asyncio.Event() + cleanup_release = asyncio.Event() + + async def fail() -> None: + raise failure + + async def stop() -> None: + try: + _ = await asyncio.Event().wait() + except asyncio.CancelledError: + cancellation_seen.set() + _ = await cleanup_release.wait() + raise + + worker_task = asyncio.create_task(fail()) + stop_task = asyncio.create_task(stop()) + waiter = asyncio.create_task(_wait_for_close(worker_task, stop_task)) + try: + _ = await asyncio.wait_for(cancellation_seen.wait(), timeout=1) + assert not waiter.done() + cleanup_release.set() + with pytest.raises(RuntimeError, match="worker failed") as caught: + await asyncio.wait_for(waiter, timeout=1) + assert caught.value is failure + assert stop_task.done() + finally: + cleanup_release.set() + await cancel_operation(waiter) + await cancel_operation(stop_task) + + @pytest.mark.parametrize( ("window", "size", "message"), [ @@ -50,16 +127,182 @@ async def test_unused_worker_can_close_and_abort_repeatedly() -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) await worker.close() + assert worker._stop_task is None + assert worker._queue.empty() await worker.close() - await worker.abort() - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) + await asyncio.wait_for(worker.abort(), timeout=1) with pytest.raises(RuntimeError, match="fetch worker is closed"): - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) assert fetch.calls == [] assert deliver.items == [] +async def test_closed_worker_rejects_new_jobs_before_abort() -> None: + worker = PostgresFetchWorker( + RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 + ) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.close(), timeout=1) + + with pytest.raises(RuntimeError, match="fetch worker is closed"): + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + + +async def test_abort_cancels_an_active_fetch() -> None: + fetch = BlockingRowFetch() + worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) + aborting: asyncio.Task[None] | None = None + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) + aborting = asyncio.create_task(worker.abort()) + _ = await asyncio.wait_for(fetch.cancelled.wait(), timeout=0.5) + await asyncio.wait_for(aborting, timeout=1) + finally: + fetch.release.set() + await cancel_operation(aborting) + await asyncio.wait_for(worker.abort(), timeout=1) + + +async def test_cancelled_enqueue_waits_for_its_queue_put_to_finish() -> None: + fetch = BlockingRowFetch() + worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) + queue = CancellationAwareQueue() + worker._queue = queue + enqueueing: asyncio.Task[None] | None = None + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + enqueueing = asyncio.create_task(worker.enqueue(fetch_job(3))) + await asyncio.sleep(0) + assert not enqueueing.done() + + _ = enqueueing.cancel() + _ = await asyncio.wait_for(queue.cancelled.wait(), timeout=0.5) + assert not enqueueing.done() + queue.release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(enqueueing, timeout=1) + finally: + queue.release.set() + await cancel_operation(queue.pending_put) + await cancel_operation(enqueueing) + await asyncio.wait_for(worker.abort(), timeout=1) + + +async def test_enqueue_rejects_cancellation_after_queue_put_completes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + worker = PostgresFetchWorker( + RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 + ) + + async def wait_forever() -> None: + _ = await asyncio.Event().wait() + + task = asyncio.create_task(wait_forever()) + + async def cancel_after_put( + worker_task: asyncio.Task[None], put_task: asyncio.Task[None] + ) -> None: + await put_task + _ = worker_task.cancel() + _ = await asyncio.gather(worker_task, return_exceptions=True) + + monkeypatch.setattr(worker, "_await_put_or_worker", cancel_after_put) + try: + with pytest.raises(RuntimeError, match="fetch worker is closed"): + await worker._put_while_running(fetch_job(1), task) + finally: + await cancel_operation(task) + + +async def test_enqueues_release_completion_callbacks( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fetch = BlockingRowFetch() + worker = PostgresFetchWorker(fetch, OutcomeRecorder(), queue_limit=1) + loop = asyncio.get_running_loop() + create_future = loop.create_future + created: list[weakref.ReferenceType[asyncio.Future[None]]] = [] + + def track_future() -> asyncio.Future[None]: + future: asyncio.Future[None] = create_future() + created.append(weakref.ref(future)) + return future + + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) + with monkeypatch.context() as patch: + patch.setattr(loop, "create_future", track_future) + async with asyncio.timeout(1): + await worker.enqueue(fetch_job(2)) + + await asyncio.sleep(0) + _ = gc.collect() + assert len(created) == 1 + assert created[0]() is None + finally: + await asyncio.wait_for(worker.abort(), timeout=1) + + +@pytest.mark.order(0) +async def test_race_helper_handles_two_already_completed_tasks() -> None: + async def complete() -> None: + return + + task = asyncio.create_task(complete()) + put_task = asyncio.create_task(complete()) + _ = await asyncio.gather(task, put_task) + + await asyncio.wait_for( + PostgresFetchWorker._await_put_or_worker(task, put_task), timeout=1 + ) + + +async def test_first_enqueue_finishes_before_worker_shutdown() -> None: + worker = PostgresFetchWorker( + RecordingBatchFetch(), OutcomeRecorder(), queue_limit=1 + ) + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=0.5) + assert worker._task is not None + assert not worker._task.done() + finally: + await asyncio.wait_for(worker.abort(), timeout=1) + + +async def test_known_worker_failure_does_not_schedule_a_stop_request() -> None: + failure = RuntimeError("delivery failed") + delivery_started = asyncio.Event() + release_delivery = asyncio.Event() + + async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: + delivery_started.set() + _ = await release_delivery.wait() + raise failure + + worker = PostgresFetchWorker(RecordingBatchFetch(), fail_delivery, queue_limit=1) + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + _ = await asyncio.wait_for(delivery_started.wait(), timeout=1) + release_delivery.set() + assert worker._task is not None + _ = await asyncio.gather(worker._task, return_exceptions=True) + + with pytest.raises(RuntimeError, match="delivery failed") as caught: + await asyncio.wait_for(worker.close(), timeout=1) + assert caught.value is failure + assert worker._stop_task is None + finally: + release_delivery.set() + await asyncio.wait_for(worker.abort(), timeout=1) + + async def test_close_failure_cancels_a_blocked_stop_request() -> None: fetch = BlockingRowFetch() failure = RuntimeError("delivery failed") @@ -70,9 +313,9 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: worker = PostgresFetchWorker(fetch, fail_delivery, queue_limit=1) closing = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) closing = asyncio.create_task(worker.close()) await asyncio.sleep(0) stop_task = worker._stop_task @@ -87,7 +330,7 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: assert stop_task.cancelled() assert fetch.row_ids == [1] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) await cancel_operation(closing) @@ -97,9 +340,9 @@ async def test_abort_unblocks_close_with_a_full_queue() -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) closing = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) closing = asyncio.create_task(worker.close()) await asyncio.sleep(0) stop_task = worker._stop_task @@ -115,7 +358,7 @@ async def test_abort_unblocks_close_with_a_full_queue() -> None: assert fetch.cancelled.is_set() assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) await cancel_operation(closing) @@ -125,9 +368,9 @@ async def test_abort_rejects_an_enqueue_waiting_for_capacity() -> None: worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) enqueueing = None try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=1) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) enqueueing = asyncio.create_task(worker.enqueue(fetch_job(3))) await asyncio.sleep(0) assert not enqueueing.done() @@ -139,7 +382,7 @@ async def test_abort_rejects_an_enqueue_waiting_for_capacity() -> None: assert fetch.cancelled.is_set() assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) await cancel_operation(enqueueing) @@ -155,13 +398,13 @@ async def no_results( deliver = OutcomeRecorder() worker = PostgresFetchWorker(no_results, deliver, queue_limit=1) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) release.set() with pytest.raises(RuntimeError, match="unexpected result count"): await asyncio.wait_for(worker.close(), timeout=1) assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_fetch_cancellation_propagates_without_delivering_a_fallback() -> None: @@ -178,7 +421,7 @@ async def cancelled_fetch( deliver = OutcomeRecorder() worker = PostgresFetchWorker(cancelled_fetch, deliver, queue_limit=1) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(started.wait(), timeout=1) release.set() @@ -187,7 +430,7 @@ async def cancelled_fetch( assert deliver.items == [] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) async def test_batch_window_flushes_without_waiting_for_another_row_or_close() -> None: @@ -202,13 +445,83 @@ async def deliver(outcome: PostgresFetchOutcome[str]) -> None: fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=0.1 ) try: - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(delivered.wait(), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) assert fetch.calls == [(1,)] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) + + +async def test_default_batch_window_flushes_before_another_row_arrives() -> None: + fetch = RecordingBatchFetch() + delivered = asyncio.Event() + + async def deliver(_outcome: PostgresFetchOutcome[str]) -> None: + delivered.set() + + worker = PostgresFetchWorker(fetch, deliver, queue_limit=2, max_batch_size=2) + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + _ = await asyncio.wait_for(delivered.wait(), timeout=0.5) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + await asyncio.wait_for(worker.close(), timeout=1) + + assert fetch.calls == [(1,), (2,)] + finally: + await asyncio.wait_for(worker.abort(), timeout=1) + + +async def test_full_batch_flushes_before_close() -> None: + fetch = RecordingBatchFetch() + delivered = asyncio.Event() + outcomes: list[PostgresFetchOutcome[str]] = [] + + async def deliver(outcome: PostgresFetchOutcome[str]) -> None: + outcomes.append(outcome) + if len(outcomes) == 2: + delivered.set() + + worker = PostgresFetchWorker( + fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=60 + ) + try: + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + _ = await asyncio.wait_for(delivered.wait(), timeout=1) + await asyncio.wait_for(worker.close(), timeout=1) + await asyncio.wait_for(worker._queue.join(), timeout=1) + + assert fetch.calls == [(1, 2)] + assert outcomes == [ + PostgresFetchOutcome(job=fetch_job(1), record={"id": 1}), + PostgresFetchOutcome(job=fetch_job(2), record={"id": 2}), + ] + finally: + await asyncio.wait_for(worker.abort(), timeout=1) + + +async def test_passthrough_does_not_enter_a_fetch_batch() -> None: + fetch = RecordingBatchFetch() + deliver = OutcomeRecorder() + worker = PostgresFetchWorker( + fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=0.01 + ) + try: + await asyncio.wait_for( + worker.enqueue(passthrough_job("full-payload")), timeout=1 + ) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.close(), timeout=1) + + assert fetch.calls == [(1,)] + assert deliver.items == [ + PostgresFetchOutcome(job=passthrough_job("full-payload")), + PostgresFetchOutcome(job=fetch_job(1), record={"id": 1}), + ] + finally: + await asyncio.wait_for(worker.abort(), timeout=1) @pytest.mark.parametrize(("window", "expected"), [(0, [(1,), (2,)]), (60, [(1, 2)])]) @@ -221,12 +534,12 @@ async def test_batch_capacity_and_zero_window_preserve_delivery_order( fetch, deliver, queue_limit=2, max_batch_size=2, batch_window_seconds=window ) try: - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.close(), timeout=1) - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) assert fetch.calls == expected assert [item.record for item in deliver.items] == [{"id": 1}, {"id": 2}] finally: - await worker.abort() + await asyncio.wait_for(worker.abort(), timeout=1) diff --git a/tests/unit/test_realtime_fetch_worker.py b/tests/unit/test_realtime_fetch_worker.py index a7effb92..4e3f66d0 100644 --- a/tests/unit/test_realtime_fetch_worker.py +++ b/tests/unit/test_realtime_fetch_worker.py @@ -103,9 +103,9 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) third_enqueue = asyncio.create_task(worker.enqueue(fetch_job(3))) await asyncio.sleep(0) @@ -143,9 +143,9 @@ async def scenario() -> None: batch_window_seconds=0.05, max_batch_size=3, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) - await worker.enqueue(fetch_job(3)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(3)), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert fetch.calls == [(1, 2, 3)] @@ -176,8 +176,8 @@ async def scenario() -> None: batch_window_seconds=0.01, max_batch_size=2, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2, table="archive")) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2, table="archive")), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert fetch.calls == [(1,), (2,)] @@ -202,8 +202,8 @@ async def scenario() -> None: batch_window_seconds=0.01, max_batch_size=2, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert fetch.calls == [(1,), (1,)] @@ -219,9 +219,11 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(passthrough_job("full-payload")) + await asyncio.wait_for( + worker.enqueue(passthrough_job("full-payload")), timeout=1 + ) assert outcomes == [] @@ -237,6 +239,7 @@ async def scenario() -> None: asyncio.run(scenario()) +@pytest.mark.order(0) def test_postgres_fetch_worker_aborts_obsolete_jobs_without_waiting() -> None: async def scenario() -> None: fetch = BlockingRowFetch() @@ -244,9 +247,9 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.abort(), timeout=0.2) @@ -274,8 +277,8 @@ async def scenario() -> None: deliver, queue_limit=2, ) - await worker.enqueue(fetch_job(1)) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) await asyncio.wait_for(worker.close(), timeout=0.2) assert outcomes == [ @@ -302,13 +305,13 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: fail_delivery, queue_limit=1, ) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) _ = await delivery_started.wait() release_delivery.set() await asyncio.sleep(0) with pytest.raises(RuntimeError) as raised: - await worker.enqueue(fetch_job(3)) + await asyncio.wait_for(worker.enqueue(fetch_job(3)), timeout=1) assert raised.value is failure @@ -322,9 +325,9 @@ async def scenario() -> None: outcomes = deliver.items worker = PostgresFetchWorker(fetch, deliver, queue_limit=1) - await worker.enqueue(fetch_job(1)) + await asyncio.wait_for(worker.enqueue(fetch_job(1)), timeout=1) _ = await asyncio.wait_for(fetch.started.wait(), timeout=0.2) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) interrupted_close = asyncio.create_task(worker.close()) await asyncio.sleep(0) @@ -362,9 +365,9 @@ async def fail_delivery(_outcome: PostgresFetchOutcome[str]) -> None: fail_delivery, queue_limit=1, ) - await worker.enqueue(fetch_job(2)) + await asyncio.wait_for(worker.enqueue(fetch_job(2)), timeout=1) _ = await delivery_started.wait() - await worker.enqueue(fetch_job(3)) + await asyncio.wait_for(worker.enqueue(fetch_job(3)), timeout=1) blocked_enqueue = asyncio.create_task(worker.enqueue(fetch_job(4))) await asyncio.sleep(0) assert not blocked_enqueue.done()