"""Fail-closed and terminal-outcome tests for Topic-59 publication.""" from __future__ import annotations import hashlib from dataclasses import dataclass, replace from datetime import date, datetime from pathlib import Path from zoneinfo import ZoneInfo import pytest from gateway.platforms.nutrition_weekly_operations import ( Topic59DayCardProjector, Topic59PublicationDisposition, Topic59PublicationRequest, ) 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, WeeklyOperationsConfigError, WeeklyOperationsReviewRoute, parse_weekly_operations_config, ) from gateway.platforms.nutrition_weekly_operations_ledger import Topic59PublicationLedger from gateway.platforms.nutrition_weekly_operations_models import ( Topic59ApprovalState, Topic59DayCard, Topic59DayCardError, Topic59DayState, Topic59DeliveryState, Topic59DraftState, project_topic59_day_card, ) from gateway.platforms.nutrition_weekly_operations_publication_contract import ( Topic59ProviderDelivered, Topic59ProviderFailure, Topic59ProviderKnownFailure, Topic59ProviderOutcome, Topic59ProviderUnknown, ) _KST = ZoneInfo("Asia/Seoul") _KEY = "client_001" _IDENTITY = hashlib.sha256(f"nutricoach-weekly-operations-customer-identity-v1\0{_KEY}".encode()).hexdigest() _CANDIDATE = hashlib.sha256(b"candidate").hexdigest() _CONSENT = hashlib.sha256(b"consent").hexdigest() @pytest.fixture def anyio_backend() -> str: return "asyncio" @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, send: Topic59ProviderOutcome, edit: Topic59ProviderOutcome) -> None: self.send_outcome: Topic59ProviderOutcome = send self.edit_outcome: Topic59ProviderOutcome = edit self.sent: int = 0 self.edited: int = 0 async def send(self, *, chat_id: str, topic_id: str, text: str) -> Topic59ProviderOutcome: assert (chat_id, topic_id) == ("10001", "59") assert text.startswith("[운영 현황]") self.sent += 1 return self.send_outcome async def edit(self, *, chat_id: str, topic_id: str, message_id: str, text: str) -> Topic59ProviderOutcome: assert (chat_id, topic_id, message_id) == ("10001", "59", "101") assert text.startswith("[운영 현황]") self.edited += 1 return self.edit_outcome def _digest(label: str) -> str: return hashlib.sha256(label.encode()).hexdigest() def _request(kst_day: date, state: Topic59DayState) -> Topic59PublicationRequest: config = WeeklyOperationsConfig( True, review_route=WeeklyOperationsReviewRoute("90001", "10001"), registry_identity_binding_digest=registry_identity().binding_digest, ) receipt = authority_receipt_fixture( _CANDIDATE, config.digest, (_KEY,), ("90001", "10001", 1), _CONSENT, datetime(2026, 8, 1, tzinfo=_KST), datetime(2026, 9, 1, tzinfo=_KST), config.feature_epoch, ) runtime = WeeklyOperationsRuntimeContext( _CANDIDATE, _KEY, "90001", "10001", 1, _CONSENT, True, config.feature_epoch, datetime(2026, 8, 20, tzinfo=_KST), ) card = Topic59DayCard.from_sidecar( _Row(_IDENTITY, kst_day, state.value, f"event-{state.value}", _digest(f"row-{state.value}"), _digest(f"digest-{state.value}")), 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 _success_transport() -> _Transport: success = Topic59ProviderDelivered("101") return _Transport(success, success) def test_wrong_topic_is_rejected_before_the_projector_can_call_a_provider() -> None: with pytest.raises(WeeklyOperationsConfigError): _ = parse_weekly_operations_config( {"nutrition_coaching": {"weekly_operations": {"enabled": True, "reminder_time": "20:00:00", "missed_cutoff_time": "23:00:00", "weekly_weekday": 0}, "operator_review": {"user_id": "90001", "chat_id": "10001", "topic_id": "58"}}} ) @pytest.mark.anyio async def test_wrong_user_chat_config_or_authority_make_zero_provider_calls(tmp_path: Path) -> None: request = _request(date(2026, 8, 17), Topic59DayState.SUBMITTED) denied = ( replace(request, config=replace(request.config, enabled=False)), replace(request, runtime=replace(request.runtime, consent_granted=False)), replace(request, config=replace(request.config, review_route=WeeklyOperationsReviewRoute("90002", "10001"))), replace(request, config=replace(request.config, review_route=WeeklyOperationsReviewRoute("90001", "10002"))), replace(request, card=replace(request.card, candidate_digest=_digest("other-candidate"))), ) for index, value in enumerate(denied): transport = _success_transport() result = await Topic59DayCardProjector(tmp_path / f"denied-{index}.jsonl").publish(value, transport) assert result.disposition is Topic59PublicationDisposition.REFUSED assert (transport.sent, transport.edited) == (0, 0) @pytest.mark.anyio async def test_unknown_and_known_send_outcomes_are_terminal_without_replacement(tmp_path: Path) -> None: request = _request(date(2026, 8, 18), Topic59DayState.SUBMITTED) outcomes = ((Topic59ProviderUnknown(Topic59ProviderFailure.TIMEOUT), "unknown"), (Topic59ProviderKnownFailure(Topic59ProviderFailure.REJECTED), "failed")) for index, (outcome, terminal_state) in enumerate(outcomes): ledger_path = tmp_path / f"send-{index}.jsonl" transport = _Transport(outcome, Topic59ProviderDelivered("101")) first = await Topic59DayCardProjector(ledger_path).publish(request, transport) second = await Topic59DayCardProjector(ledger_path).publish(request, transport) assert first.disposition is second.disposition is Topic59PublicationDisposition.INCIDENT assert transport.sent == 1 assert f'"state":"{terminal_state}"' in ledger_path.read_text(encoding="utf-8") @pytest.mark.anyio async def test_unknown_and_known_edit_outcomes_are_terminal_without_replacement(tmp_path: Path) -> None: day = date(2026, 8, 19) outcomes = ((Topic59ProviderUnknown(Topic59ProviderFailure.CONNECTION_LOST), "unknown"), (Topic59ProviderKnownFailure(Topic59ProviderFailure.REJECTED), "failed")) for index, (outcome, terminal_state) in enumerate(outcomes): ledger_path = tmp_path / f"edit-{index}.jsonl" transport = _Transport(Topic59ProviderDelivered("101"), outcome) projector = Topic59DayCardProjector(ledger_path) assert (await projector.publish(_request(day, Topic59DayState.MISSED), transport)).message_id == "101" first = await projector.publish(_request(day, Topic59DayState.LATE_SUBMITTED), transport) second = await projector.publish(_request(day, Topic59DayState.LATE_SUBMITTED), transport) assert first.disposition is second.disposition is Topic59PublicationDisposition.INCIDENT assert (transport.sent, transport.edited) == (1, 1) assert f'"state":"{terminal_state}"' in ledger_path.read_text(encoding="utf-8") @pytest.mark.anyio async def test_unresolved_reservation_is_an_incident_not_a_blind_new_send(tmp_path: Path) -> None: request = _request(date(2026, 8, 20), Topic59DayState.SUBMITTED) route = request.config.review_route assert route is not None ledger_path = tmp_path / "reserved.jsonl" projection = project_topic59_day_card(request.card, config_digest=request.config.digest, route_key=route.key) assert Topic59PublicationLedger(ledger_path).claim(projection).kind == "send" transport = _success_transport() result = await Topic59DayCardProjector(ledger_path).publish(request, transport) assert result.disposition is Topic59PublicationDisposition.INCIDENT assert (transport.sent, transport.edited) == (0, 0) @pytest.mark.anyio async def test_corrupt_ledger_becomes_an_incident_without_a_provider_call(tmp_path: Path) -> None: ledger_path = tmp_path / "corrupt.jsonl" _ = ledger_path.write_text("{}\n", encoding="utf-8") transport = _success_transport() result = await Topic59DayCardProjector(ledger_path).publish( _request(date(2026, 8, 21), Topic59DayState.SUBMITTED), transport ) assert result.disposition is Topic59PublicationDisposition.INCIDENT assert (transport.sent, transport.edited) == (0, 0) def test_malformed_sidecar_is_rejected_before_a_provider_call() -> None: with pytest.raises(Topic59DayCardError): _ = Topic59DayCard.from_sidecar( _Row(_IDENTITY, date(2026, 8, 21), "submitted", "event", _digest("row"), "bad"), completed_field_count=5, draft_state=Topic59DraftState.DRAFTED, approval_state=Topic59ApprovalState.PENDING, delivery_state=Topic59DeliveryState.NOT_SENT, candidate_digest=_CANDIDATE, )