From d31a96892054717312c9283469be0d26684042db Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 10 Oct 2026 15:37:16 +0300 Subject: [PATCH] fix: import datetime as dt for ruff 0.17 and restore batch_bodies setter for ty 0.0.86 --- faststream_outbox/_scheduling.py | 22 ++++----- faststream_outbox/_time.py | 6 +-- faststream_outbox/broker.py | 10 ++-- faststream_outbox/client.py | 10 ++-- faststream_outbox/message.py | 12 ++--- faststream_outbox/publisher/usecase.py | 6 +-- faststream_outbox/response.py | 20 +++++--- faststream_outbox/testing.py | 38 +++++++------- tests/test_client_contract.py | 16 +++--- tests/test_fake.py | 60 +++++++++++------------ tests/test_integration.py | 34 ++++++------- tests/test_middleware_opentelemetry.py | 4 +- tests/test_middleware_prometheus.py | 4 +- tests/test_unit.py | 68 ++++++++++++++------------ 14 files changed, 161 insertions(+), 149 deletions(-) diff --git a/faststream_outbox/_scheduling.py b/faststream_outbox/_scheduling.py index 3a969af..353a366 100644 --- a/faststream_outbox/_scheduling.py +++ b/faststream_outbox/_scheduling.py @@ -11,27 +11,27 @@ use ``resolve_next_attempt_client_side`` — only the batch and fake paths do. """ -import datetime as _dt +import datetime as dt def is_future_dated( - activate_in: _dt.timedelta | None, - activate_at: _dt.datetime | None, - now: _dt.datetime, + activate_in: dt.timedelta | None, + activate_at: dt.datetime | None, + now: dt.datetime, ) -> bool: """Whether a row is genuinely future-dated (so NOTIFY is skipped — polling fires it at the gate).""" if activate_in is not None: - return activate_in > _dt.timedelta(0) + return activate_in > dt.timedelta(0) if activate_at is not None: return activate_at > now return False def resolve_next_attempt_client_side( - activate_in: _dt.timedelta | None, - activate_at: _dt.datetime | None, - now: _dt.datetime, -) -> _dt.datetime | None: + activate_in: dt.timedelta | None, + activate_at: dt.datetime | None, + now: dt.datetime, +) -> dt.datetime | None: """Resolve activate_in / activate_at to a single ``next_attempt_at`` value (client clock).""" if activate_in is not None: return now + activate_in @@ -40,8 +40,8 @@ def resolve_next_attempt_client_side( def validate_activate_args( method_name: str, - activate_in: _dt.timedelta | None, - activate_at: _dt.datetime | None, + activate_in: dt.timedelta | None, + activate_at: dt.datetime | None, ) -> None: """Mutex + tz-aware checks shared by the test fakes. Real broker delegates to ``OutboxPublishCommand``.""" if activate_in is not None and activate_at is not None: diff --git a/faststream_outbox/_time.py b/faststream_outbox/_time.py index 30d4c5f..dcee0a6 100644 --- a/faststream_outbox/_time.py +++ b/faststream_outbox/_time.py @@ -1,8 +1,8 @@ """Shared time helper — single source so the five former ``_utcnow`` copies can't drift.""" -import datetime as _dt +import datetime as dt -def utcnow() -> _dt.datetime: +def utcnow() -> dt.datetime: """Timezone-aware current UTC time.""" - return _dt.datetime.now(tz=_dt.UTC) + return dt.datetime.now(tz=dt.UTC) diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index dadf969..e72d947 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -6,7 +6,7 @@ """ import asyncio -import datetime as _dt +import datetime as dt import logging import typing import warnings @@ -381,8 +381,8 @@ async def publish( # ty: ignore[invalid-method-override] session: AsyncSession, headers: dict[str, str] | None = None, correlation_id: str | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, timer_id: str | None = None, ) -> int | None: """Insert one outbox row using *session*'s open transaction. @@ -420,8 +420,8 @@ async def publish_batch( # ty: ignore[invalid-method-override] queue: str, session: AsyncSession, headers: dict[str, str] | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, ) -> None: """Insert multiple outbox rows via *session*. Same transactional contract as ``publish``. diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 0c3341a..0e85549 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -16,7 +16,7 @@ import abc import asyncio -import datetime as _dt +import datetime as dt import uuid from typing import TYPE_CHECKING @@ -117,8 +117,8 @@ async def mark_pending_with_lease( *, delay_seconds: float, attempts_count: int, - first_attempt_at: _dt.datetime, - last_attempt_at: _dt.datetime, + first_attempt_at: dt.datetime, + last_attempt_at: dt.datetime, ) -> bool: ... @abc.abstractmethod @@ -398,8 +398,8 @@ async def mark_pending_with_lease( *, delay_seconds: float, attempts_count: int, - first_attempt_at: _dt.datetime, - last_attempt_at: _dt.datetime, + first_attempt_at: dt.datetime, + last_attempt_at: dt.datetime, ) -> bool: """Release the lease on *message_id* and reschedule it for retry, iff it still holds the lease. diff --git a/faststream_outbox/message.py b/faststream_outbox/message.py index 6cafa07..c89ad3c 100644 --- a/faststream_outbox/message.py +++ b/faststream_outbox/message.py @@ -8,7 +8,7 @@ ``OutboxMessage`` adapts the inner message to FastStream's ``StreamMessage``. """ -import datetime as _dt +import datetime as dt import logging import uuid from collections.abc import Awaitable, Callable @@ -89,11 +89,11 @@ class OutboxInnerMessage: headers: dict[str, str] | None attempts_count: int deliveries_count: int - created_at: _dt.datetime - next_attempt_at: _dt.datetime - first_attempt_at: _dt.datetime | None - last_attempt_at: _dt.datetime | None - acquired_at: _dt.datetime | None + created_at: dt.datetime + next_attempt_at: dt.datetime + first_attempt_at: dt.datetime | None + last_attempt_at: dt.datetime | None + acquired_at: dt.datetime | None acquired_token: uuid.UUID | None # P9: the originating timer_id (single-publish dedup key), so a terminally-failed diff --git a/faststream_outbox/publisher/usecase.py b/faststream_outbox/publisher/usecase.py index a43ed0f..a2fc7ed 100644 --- a/faststream_outbox/publisher/usecase.py +++ b/faststream_outbox/publisher/usecase.py @@ -8,7 +8,7 @@ handler on the same session that owns the inbound row's terminal write. """ -import datetime as _dt +import datetime as dt import typing from faststream._internal.endpoint.publisher import PublisherUsecase @@ -71,8 +71,8 @@ async def publish( # ty: ignore[invalid-method-override] session: AsyncSession, headers: dict[str, str] | None = None, correlation_id: str | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, timer_id: str | None = None, ) -> int | None: """Insert one outbox row scoped to this publisher's queue. diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py index 03a0cdc..b880b0f 100644 --- a/faststream_outbox/response.py +++ b/faststream_outbox/response.py @@ -11,8 +11,9 @@ ``from_cmd`` raises — see ``OutboxPublisher.__call__`` for the rationale. """ -import datetime as _dt +import datetime as dt import typing +from collections.abc import Sequence from faststream.response.publish_type import PublishType from faststream.response.response import BatchPublishCommand, PublishCommand, Response @@ -32,8 +33,8 @@ def _validate_publish_args( *, queue: object, session: object, - activate_in: _dt.timedelta | None, - activate_at: _dt.datetime | None, + activate_in: dt.timedelta | None, + activate_at: dt.datetime | None, ) -> None: """Fail-fast validation shared by every real outbox publish entry point. @@ -79,8 +80,8 @@ def __init__( session: AsyncSession, headers: dict[str, str] | None = None, correlation_id: str | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, timer_id: str | None = None, _publish_type: PublishType = PublishType.PUBLISH, ) -> None: @@ -125,6 +126,11 @@ def batch_bodies(self) -> tuple[typing.Any, ...]: # OpenTelemetry batch-count attribute in one place. return (self.body, *self.extra_bodies) + @batch_bodies.setter + def batch_bodies(self, value: Sequence[typing.Any]) -> None: + self.body, *rest = value or (None,) + self.extra_bodies = tuple(rest) + @classmethod def from_cmd( cls, @@ -173,8 +179,8 @@ def __init__( session: AsyncSession, headers: dict[str, str] | None = None, correlation_id: str | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, timer_id: str | None = None, ) -> None: _validate_publish_args( diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index dd40c58..e5bb5f4 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -11,7 +11,7 @@ or scheduled delivery actually waiting. ``feed()`` simulates a row insert. """ -import datetime as _dt +import datetime as dt import typing import uuid from contextlib import contextmanager @@ -47,16 +47,16 @@ class _FakeRow: headers: dict[str, str] | None attempts_count: int = 0 deliveries_count: int = 0 - created_at: _dt.datetime = field(default_factory=utcnow) - next_attempt_at: _dt.datetime = field(default_factory=utcnow) - first_attempt_at: _dt.datetime | None = None - last_attempt_at: _dt.datetime | None = None - acquired_at: _dt.datetime | None = None + created_at: dt.datetime = field(default_factory=utcnow) + next_attempt_at: dt.datetime = field(default_factory=utcnow) + first_attempt_at: dt.datetime | None = None + last_attempt_at: dt.datetime | None = None + acquired_at: dt.datetime | None = None acquired_token: uuid.UUID | None = None timer_id: str | None = None -def _claim_fake_row(row: _FakeRow, *, now: _dt.datetime, token: uuid.UUID) -> None: +def _claim_fake_row(row: _FakeRow, *, now: dt.datetime, token: uuid.UUID) -> None: """Set the lease and bump ``deliveries_count`` — the shared claim mechanics (F1-08). Both ``FakeOutboxClient.fetch`` (loop mode) and ``_sync_dispatch`` (sync mode) route @@ -99,7 +99,7 @@ def feed( queue: str, payload: bytes, headers: dict[str, str] | None = None, - next_attempt_at: _dt.datetime | None = None, + next_attempt_at: dt.datetime | None = None, timer_id: str | None = None, ) -> int | None: # P31: reject naive datetimes up front — the publish path is tz-strict, and a @@ -156,7 +156,7 @@ async def fetch( if not queues: return [] now = utcnow() - lease_cutoff = now - _dt.timedelta(seconds=max(0.0, lease_ttl_seconds)) + lease_cutoff = now - dt.timedelta(seconds=max(0.0, lease_ttl_seconds)) token = uuid.uuid4() out: list[OutboxInnerMessage] = [] eligible = sorted( @@ -226,12 +226,12 @@ async def mark_pending_with_lease( *, delay_seconds: float, attempts_count: int, - first_attempt_at: _dt.datetime, - last_attempt_at: _dt.datetime, + first_attempt_at: dt.datetime, + last_attempt_at: dt.datetime, ) -> bool: for row in self._rows: if row.id == message_id and _lease_token_matches(row.acquired_token, acquired_token): - row.next_attempt_at = utcnow() + _dt.timedelta(seconds=max(0.0, delay_seconds)) + row.next_attempt_at = utcnow() + dt.timedelta(seconds=max(0.0, delay_seconds)) row.attempts_count = attempts_count row.first_attempt_at = first_attempt_at row.last_attempt_at = last_attempt_at @@ -371,7 +371,7 @@ async def _fake_publish_one( queue: str, headers: dict[str, str] | None, correlation_id: str | None, - next_at: "_dt.datetime | None", + next_at: "dt.datetime | None", timer_id: str | None, run_loops: bool, ) -> int | None: @@ -396,7 +396,7 @@ async def _fake_publish_many( bodies: "Sequence[typing.Any]", queue: str, headers: dict[str, str] | None, - next_at: "_dt.datetime | None", + next_at: "dt.datetime | None", run_loops: bool, ) -> None: """Shared batch insert for both batch paths (P29). @@ -507,8 +507,8 @@ async def fake_publish( session: typing.Any = None, headers: dict[str, str] | None = None, correlation_id: str | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, timer_id: str | None = None, ) -> int | None: # P33: test mode is intentionally lenient about ``session`` — the fake client has @@ -546,8 +546,8 @@ async def fake_publish_batch( queue: str, session: typing.Any = None, headers: dict[str, str] | None = None, - activate_in: _dt.timedelta | None = None, - activate_at: _dt.datetime | None = None, + activate_in: dt.timedelta | None = None, + activate_at: dt.datetime | None = None, ) -> None: del session validate_activate_args("broker.publish_batch", activate_in, activate_at) @@ -604,7 +604,7 @@ def feed( payload: bytes, *, headers: dict[str, str] | None = None, - next_attempt_at: _dt.datetime | None = None, + next_attempt_at: dt.datetime | None = None, timer_id: str | None = None, ) -> int | None: """Insert a row directly into the in-memory store. Returns the row id, or None on timer_id conflict.""" diff --git a/tests/test_client_contract.py b/tests/test_client_contract.py index b0a6838..da2716f 100644 --- a/tests/test_client_contract.py +++ b/tests/test_client_contract.py @@ -25,7 +25,7 @@ authority stays a documented invariant (AGENTS.md), not an assertion here. """ -import datetime as _dt +import datetime as dt import os import uuid from collections.abc import AsyncIterator @@ -68,7 +68,7 @@ async def seed( rid = self._client.feed( queue=queue, payload=payload, - next_attempt_at=utcnow() + _dt.timedelta(seconds=next_attempt_offset), + next_attempt_at=utcnow() + dt.timedelta(seconds=next_attempt_offset), timer_id=timer_id, ) assert rid is not None @@ -76,7 +76,7 @@ async def seed( row.deliveries_count = deliveries_count if leased_token is not None: row.acquired_token = leased_token - row.acquired_at = utcnow() - _dt.timedelta(seconds=acquired_age or 0.0) + row.acquired_at = utcnow() - dt.timedelta(seconds=acquired_age or 0.0) return rid async def fetch(self, queues: list[str], *, limit: int, lease_ttl_seconds: float) -> list[OutboxInnerMessage]: @@ -101,8 +101,8 @@ async def mark_pending( *, delay_seconds: float, attempts_count: int, - first_attempt_at: _dt.datetime, - last_attempt_at: _dt.datetime, + first_attempt_at: dt.datetime, + last_attempt_at: dt.datetime, ) -> bool: return await self._client.mark_pending_with_lease( None, @@ -200,8 +200,8 @@ async def mark_pending( *, delay_seconds: float, attempts_count: int, - first_attempt_at: _dt.datetime, - last_attempt_at: _dt.datetime, + first_attempt_at: dt.datetime, + last_attempt_at: dt.datetime, ) -> bool: async with self._engine.connect() as raw: writer = await raw.execution_options(isolation_level="AUTOCOMMIT") @@ -449,7 +449,7 @@ async def test_mark_pending_reschedules_on_token_match(contract: _Harness) -> No assert row["acquired_at"] is None assert row["attempts_count"] == 1 next_attempt_at = row["next_attempt_at"] - assert isinstance(next_attempt_at, _dt.datetime) + assert isinstance(next_attempt_at, dt.datetime) assert next_attempt_at > utcnow() diff --git a/tests/test_fake.py b/tests/test_fake.py index 336dc4b..9496a63 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -1,5 +1,5 @@ import asyncio -import datetime as _dt +import datetime as dt import logging import typing import uuid @@ -223,7 +223,7 @@ async def handle(body: str) -> None: assert len(test_broker.fake_client.rows) == 1 row = test_broker.fake_client.rows[0] assert row.attempts_count == 1 - assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC) # rescheduled, not deleted + assert row.next_attempt_at > dt.datetime.now(tz=dt.UTC) # rescheduled, not deleted async def test_fake_broker_correlation_id_in_handler_context() -> None: @@ -287,7 +287,7 @@ async def handle(body: str) -> None: await broker.publish( # ty: ignore[missing-argument] "delayed", queue="orders", - activate_in=_dt.timedelta(seconds=5), + activate_in=dt.timedelta(seconds=5), ) assert received == ["delayed"] @@ -304,7 +304,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - future = _dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(seconds=5) + future = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=5) await broker.publish("at-future", queue="orders", activate_at=future) # ty: ignore[missing-argument] assert received == ["at-future"] @@ -325,7 +325,7 @@ async def handle(body: str) -> None: "a", "b", queue="orders", - activate_in=_dt.timedelta(seconds=5), + activate_in=dt.timedelta(seconds=5), ) assert received == ["a", "b"] @@ -342,7 +342,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker) async with test_broker: - future = _dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(seconds=5) + future = dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=5) await broker.publish_batch("a", "b", queue="orders", activate_at=future) # ty: ignore[missing-argument] assert received == ["a", "b"] @@ -357,8 +357,8 @@ async def test_fake_broker_publish_rejects_both_activate_in_and_activate_at() -> await broker.publish( # ty: ignore[missing-argument] "x", queue="orders", - activate_in=_dt.timedelta(seconds=1), - activate_at=_dt.datetime.now(tz=_dt.UTC), + activate_in=dt.timedelta(seconds=1), + activate_at=dt.datetime.now(tz=dt.UTC), ) @@ -370,8 +370,8 @@ async def test_fake_broker_publish_batch_rejects_both_activate_in_and_activate_a await broker.publish_batch( # ty: ignore[missing-argument] "x", queue="orders", - activate_in=_dt.timedelta(seconds=1), - activate_at=_dt.datetime.now(tz=_dt.UTC), + activate_in=dt.timedelta(seconds=1), + activate_at=dt.datetime.now(tz=dt.UTC), ) @@ -384,7 +384,7 @@ async def test_fake_broker_publish_rejects_naive_activate_at() -> None: await broker.publish( # ty: ignore[missing-argument] "x", queue="orders", - activate_at=_dt.datetime.now(), # noqa: DTZ005 + activate_at=dt.datetime.now(), # noqa: DTZ005 ) @@ -396,7 +396,7 @@ async def test_fake_broker_publish_batch_rejects_naive_activate_at() -> None: await broker.publish_batch( # ty: ignore[missing-argument] "x", queue="orders", - activate_at=_dt.datetime.now(), # noqa: DTZ005 + activate_at=dt.datetime.now(), # noqa: DTZ005 ) @@ -623,8 +623,8 @@ class RecordingStrategy(RetryStrategyProto): def get_next_attempt_delay( self, *, - first_attempt_at: _dt.datetime, # noqa: ARG002 - last_attempt_at: _dt.datetime, # noqa: ARG002 + first_attempt_at: dt.datetime, # noqa: ARG002 + last_attempt_at: dt.datetime, # noqa: ARG002 attempts_count: int, # noqa: ARG002 exception: BaseException | None = None, ) -> float | None: @@ -772,7 +772,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker, run_loops=True) async with test_broker: # Manually create a row with an expired lease — fetch must reclaim it. - old = _dt.datetime.now(tz=_dt.UTC) - _dt.timedelta(seconds=10) + old = dt.datetime.now(tz=dt.UTC) - dt.timedelta(seconds=10) test_broker.fake_client._rows.append( # noqa: SLF001 _FakeRow( id=99, @@ -797,7 +797,7 @@ async def handle(body: str) -> None: test_broker = TestOutboxBroker(broker, run_loops=True) async with test_broker: - future = _dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(milliseconds=300) + future = dt.datetime.now(tz=dt.UTC) + dt.timedelta(milliseconds=300) test_broker.feed("orders", b'"delayed"', next_attempt_at=future, headers={"content-type": "application/json"}) # Before the gate opens: nothing delivered. await asyncio.sleep(0.1) @@ -938,8 +938,8 @@ class TransientOnlyStrategy(RetryStrategyProto): def get_next_attempt_delay( self, *, - first_attempt_at: _dt.datetime, # noqa: ARG002 - last_attempt_at: _dt.datetime, # noqa: ARG002 + first_attempt_at: dt.datetime, # noqa: ARG002 + last_attempt_at: dt.datetime, # noqa: ARG002 attempts_count: int, # noqa: ARG002 exception: BaseException | None = None, ) -> float | None: @@ -1234,7 +1234,7 @@ def test_fake_client_feed_rejects_naive_next_attempt_at() -> None: """P31: feed() rejects a naive datetime up front, matching the tz-strict publish path.""" fake = FakeOutboxClient() with pytest.raises(ValueError, match="timezone-aware"): - fake.feed(queue="q", payload=b"x", next_attempt_at=_dt.datetime.now()) # noqa: DTZ005 # naive on purpose + fake.feed(queue="q", payload=b"x", next_attempt_at=dt.datetime.now()) # noqa: DTZ005 # naive on purpose async def test_fake_headers_not_shared_by_reference() -> None: @@ -1260,7 +1260,7 @@ async def test_fake_client_cancel_timer_skips_leased_row() -> None: fake = FakeOutboxClient() fake.feed(queue="q", payload=b"x", timer_id="email-1") fake.rows[0].acquired_token = uuid.uuid4() - fake.rows[0].acquired_at = _dt.datetime.now(tz=_dt.UTC) + fake.rows[0].acquired_at = dt.datetime.now(tz=dt.UTC) assert await fake.cancel_timer(queue="q", timer_id="email-1", session=object()) is False # ty: ignore[invalid-argument-type] assert len(fake.rows) == 1 @@ -1271,7 +1271,7 @@ async def test_fake_client_terminal_writes_reject_none_token() -> None: row_id = fake.feed(queue="q", payload=b"x") # unleased -> acquired_token is None assert row_id is not None assert fake.rows[0].acquired_token is None - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) # Without the ``acquired_token is not None`` guard the fake's ``None == None`` would match # this row, deleting / rescheduling where the real client's SQL no-ops. assert await fake.delete_with_lease(None, row_id, None) is False # ty: ignore[invalid-argument-type] @@ -1338,7 +1338,7 @@ async def handle(body: str) -> None: assert len(test_broker.fake_client.rows) == 1 row = test_broker.fake_client.rows[0] assert row.attempts_count == 1 - assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC) + assert row.next_attempt_at > dt.datetime.now(tz=dt.UTC) async def test_fake_broker_manual_policy_handler_exception_retries_not_deletes() -> None: @@ -1360,7 +1360,7 @@ async def handle(body: str) -> None: assert len(test_broker.fake_client.rows) == 1 row = test_broker.fake_client.rows[0] assert row.attempts_count == 1 - assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC) + assert row.next_attempt_at > dt.datetime.now(tz=dt.UTC) async def test_fake_broker_nack_message_exception_retries_not_deletes() -> None: @@ -1380,7 +1380,7 @@ async def handle(body: str) -> None: assert len(test_broker.fake_client.rows) == 1 row = test_broker.fake_client.rows[0] assert row.attempts_count == 1 - assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC) + assert row.next_attempt_at > dt.datetime.now(tz=dt.UTC) async def test_fake_broker_manual_handler_without_ack_is_rejected_via_dispatch() -> None: @@ -1467,9 +1467,9 @@ async def test_publisher_with_activate_in_records_next_attempt() -> None: # inspect next_attempt_at without sync dispatch deleting it. pub = broker.publisher("backlog") test_broker = TestOutboxBroker(broker) - before = _dt.datetime.now(tz=_dt.UTC) + before = dt.datetime.now(tz=dt.UTC) async with test_broker: - await pub.publish(b"x", session=_fake_session(), activate_in=_dt.timedelta(seconds=30)) + await pub.publish(b"x", session=_fake_session(), activate_in=dt.timedelta(seconds=30)) assert len(test_broker.fake_client.rows) == 1 assert test_broker.fake_client.rows[0].next_attempt_at > before @@ -1654,7 +1654,7 @@ async def handle(body: dict) -> OutboxResponse: body="delayed", queue="downstream", # no subscriber — row persists for inspection session=_fake_session(), - activate_in=_dt.timedelta(seconds=60), + activate_in=dt.timedelta(seconds=60), ) test_broker = TestOutboxBroker(broker) @@ -1663,7 +1663,7 @@ async def handle(body: dict) -> OutboxResponse: downstream = [r for r in test_broker.fake_client.rows if r.queue == "downstream"] assert len(downstream) == 1 - assert downstream[0].next_attempt_at > _dt.datetime.now(tz=_dt.UTC) + assert downstream[0].next_attempt_at > dt.datetime.now(tz=dt.UTC) async def test_handler_returning_outbox_response_with_timer_id_dedups() -> None: @@ -1891,7 +1891,7 @@ async def test_fake_dlq_captures_max_deliveries_failure() -> None: row = test_broker.fake_client.rows[0] row.deliveries_count = 5 row.acquired_token = uuid.uuid4() - row.acquired_at = _dt.datetime.now(tz=_dt.UTC) + row.acquired_at = dt.datetime.now(tz=dt.UTC) sub = next(iter(broker._subscribers)) # noqa: SLF001 await sub.dispatch_one(_to_inner(row), writer_conn=None) @@ -1972,7 +1972,7 @@ def recorder(event: str, tags: typing.Any) -> None: row = test_broker.fake_client.rows[0] row.deliveries_count = 5 # over max_deliveries=1 → terminal without ever running the handler row.acquired_token = uuid.uuid4() - row.acquired_at = _dt.datetime.now(tz=_dt.UTC) + row.acquired_at = dt.datetime.now(tz=dt.UTC) await sub.dispatch_one(_to_inner(row), writer_conn=None) dlq_events = [t for e, t in events if e == "dlq_written"] diff --git a/tests/test_integration.py b/tests/test_integration.py index f75fce7..82ba2a6 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2,7 +2,7 @@ import asyncio import contextlib -import datetime as _dt +import datetime as dt import logging import uuid from collections.abc import Mapping @@ -229,7 +229,7 @@ async def test_writer_connection_autocommit_round_trip(pg_engine: AsyncEngine, o is valid on the asyncpg dialect. """ token = uuid.uuid4() - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) async with pg_engine.begin() as conn: result = await conn.execute( insert(outbox_table) @@ -273,15 +273,15 @@ async def test_mark_pending_with_lease_uses_db_clock(pg_engine: AsyncEngine, out msg.acquired_token, # ty: ignore[invalid-argument-type] delay_seconds=delay, attempts_count=1, - first_attempt_at=_dt.datetime.now(tz=_dt.UTC), - last_attempt_at=_dt.datetime.now(tz=_dt.UTC), + first_attempt_at=dt.datetime.now(tz=dt.UTC), + last_attempt_at=dt.datetime.now(tz=dt.UTC), ) async with pg_engine.connect() as conn: db_after = (await conn.execute(text("SELECT clock_timestamp()"))).scalar_one() next_at = (await conn.execute(select(outbox_table.c.next_attempt_at))).scalar_one() # next_attempt_at was set by the autocommit'd UPDATE whose server-side # now() falls between db_before and db_after. - assert db_before + _dt.timedelta(seconds=delay) <= next_at <= db_after + _dt.timedelta(seconds=delay) + assert db_before + dt.timedelta(seconds=delay) <= next_at <= db_after + dt.timedelta(seconds=delay) async def test_end_to_end_subscriber_delivers_inserted_row(pg_engine, outbox_table) -> None: @@ -574,7 +574,7 @@ async def handle(body: dict) -> None: {"order_id": 1}, queue="orders", session=session, - activate_in=_dt.timedelta(milliseconds=500), + activate_in=dt.timedelta(milliseconds=500), ) # Before the gate opens: nothing delivered. await asyncio.sleep(0.2) @@ -592,7 +592,7 @@ async def test_publish_with_activate_at_delays_delivery(pg_engine, outbox_table) async def handle(body: dict) -> None: received.append(body) - fire_at = _dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(milliseconds=500) + fire_at = dt.datetime.now(tz=dt.UTC) + dt.timedelta(milliseconds=500) session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) async with broker: async with session_factory() as session, session.begin(): @@ -617,8 +617,8 @@ async def test_publish_rejects_activate_in_and_at_together(pg_engine, outbox_tab {"x": 1}, queue="orders", session=session, - activate_in=_dt.timedelta(seconds=1), - activate_at=_dt.datetime.now(tz=_dt.UTC), + activate_in=dt.timedelta(seconds=1), + activate_at=dt.datetime.now(tz=dt.UTC), ) @@ -680,7 +680,7 @@ async def handle(body: dict) -> None: {"order_id": 9}, queue="orders", session=session, - activate_in=_dt.timedelta(seconds=1), + activate_in=dt.timedelta(seconds=1), timer_id="email-cancel", ) # Cancel before activate_in elapses. @@ -752,7 +752,7 @@ async def test_publish_batch_with_activate_in_delays_all_rows(pg_engine, outbox_ {"i": 3}, queue="orders", session=session, - activate_in=_dt.timedelta(minutes=10), # well past any test horizon + activate_in=dt.timedelta(minutes=10), # well past any test horizon ) # Rows are inserted but invisible to fetch (next_attempt_at in the future). assert await _row_count(pg_engine, outbox_table) == 3 @@ -830,15 +830,15 @@ async def test_fetch_unprocessed_includes_future_dated_rows(pg_engine, outbox_ta "later", queue="orders", session=session, - activate_in=_dt.timedelta(minutes=5), + activate_in=dt.timedelta(minutes=5), ) async with session_factory() as session: rows = await broker.fetch_unprocessed(session=session, queue="orders") assert len(rows) == 2 - now = _dt.datetime.now(tz=_dt.UTC) - future = [r for r in rows if r.next_attempt_at > now + _dt.timedelta(minutes=1)] + now = dt.datetime.now(tz=dt.UTC) + future = [r for r in rows if r.next_attempt_at > now + dt.timedelta(minutes=1)] assert len(future) == 1 @@ -1380,7 +1380,7 @@ async def handle(body: dict) -> None: start = asyncio.get_event_loop().time() async with broker: async with session_factory() as session, session.begin(): - await pub.publish({"x": 1}, session=session, activate_in=_dt.timedelta(seconds=delay_seconds)) + await pub.publish({"x": 1}, session=session, activate_in=dt.timedelta(seconds=delay_seconds)) await _wait_until(lambda: received, timeout=5.0) elapsed = asyncio.get_event_loop().time() - start @@ -1675,7 +1675,7 @@ async def test_lease_pairing_check_rejects_half_set_lease(pg_engine: AsyncEngine insert(outbox_table).values( queue="orders", payload=b"x", - acquired_at=_dt.datetime.now(tz=_dt.UTC), # acquired_token left NULL + acquired_at=dt.datetime.now(tz=dt.UTC), # acquired_token left NULL ), ) @@ -2343,7 +2343,7 @@ async def test_notify_still_skipped_for_future_dated(pg_engine, outbox_table) -> session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) with _count_pg_notify(pg_engine) as count: async with session_factory() as session, session.begin(): - await broker.publish({"n": 1}, queue="orders", session=session, activate_in=_dt.timedelta(minutes=5)) + await broker.publish({"n": 1}, queue="orders", session=session, activate_in=dt.timedelta(minutes=5)) assert count[0] == 0 # future-dated -> no NOTIFY (unchanged), dedup did not break the skip diff --git a/tests/test_middleware_opentelemetry.py b/tests/test_middleware_opentelemetry.py index d7d21d5..1131e6e 100644 --- a/tests/test_middleware_opentelemetry.py +++ b/tests/test_middleware_opentelemetry.py @@ -6,7 +6,7 @@ ``tests/test_metrics_opentelemetry.py``. """ -import datetime as _dt +import datetime as dt import typing import uuid from unittest.mock import AsyncMock, MagicMock, patch @@ -61,7 +61,7 @@ def _session_mock() -> AsyncMock: def _make_inner_message(*, payload: bytes = b"hello", queue: str = "orders") -> OutboxInnerMessage: - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) return OutboxInnerMessage( id=7, queue=queue, diff --git a/tests/test_middleware_prometheus.py b/tests/test_middleware_prometheus.py index 37c984d..c0a3131 100644 --- a/tests/test_middleware_prometheus.py +++ b/tests/test_middleware_prometheus.py @@ -5,7 +5,7 @@ publish-scope middleware fires through ``_basic_publish``. """ -import datetime as _dt +import datetime as dt import uuid from unittest.mock import AsyncMock, MagicMock, patch @@ -47,7 +47,7 @@ def _session_mock() -> AsyncMock: def _make_inner_message(*, payload: bytes = b"hello", queue: str = "orders") -> OutboxInnerMessage: - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) return OutboxInnerMessage( id=7, queue=queue, diff --git a/tests/test_unit.py b/tests/test_unit.py index e5a9097..8f2a31e 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -1,5 +1,5 @@ import asyncio -import datetime as _dt +import datetime as dt import json import logging import math @@ -267,9 +267,9 @@ def test_encode_payload_serializes_pydantic_model_with_default_serializer() -> N # --- retry strategies --- -def _make_times() -> tuple[_dt.datetime, _dt.datetime]: - first = _dt.datetime(2026, 1, 1, tzinfo=_dt.UTC) - last = first + _dt.timedelta(seconds=10) +def _make_times() -> tuple[dt.datetime, dt.datetime]: + first = dt.datetime(2026, 1, 1, tzinfo=dt.UTC) + last = first + dt.timedelta(seconds=10) return first, last @@ -360,11 +360,11 @@ def _make_msg(**overrides: object) -> OutboxInnerMessage: "headers": None, "attempts_count": 0, "deliveries_count": 1, - "created_at": _dt.datetime.now(tz=_dt.UTC), - "next_attempt_at": _dt.datetime.now(tz=_dt.UTC), + "created_at": dt.datetime.now(tz=dt.UTC), + "next_attempt_at": dt.datetime.now(tz=dt.UTC), "first_attempt_at": None, "last_attempt_at": None, - "acquired_at": _dt.datetime.now(tz=_dt.UTC), + "acquired_at": dt.datetime.now(tz=dt.UTC), "acquired_token": uuid.uuid4(), } base.update(overrides) @@ -561,7 +561,7 @@ async def test_nack_with_raising_strategy_degrades_to_retry_terminal() -> None: def test_exponential_retry_does_not_overflow_at_extreme_attempts() -> None: """B7: ExponentialRetry with no max_attempts/max_delay must not raise OverflowError.""" strategy = ExponentialRetry(initial_delay_seconds=1.0, multiplier=2.0) - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) delay = strategy.get_next_attempt_delay(first_attempt_at=now, last_attempt_at=now, attempts_count=2000) assert delay is not None assert math.isfinite(delay) @@ -849,14 +849,14 @@ async def test_publish_command_validates_activate_args_mutex() -> None: b"x", queue="orders", session=session, - activate_in=_dt.timedelta(seconds=1), - activate_at=_dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(seconds=1), + activate_in=dt.timedelta(seconds=1), + activate_at=dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=1), ) async def test_publish_command_rejects_naive_activate_at() -> None: session = _make_session_mock() - naive = _dt.datetime(2026, 5, 23, 12, 0, 0) # noqa: DTZ001 + naive = dt.datetime(2026, 5, 23, 12, 0, 0) # noqa: DTZ001 with pytest.raises(ValueError, match="timezone-aware"): OutboxPublishCommand(b"x", queue="orders", session=session, activate_at=naive) @@ -869,6 +869,12 @@ async def test_publish_command_batch_bodies_preserves_none_body() -> None: assert OutboxPublishCommand(b"a", b"b", queue="q", session=session).batch_bodies == (b"a", b"b") +async def test_publish_command_batch_bodies_is_assignable() -> None: + cmd = OutboxPublishCommand(b"a", b"b", queue="q", session=_make_session_mock()) + cmd.batch_bodies = (None, b"c") + assert cmd.batch_bodies == (None, b"c") + + def test_publish_command_rejects_empty_queue() -> None: """P5: queue validation in the single-source-of-truth constructor.""" with pytest.raises(ValueError, match="non-empty"): @@ -970,15 +976,15 @@ async def test_broker_publish_rejects_activate_in_and_at_together() -> None: b"x", queue="orders", session=session, - activate_in=_dt.timedelta(seconds=1), - activate_at=_dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(seconds=1), + activate_in=dt.timedelta(seconds=1), + activate_at=dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=1), ) async def test_broker_publish_with_activate_in_skips_notify() -> None: broker = _make_broker() session = _make_session_mock() - await broker.publish(b"x", queue="orders", session=session, activate_in=_dt.timedelta(seconds=30)) + await broker.publish(b"x", queue="orders", session=session, activate_in=dt.timedelta(seconds=30)) # Only the INSERT — no NOTIFY for future-dated rows. assert session.execute.await_count == 1 insert_stmt = session.execute.await_args_list[0].args[0] @@ -989,7 +995,7 @@ async def test_broker_publish_with_activate_in_skips_notify() -> None: async def test_broker_publish_with_activate_at_skips_notify() -> None: broker = _make_broker() session = _make_session_mock() - fire = _dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(minutes=5) + fire = dt.datetime.now(tz=dt.UTC) + dt.timedelta(minutes=5) await broker.publish(b"x", queue="orders", session=session, activate_at=fire) assert session.execute.await_count == 1 params = session.execute.await_args_list[0].args[0].compile().params @@ -1001,7 +1007,7 @@ async def test_broker_publish_emits_notify_when_activate_at_is_past() -> None: # listeners wake without waiting for the next poll tick. broker = _make_broker() session = _make_session_mock() - past = _dt.datetime.now(tz=_dt.UTC) - _dt.timedelta(seconds=5) + past = dt.datetime.now(tz=dt.UTC) - dt.timedelta(seconds=5) await broker.publish(b"x", queue="orders", session=session, activate_at=past) assert session.execute.await_count == 2 notify_stmt, notify_params = session.execute.await_args_list[1].args @@ -1012,7 +1018,7 @@ async def test_broker_publish_emits_notify_when_activate_at_is_past() -> None: async def test_broker_publish_emits_notify_when_activate_in_is_zero() -> None: broker = _make_broker() session = _make_session_mock() - await broker.publish(b"x", queue="orders", session=session, activate_in=_dt.timedelta(0)) + await broker.publish(b"x", queue="orders", session=session, activate_in=dt.timedelta(0)) assert session.execute.await_count == 2 notify_stmt, _params = session.execute.await_args_list[1].args assert "pg_notify" in str(notify_stmt) @@ -1021,7 +1027,7 @@ async def test_broker_publish_emits_notify_when_activate_in_is_zero() -> None: async def test_broker_publish_rejects_naive_activate_at() -> None: broker = _make_broker() session = _make_session_mock() - naive = _dt.datetime(2026, 5, 23, 12, 0, 0) # noqa: DTZ001 + naive = dt.datetime(2026, 5, 23, 12, 0, 0) # noqa: DTZ001 with pytest.raises(ValueError, match="timezone-aware"): await broker.publish(b"x", queue="orders", session=session, activate_at=naive) @@ -1064,8 +1070,8 @@ async def test_broker_publish_batch_rejects_activate_in_and_at_together() -> Non b"a", queue="orders", session=session, - activate_in=_dt.timedelta(seconds=1), - activate_at=_dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(seconds=1), + activate_in=dt.timedelta(seconds=1), + activate_at=dt.datetime.now(tz=dt.UTC) + dt.timedelta(seconds=1), ) @@ -1077,7 +1083,7 @@ async def test_broker_publish_batch_with_activate_in_skips_notify() -> None: b"b", queue="orders", session=session, - activate_in=_dt.timedelta(seconds=30), + activate_in=dt.timedelta(seconds=30), ) # Insert only — no NOTIFY for future-dated batch. assert session.execute.await_count == 1 @@ -1088,7 +1094,7 @@ async def test_broker_publish_batch_with_activate_in_skips_notify() -> None: async def test_broker_publish_batch_with_activate_at_skips_notify() -> None: broker = _make_broker() session = AsyncMock(spec=AsyncSession) - fire = _dt.datetime.now(tz=_dt.UTC) + _dt.timedelta(minutes=5) + fire = dt.datetime.now(tz=dt.UTC) + dt.timedelta(minutes=5) await broker.publish_batch(b"a", b"b", queue="orders", session=session, activate_at=fire) # No NOTIFY: future-dated rows. assert session.execute.await_count == 1 @@ -1099,7 +1105,7 @@ async def test_broker_publish_batch_with_activate_at_skips_notify() -> None: async def test_broker_publish_batch_emits_notify_when_activate_at_is_past() -> None: broker = _make_broker() session = _make_session_mock() - past = _dt.datetime.now(tz=_dt.UTC) - _dt.timedelta(seconds=5) + past = dt.datetime.now(tz=dt.UTC) - dt.timedelta(seconds=5) await broker.publish_batch(b"a", b"b", queue="orders", session=session, activate_at=past) # INSERT + NOTIFY: past activate_at is immediately eligible. assert session.execute.await_count == 2 @@ -1111,7 +1117,7 @@ async def test_broker_publish_batch_emits_notify_when_activate_at_is_past() -> N async def test_broker_publish_batch_rejects_naive_activate_at() -> None: broker = _make_broker() session = AsyncMock(spec=AsyncSession) - naive = _dt.datetime(2026, 5, 23, 12, 0, 0) # noqa: DTZ001 + naive = dt.datetime(2026, 5, 23, 12, 0, 0) # noqa: DTZ001 with pytest.raises(ValueError, match="timezone-aware"): await broker.publish_batch(b"a", queue="orders", session=session, activate_at=naive) @@ -1682,7 +1688,7 @@ async def test_client_mark_pending_with_lease_raises_typeerror_on_none_conn() -> metadata = MetaData() t = make_outbox_table(metadata) client = OutboxClient(AsyncMock(), t) - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) with pytest.raises(TypeError, match=r"OutboxClient\.mark_pending_with_lease requires a live AsyncConnection"): await client.mark_pending_with_lease( None, @@ -1818,7 +1824,7 @@ async def test_fake_client_delete_miss() -> None: async def test_fake_client_mark_pending_miss() -> None: client = FakeOutboxClient() - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) updated = await client.mark_pending_with_lease( None, 999, @@ -2098,7 +2104,7 @@ def _make_msg_over_max_deliveries(**overrides: object) -> OutboxInnerMessage: async def test_fake_client_fetch_ties_break_on_id() -> None: """L2: rows with identical ``next_attempt_at`` are claimed in ``id`` order.""" client = FakeOutboxClient() - same_time = _dt.datetime(2026, 5, 23, 12, 0, 0, tzinfo=_dt.UTC) + same_time = dt.datetime(2026, 5, 23, 12, 0, 0, tzinfo=dt.UTC) id_a = client.feed(queue="orders", payload=b"a", next_attempt_at=same_time) id_b = client.feed(queue="orders", payload=b"b", next_attempt_at=same_time) id_c = client.feed(queue="orders", payload=b"c", next_attempt_at=same_time) @@ -3050,7 +3056,7 @@ async def test_outbox_client_mark_pending_with_lease_uses_caller_conn() -> None: fake_conn = MagicMock() fake_conn.begin = MagicMock() fake_conn.execute = AsyncMock(return_value=MagicMock(rowcount=1)) - now = _dt.datetime.now(tz=_dt.UTC) + now = dt.datetime.now(tz=dt.UTC) updated = await client.mark_pending_with_lease( fake_conn, @@ -4158,7 +4164,7 @@ async def test_dlq_cte_insert_columns_match_make_dlq_table() -> None: def test_outbox_response_rejects_naive_activate_at_eagerly() -> None: """OutboxResponse must reject a naive activate_at at construction, not defer it to dispatch time.""" - naive = _dt.datetime(2030, 1, 1, 12, 0, 0) # noqa: DTZ001 # deliberately tz-naive + naive = dt.datetime(2030, 1, 1, 12, 0, 0) # noqa: DTZ001 # deliberately tz-naive with pytest.raises(ValueError, match="OutboxResponse requires activate_at to be timezone-aware"): OutboxResponse( {"x": 1}, @@ -4170,13 +4176,13 @@ def test_outbox_response_rejects_naive_activate_at_eagerly() -> None: def test_outbox_response_rejects_both_activate_args_eagerly() -> None: """OutboxResponse must reject activate_in + activate_at together at construction.""" - aware = _dt.datetime(2030, 1, 1, 12, 0, 0, tzinfo=_dt.UTC) + aware = dt.datetime(2030, 1, 1, 12, 0, 0, tzinfo=dt.UTC) with pytest.raises(ValueError, match="OutboxResponse accepts at most one of activate_in / activate_at"): OutboxResponse( {"x": 1}, queue="q", session=None, # ty: ignore[invalid-argument-type] # error raises before session is used - activate_in=_dt.timedelta(seconds=5), + activate_in=dt.timedelta(seconds=5), activate_at=aware, )