"""Focused behavior tests for immutable Topic-59 customer/day cards.""" from __future__ import annotations import hashlib from dataclasses import dataclass from datetime import date, datetime from pathlib import Path from zoneinfo import ZoneInfo import pytest from pydantic import BaseModel 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, WeeklyOperationsReviewRoute from gateway.platforms.nutrition_weekly_operations_models import ( Topic59ApprovalState, Topic59DayCard, Topic59DayState, Topic59DeliveryState, Topic59DraftState, ) from gateway.platforms.nutrition_weekly_operations_publication_contract import Topic59ProviderDelivered _KST = ZoneInfo("Asia/Seoul") _CUSTOMER_KEY = "client_001" _CUSTOMER_IDENTITY = hashlib.sha256( f"nutricoach-weekly-operations-customer-identity-v1\0{_CUSTOMER_KEY}".encode("utf-8") ).hexdigest() _CANDIDATE_DIGEST = hashlib.sha256(b"candidate").hexdigest() _CONSENT_DIGEST = hashlib.sha256(b"consent").hexdigest() _SAFE_KEYS = { "customer_identity_digest", "kst_day", "state", "completed_field_count", "draft_state", "approval_state", "delivery_state", "event_id", "status_row_id", } @pytest.fixture def anyio_backend() -> str: return "asyncio" @dataclass(frozen=True, slots=True) class _SidecarRow: customer_identity_digest: str kst_day: date state: str source_event_id: str | None logical_key: str row_digest: str class _LedgerRow(BaseModel): card_slot: str payload: dict[str, str | int] @dataclass(frozen=True, slots=True) class _LeakySidecarRow: customer_identity_digest: str kst_day: date state: str source_event_id: str | None logical_key: str row_digest: str customer_name: str raw_weight: str meal_note: str class _Transport: def __init__(self) -> None: self.sent: list[dict[str, str]] = [] self.edited: list[dict[str, str]] = [] async def send(self, *, chat_id: str, topic_id: str, text: str) -> Topic59ProviderDelivered: self.sent.append({"chat_id": chat_id, "topic_id": topic_id, "text": text}) return Topic59ProviderDelivered(str(100 + len(self.sent))) async def edit( self, *, chat_id: str, topic_id: str, message_id: str, text: str ) -> Topic59ProviderDelivered: self.edited.append({"chat_id": chat_id, "topic_id": topic_id, "message_id": message_id, "text": text}) return Topic59ProviderDelivered(message_id) def _digest(label: str) -> str: return hashlib.sha256(label.encode("utf-8")).hexdigest() def _row(kst_day: date, state: Topic59DayState) -> _SidecarRow: return _SidecarRow( _CUSTOMER_IDENTITY, kst_day, state.value, f"event-{kst_day.isoformat()}-{state.value}", _digest(f"row-{kst_day.isoformat()}-{state.value}"), _digest(f"digest-{kst_day.isoformat()}-{state.value}"), ) 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_DIGEST, config.digest, (_CUSTOMER_KEY,), ("90001", "10001", 1), _CONSENT_DIGEST, datetime(2026, 8, 1, tzinfo=_KST), datetime(2026, 9, 1, tzinfo=_KST), config.feature_epoch, ) runtime = WeeklyOperationsRuntimeContext( _CANDIDATE_DIGEST, _CUSTOMER_KEY, "90001", "10001", 1, _CONSENT_DIGEST, True, config.feature_epoch, datetime(2026, 8, 20, tzinfo=_KST), ) card = Topic59DayCard.from_sidecar( _row(kst_day, state), completed_field_count=5, draft_state=Topic59DraftState.DRAFTED, approval_state=Topic59ApprovalState.PENDING, delivery_state=Topic59DeliveryState.NOT_SENT, candidate_digest=_CANDIDATE_DIGEST, ) return Topic59PublicationRequest(card, config, receipt, runtime) @pytest.mark.anyio async def test_topic59_projector_exposes_one_customer_day_publication_binding(tmp_path: Path) -> None: result = await Topic59DayCardProjector(tmp_path / "topic59.jsonl").publish( _request(date(2026, 8, 17), Topic59DayState.SUBMITTED), _Transport() ) assert result.disposition is Topic59PublicationDisposition.SENT assert result.message_id == "101" @pytest.mark.anyio async def test_seven_kst_days_create_seven_logical_cards(tmp_path: Path) -> None: ledger_path = tmp_path / "topic59.jsonl" transport = _Transport() projector = Topic59DayCardProjector(ledger_path) for day in range(17, 24): assert (await projector.publish(_request(date(2026, 8, day), Topic59DayState.SUBMITTED), transport)).disposition is Topic59PublicationDisposition.SENT rows = [_LedgerRow.model_validate_json(line) for line in ledger_path.read_text(encoding="utf-8").splitlines()] assert len(transport.sent) == 7 assert len({row.card_slot for row in rows}) == 7 assert {call["topic_id"] for call in transport.sent} == {"59"} @pytest.mark.anyio async def test_missed_to_late_submitted_edits_the_persisted_message(tmp_path: Path) -> None: transport = _Transport() projector = Topic59DayCardProjector(tmp_path / "topic59.jsonl") day = date(2026, 8, 20) first = await projector.publish(_request(day, Topic59DayState.MISSED), transport) later = await projector.publish(_request(day, Topic59DayState.LATE_SUBMITTED), transport) assert (first.message_id, later.message_id) == ("101", "101") assert later.disposition is Topic59PublicationDisposition.EDITED assert (len(transport.sent), len(transport.edited)) == (1, 1) assert "지연 제출" in transport.edited[0]["text"] @pytest.mark.anyio async def test_restart_and_duplicate_tick_do_not_repeat_a_send(tmp_path: Path) -> None: ledger_path = tmp_path / "topic59.jsonl" transport = _Transport() request = _request(date(2026, 8, 21), Topic59DayState.SUBMITTED) first = await Topic59DayCardProjector(ledger_path).publish(request, transport) second = await Topic59DayCardProjector(ledger_path).publish(request, transport) assert first.disposition is Topic59PublicationDisposition.SENT assert second.disposition is Topic59PublicationDisposition.NOOP assert len(transport.sent) == 1 @pytest.mark.anyio async def test_rendered_and_durable_payloads_match_the_machine_whitelist(tmp_path: Path) -> None: ledger_path = tmp_path / "topic59.jsonl" transport = _Transport() result = await Topic59DayCardProjector(ledger_path).publish( _request(date(2026, 8, 22), Topic59DayState.SUBMITTED), transport ) rows = [_LedgerRow.model_validate_json(line) for line in ledger_path.read_text(encoding="utf-8").splitlines()] headings = tuple(line.split(":", 1)[0] for line in transport.sent[0]["text"].splitlines()) forbidden = {"홍길동", "70.3", "2300", "kcal", "체중", "칼로리", "식사", "증상", "@"} assert result.disposition is Topic59PublicationDisposition.SENT assert headings == ("[운영 현황]", "고객 식별자", "KST 날짜", "제출 상태", "완료 필드 수", "초안 상태", "승인 상태", "발송 상태", "이벤트 ID", "상태 행 ID") assert all(set(row.payload) == _SAFE_KEYS for row in rows) assert all(value not in transport.sent[0]["text"] for value in forbidden) assert "승인 버튼" not in transport.sent[0]["text"] def test_sidecar_extra_health_fields_cannot_cross_the_topic59_whitelist() -> None: leaky = _LeakySidecarRow( _CUSTOMER_IDENTITY, date(2026, 8, 23), "submitted", "event-submitted", _digest("row"), _digest("digest"), "홍길동", "70.3kg", "2300kcal 식사 기록", ) card = Topic59DayCard.from_sidecar( leaky, completed_field_count=5, draft_state=Topic59DraftState.DRAFTED, approval_state=Topic59ApprovalState.PENDING, delivery_state=Topic59DeliveryState.NOT_SENT, candidate_digest=_CANDIDATE_DIGEST, ) assert set(card.payload()) == _SAFE_KEYS assert all(value not in card.render() for value in {leaky.customer_name, leaky.raw_weight, leaky.meal_note})