diff --git a/clients/python/tests/conftest.py b/clients/python/tests/conftest.py index 2d777a25..1ba6ff42 100644 --- a/clients/python/tests/conftest.py +++ b/clients/python/tests/conftest.py @@ -1,7 +1,13 @@ +from collections.abc import Callable, Generator from datetime import UTC, datetime +from typing import Any +import pytest +import sentry_sdk import time_machine from arroyo.backends.kafka import KafkaProducer +from sentry_sdk.envelope import Envelope +from sentry_sdk.transport import Transport from taskbroker_client.types import AtMostOnceStore @@ -21,6 +27,32 @@ def freeze_time(t: str | datetime | None = None) -> time_machine.travel: return time_machine.travel(t, tick=False) +@pytest.fixture +def sentry_init() -> Generator[Callable[..., None], None, None]: + clients = [] + + def inner(*a: Any, **kw: Any) -> None: + kw.setdefault("transport", TestTransport()) + client = sentry_sdk.Client(*a, **kw) + clients.append(client) + sentry_sdk.get_global_scope().set_client(client) + + old_client = sentry_sdk.get_global_scope().client + try: + sentry_sdk.get_current_scope().set_client(None) + yield inner + finally: + for client in clients: + client.close() + sentry_sdk.get_global_scope().set_client(old_client) + + +class TestTransport(Transport): + def capture_envelope(self, _: Envelope) -> None: + """No-op capture_envelope for tests""" + pass + + class StubAtMostOnce(AtMostOnceStore): def __init__(self) -> None: self._keys: dict[str, str] = {} diff --git a/clients/python/tests/test_task.py b/clients/python/tests/test_task.py index 2556b9e7..c9e8501e 100644 --- a/clients/python/tests/test_task.py +++ b/clients/python/tests/test_task.py @@ -2,7 +2,7 @@ import datetime from collections.abc import MutableMapping from concurrent.futures import Future -from typing import Any +from typing import Any, Callable from unittest.mock import patch import msgpack @@ -297,7 +297,11 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert activation.parameters == "" -def test_create_activation_tracing(task_namespace: TaskNamespace) -> None: +def test_create_activation_tracing( + sentry_init: Callable[..., None], task_namespace: TaskNamespace +) -> None: + sentry_init(traces_sample_rate=1.0) + @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError @@ -310,7 +314,11 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert "baggage" in headers -def test_create_activation_tracing_headers(task_namespace: TaskNamespace) -> None: +def test_create_activation_tracing_headers( + sentry_init: Callable[..., None], task_namespace: TaskNamespace +) -> None: + sentry_init(traces_sample_rate=1.0) + @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError @@ -326,7 +334,11 @@ def with_parameters(one: str, two: int, org_id: int) -> None: assert headers["key"] == "value" -def test_create_activation_tracing_disable(task_namespace: TaskNamespace) -> None: +def test_create_activation_tracing_disable( + sentry_init: Callable[..., None], task_namespace: TaskNamespace +) -> None: + sentry_init(traces_sample_rate=1.0) + @task_namespace.register(name="test.parameters") def with_parameters(one: str, two: int, org_id: int) -> None: raise NotImplementedError diff --git a/clients/python/tests/worker/test_worker.py b/clients/python/tests/worker/test_worker.py index 7253a74e..858ad939 100644 --- a/clients/python/tests/worker/test_worker.py +++ b/clients/python/tests/worker/test_worker.py @@ -1022,7 +1022,12 @@ def test_push_task_worker_busy(self) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None: +def test_child_process_complete( + mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1047,7 +1052,11 @@ def test_child_process_complete(mock_capture_checkin: mock.MagicMock) -> None: assert mock_capture_checkin.call_count == 0 -def test_child_process_canary_task(capsys: pytest.CaptureFixture[str]) -> None: +def test_child_process_canary_task( + sentry_init: Callable[..., None], capsys: pytest.CaptureFixture[str] +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1073,7 +1082,9 @@ def test_child_process_canary_task(capsys: pytest.CaptureFixture[str]) -> None: assert capsys.readouterr().out == "Done running canary task!\n" -def test_child_process_emits_running_message() -> None: +def test_child_process_emits_running_message(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1107,7 +1118,10 @@ def test_child_process_emits_running_message() -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") def test_child_process_emits_exiting_once_and_continues_until_release( mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], ) -> None: + sentry_init(traces_sample_rate=1.0) + shutdown = Event() ctx = get_context("fork") child_id = uuid4() @@ -1167,7 +1181,9 @@ def test_child_process_emits_exiting_once_and_continues_until_release( assert mock_capture_checkin.call_count == 0 -def test_child_process_emits_busy_and_idle_messages() -> None: +def test_child_process_emits_busy_and_idle_messages(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1199,7 +1215,9 @@ def test_child_process_emits_busy_and_idle_messages() -> None: assert processed.get(timeout=1).task_id == SIMPLE_TASK.activation.id -def test_child_process_remove_start_time_kwargs() -> None: +def test_child_process_remove_start_time_kwargs(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + activation = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -1236,7 +1254,9 @@ def test_child_process_remove_start_time_kwargs() -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_retry_task() -> None: +def test_child_process_retry_task(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1263,8 +1283,12 @@ def test_child_process_retry_task() -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_retry_task_max_attempts( - mock_capture: mock.Mock, mock_logger: mock.Mock + mock_capture: mock.Mock, + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: + sentry_init(traces_sample_rate=1.0) + # Create an activation that is on its final attempt and # will raise an error again. activation = InflightTaskActivation( @@ -1324,7 +1348,9 @@ def test_child_process_retry_task_max_attempts( assert extra["retry_max_attempts"] == 3 -def test_child_process_failure_task() -> None: +def test_child_process_failure_task(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1348,7 +1374,9 @@ def test_child_process_failure_task() -> None: assert result.status == TASK_ACTIVATION_STATUS_FAILURE -def test_child_process_shutdown() -> None: +def test_child_process_shutdown(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1372,7 +1400,9 @@ def test_child_process_shutdown() -> None: assert processed.qsize() == 0 -def test_child_process_unknown_task() -> None: +def test_child_process_unknown_task(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1400,7 +1430,9 @@ def test_child_process_unknown_task() -> None: assert result.status == TASK_ACTIVATION_STATUS_COMPLETE -def test_child_process_at_most_once() -> None: +def test_child_process_at_most_once(sentry_init: Callable[..., None]) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1431,7 +1463,12 @@ def test_child_process_at_most_once() -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_record_checkin(mock_capture_checkin: mock.Mock) -> None: +def test_child_process_record_checkin( + mock_capture_checkin: mock.Mock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1463,8 +1500,10 @@ def test_child_process_record_checkin(mock_capture_checkin: mock.Mock) -> None: ) -def test_child_process_pass_headers() -> None: +def test_child_process_pass_headers(sentry_init: Callable[..., None]) -> None: """Task with pass_headers=True receives headers from the activation.""" + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1494,7 +1533,12 @@ def test_child_process_pass_headers() -> None: @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_terminate_task(mock_logger: mock.Mock) -> None: +def test_child_process_terminate_task( + mock_logger: mock.Mock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1541,7 +1585,11 @@ def test_child_process_terminate_task(mock_logger: mock.Mock) -> None: @mock.patch("taskbroker_client.worker.workerchild.capture_checkin") -def test_child_process_decompression(mock_capture_checkin: mock.MagicMock) -> None: +def test_child_process_decompression( + mock_capture_checkin: mock.MagicMock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1567,8 +1615,10 @@ def test_child_process_decompression(mock_capture_checkin: mock.MagicMock) -> No assert mock_capture_checkin.call_count == 0 -def test_child_process_context_hooks() -> None: +def test_child_process_context_hooks(sentry_init: Callable[..., None]) -> None: """Context hooks' on_execute is called with activation headers during task execution.""" + sentry_init(traces_sample_rate=1.0) + executed_headers: list[dict[str, str]] = [] class RecordingHook: @@ -1625,7 +1675,12 @@ def on_execute(self, headers: dict[str, str]) -> contextlib.AbstractContextManag @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_silenced_timeout(mock_logger: mock.Mock) -> None: +def test_child_process_silenced_timeout( + mock_logger: mock.Mock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1656,7 +1711,12 @@ def test_child_process_silenced_timeout(mock_logger: mock.Mock) -> None: @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") -def test_child_process_silenced_exception_with_retries(mock_capture: mock.Mock) -> None: +def test_child_process_silenced_exception_with_retries( + mock_capture: mock.Mock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1684,7 +1744,12 @@ def test_child_process_silenced_exception_with_retries(mock_capture: mock.Mock) @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") -def test_child_process_expected_ignored_exception_max_attempts(mock_capture: mock.Mock) -> None: +def test_child_process_expected_ignored_exception_max_attempts( + mock_capture: mock.Mock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1714,9 +1779,13 @@ def test_child_process_expected_ignored_exception_max_attempts(mock_capture: moc @mock.patch("taskbroker_client.worker.workerchild.logger") @mock.patch("taskbroker_client.worker.workerchild.sentry_sdk.capture_exception") def test_child_process_silenced_exception_max_attempts( - mock_capture: mock.Mock, mock_logger: mock.Mock + mock_capture: mock.Mock, + mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: """Silenced exceptions do not raise on retry exhaustion.""" + sentry_init(traces_sample_rate=1.0) + activation = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -1768,7 +1837,12 @@ def test_child_process_silenced_exception_max_attempts( @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_retry_on_deadline_exceeded(mock_logger: mock.Mock) -> None: +def test_child_process_retry_on_deadline_exceeded( + mock_logger: mock.Mock, + sentry_init: Callable[..., None], +) -> None: + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1801,8 +1875,13 @@ def test_child_process_retry_on_deadline_exceeded(mock_logger: mock.Mock) -> Non @mock.patch("taskbroker_client.worker.workerchild.logger") -def test_child_process_general_exception_logs_task_failed(mock_logger: mock.Mock) -> None: +def test_child_process_general_exception_logs_task_failed( + mock_logger: mock.Mock, + sentry_init: Callable[..., None], +) -> None: """A non-retriable Exception emits taskworker.task.failed with all fields.""" + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1839,9 +1918,12 @@ def test_child_process_general_exception_logs_task_failed(mock_logger: mock.Mock @mock.patch("taskbroker_client.worker.workerchild.logger") def test_child_process_silenced_exception_does_not_log_task_failed( mock_logger: mock.Mock, + sentry_init: Callable[..., None], ) -> None: """When err is in silenced_exceptions, taskworker.task.failed is NOT logged. Preserves the silencing semantics added in #608.""" + sentry_init(traces_sample_rate=1.0) + todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() shutdown = Event() @@ -1936,10 +2018,13 @@ def _producing_task(task_id: str = "task-with-futures") -> InflightTaskActivatio @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_tracks_producer_futures( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -1974,10 +2059,13 @@ def test_child_process_tracks_producer_futures( @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_holds_result_until_futures_done( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2029,10 +2117,13 @@ def observe_and_resolve() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_skip_awaiting_futures_places_result_immediately( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2091,10 +2182,13 @@ def observe_and_resolve() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_drains_pending_futures_on_sigterm( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + task = _producing_task() todo: queue.Queue[InflightTaskActivation] = queue.Queue() processed: queue.Queue[ProcessingResult] = queue.Queue() @@ -2139,10 +2233,13 @@ def deliver_sigterm() -> None: @pytest.mark.parametrize("producer_cls", _PRODUCER_CLASSES) def test_child_process_retries_on_failed_future( + sentry_init: Callable[..., None], producer_cls: type, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + retriable_task = InflightTaskActivation( host="localhost:50051", receive_timestamp=0, @@ -2189,10 +2286,13 @@ def test_child_process_retries_on_failed_future( @pytest.mark.parametrize("pending_registry", _PENDING_REGISTRIES) def test_child_process_clears_pending_futures_when_task_fails( + sentry_init: Callable[..., None], pending_registry: Any, clear_pending_futures: None, restore_signal_handlers: None, ) -> None: + sentry_init(traces_sample_rate=1.0) + leftover_future: Future[BrokerValue[KafkaPayload]] = Future() leftover_future.set_result(_make_broker_value()) pending_registry["test.producer"].append(leftover_future) @@ -2226,9 +2326,14 @@ def test_child_process_clears_pending_futures_when_task_fails( def test_child_process_uses_configured_future_checking_frequency( - clear_pending_futures: None, restore_signal_handlers: None + sentry_init: Callable[..., None], clear_pending_futures: None, restore_signal_handlers: None ) -> None: """The idle future-checking loop polls on the configured interval.""" + sentry_init( + traces_sample_rate=1.0, + enable_backpressure_handling=False, # To avoid time.sleep which the test patches. + ) + # A task that runs long enough for the idle future-checking loop to poll a # few times before max_task_count triggers shutdown. slow_task = InflightTaskActivation( @@ -2255,6 +2360,8 @@ def recording_sleep(seconds: float) -> None: idle_sleeps.append(seconds) real_sleep(seconds) + import examples.tasks # noqa: F401; Ensure time.sleep reference is set before patching. + # time.sleep is only used by the idle branch of check_task_future_completion # inside workerchild, so every recorded call comes from that loop. The task's # own sleep uses a separate `from time import sleep` import in examples.tasks.