"""R2 immutable-slot drift and concurrent reservation tests.""" from __future__ import annotations import anyio import hashlib from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from datetime import date, datetime from pathlib import Path from threading import Event from zoneinfo import ZoneInfo import pytest from typing_extensions import override from gateway.platforms.nutrition_weekly_operations import ( Topic59DayCardProjector, Topic59PublicationDisposition, Topic59PublicationRequest, Topic59PublicationResult, ) from tests.gateway._weekly_operations_authority_support import ( authority_receipt_fixture, test_registry_identity as registry_identity, ) from gateway.platforms.nutrition_weekly_operations_authority import ( WeeklyOperationsAuthorityReceipt, WeeklyOperationsOwner, WeeklyOperationsRuntimeContext, ) from gateway.platforms.nutrition_weekly_operations_config import WeeklyOperationsConfig, WeeklyOperationsReviewRoute from gateway.platforms.nutrition_weekly_operations_models import ( Topic59ApprovalState, Topic59DayCard, Topic59DayState, Topic59DeliveryState, Topic59DraftState, Topic59Projection, project_topic59_day_card, ) from gateway.platforms.nutrition_weekly_operations_publication_contract import ( Topic59IncidentReason, Topic59ProviderDelivered, ) _KST = ZoneInfo("Asia/Seoul") _KEY = "client_001" _IDENTITY = hashlib.sha256(f"nutricoach-weekly-operations-customer-identity-v1\0{_KEY}".encode()).hexdigest() _CONSENT = hashlib.sha256(b"consent").hexdigest() @pytest.fixture def anyio_backend() -> str: return "asyncio" class Topic59ConcurrentTestError(Exception): """A barrier-driven concurrent test did not observe its subscribed send event.""" @dataclass(frozen=True, slots=True) class _Row: customer_identity_digest: str kst_day: date state: str source_event_id: str | None logical_key: str row_digest: str class _Transport: def __init__(self) -> None: self.sent: int = 0 self.edited: int = 0 async def send(self, *, chat_id: str, topic_id: str, text: str) -> Topic59ProviderDelivered: assert (chat_id, topic_id) == ("10001", "59") assert text.startswith("[운영 현황]") self.sent += 1 return Topic59ProviderDelivered("101") async def edit(self, *, chat_id: str, topic_id: str, message_id: str, text: str) -> Topic59ProviderDelivered: assert (chat_id, topic_id, message_id) == ("10001", "59", "101") assert text.startswith("[운영 현황]") self.edited += 1 return Topic59ProviderDelivered(message_id) class _BarrierTransport(_Transport): def __init__(self) -> None: super().__init__() self.started: Event = Event() self.release: Event = Event() @override async def send(self, *, chat_id: str, topic_id: str, text: str) -> Topic59ProviderDelivered: self.started.set() if not self.release.wait(timeout=1): raise Topic59ConcurrentTestError("send release was not observed") return await super().send(chat_id=chat_id, topic_id=topic_id, text=text) def _digest(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() def _request(candidate: str, *, user: str = "90001", chat: str = "10001") -> Topic59PublicationRequest: config = WeeklyOperationsConfig( True, review_route=WeeklyOperationsReviewRoute(user, chat), registry_identity_binding_digest=registry_identity().binding_digest, ) receipt = authority_receipt_fixture( candidate, config.digest, (_KEY,), (user, chat, 1), _CONSENT, datetime(2026, 8, 1, tzinfo=_KST), datetime(2026, 9, 1, tzinfo=_KST), config.feature_epoch, ) runtime = WeeklyOperationsRuntimeContext( candidate, _KEY, user, chat, 1, _CONSENT, True, config.feature_epoch, datetime(2026, 8, 24, tzinfo=_KST) ) card = Topic59DayCard.from_sidecar( _Row(_IDENTITY, date(2026, 8, 24), Topic59DayState.SUBMITTED.value, "source-row", "status-row", _digest("row")), completed_field_count=5, draft_state=Topic59DraftState.DRAFTED, approval_state=Topic59ApprovalState.PENDING, delivery_state=Topic59DeliveryState.NOT_SENT, candidate_digest=candidate, ) return Topic59PublicationRequest(card, config, receipt, runtime) def _projection(request: Topic59PublicationRequest) -> Topic59Projection: route = request.config.review_route assert route is not None return project_topic59_day_card(request.card, config_digest=request.config.digest, route_key=route.key) @pytest.mark.anyio async def test_candidate_config_and_route_drift_are_terminal_slot_incidents(tmp_path: Path) -> None: base = _request(_digest("candidate-a")) drifts = ( _request(_digest("candidate-b")), _request(_digest("candidate-a"), chat="10002"), _request(_digest("candidate-a"), user="90002"), ) for index, drift in enumerate(drifts): ledger_path = tmp_path / f"drift-{index}.jsonl" transport = _Transport() first = await Topic59DayCardProjector(ledger_path).publish(base, transport) second = await Topic59DayCardProjector(ledger_path).publish(drift, transport) third = await Topic59DayCardProjector(ledger_path).publish(drift, transport) assert _projection(base).card_slot == _projection(drift).card_slot assert first.disposition is Topic59PublicationDisposition.SENT assert second.disposition is third.disposition is Topic59PublicationDisposition.INCIDENT assert second.reason is Topic59IncidentReason.AUTHORITY_MISMATCH assert ledger_path.read_text(encoding="utf-8").count('"state":"authority_mismatch"') == 1 assert (transport.sent, transport.edited) == (1, 0) @pytest.mark.anyio async def test_malformed_stored_slot_is_an_incident_without_provider_replacement(tmp_path: Path) -> None: ledger_path = tmp_path / "malformed-slot.jsonl" _ = ledger_path.write_text('{"card_slot":"not-a-digest"}\n', encoding="utf-8") transport = _Transport() result = await Topic59DayCardProjector(ledger_path).publish(_request(_digest("candidate-a")), transport) assert result.disposition is Topic59PublicationDisposition.INCIDENT assert result.reason is Topic59IncidentReason.LEDGER_CORRUPTION assert (transport.sent, transport.edited) == (0, 0) def _publish_in_thread( projector: Topic59DayCardProjector, request: Topic59PublicationRequest, transport: _BarrierTransport ) -> Topic59PublicationResult: return anyio.run(projector.publish, request, transport) def test_concurrent_differing_pins_preserve_one_slot_and_one_initial_send(tmp_path: Path) -> None: ledger_path = tmp_path / "concurrent.jsonl" original = _request(_digest("candidate-a")) drift = _request(_digest("candidate-b")) transport = _BarrierTransport() with ThreadPoolExecutor(max_workers=2) as workers: first_future = workers.submit(_publish_in_thread, Topic59DayCardProjector(ledger_path), original, transport) if not transport.started.wait(timeout=1): raise Topic59ConcurrentTestError("initial send was not observed") second = workers.submit(_publish_in_thread, Topic59DayCardProjector(ledger_path), drift, transport).result(timeout=1) transport.release.set() first = first_future.result(timeout=1) assert first.disposition is Topic59PublicationDisposition.SENT assert second.disposition is Topic59PublicationDisposition.INCIDENT restart = anyio.run(Topic59DayCardProjector(ledger_path).publish, original, transport) assert second.reason is Topic59IncidentReason.AUTHORITY_MISMATCH assert restart.disposition is Topic59PublicationDisposition.INCIDENT assert (transport.sent, transport.edited) == (1, 0)