"""Persistence contract for content-free Telegram check-in bindings.""" from __future__ import annotations import json import os import stat from dataclasses import replace from pathlib import Path from typing import cast import pytest from pytest import MonkeyPatch from gateway.platforms.physique_checkin_bindings import ( BindingStore, BindingStoreConflict, BindingStoreCorruption, BindingStoreError, CursorIdentity, IngressIdentity, IngressKind, ProjectionPhase, TelegramProjection, WizardBinding, ) _NOW = 1_900_000_000 _EXPIRES = 2_000_000_000 _PROJECTION_SESSION_ID = "0123456789abcdef0123456789abcdef" def _binding(*, session_id: str = "session-1") -> WizardBinding: return WizardBinding( session_id=session_id, owner_id="owner-1", chat_id="chat-1", topic_id="topic-1", step="calories", version=1, message_id="101", expires_at=_EXPIRES, ) def _legacy_payload() -> dict[str, object]: return { "version": 1, "active_session_id": "session-1", "bindings": [_binding().to_dict()], } def _write_private_json(path: Path, payload: object) -> str: raw = json.dumps(payload, separators=(",", ":"), sort_keys=True) assert path.write_text(raw, encoding="utf-8") == len(raw) os.chmod(path, 0o600) return raw def _projection( *, update_id: int = 41, session_id: str = _PROJECTION_SESSION_ID, source_step: str = "calories", source_version: int = 1, target_step: str = "macros", target_version: int = 2, phase: ProjectionPhase = ProjectionPhase.PREPARED, expires_at: int = _EXPIRES, receipt_message_id: int | None = None, ) -> TelegramProjection: return TelegramProjection( ingress=IngressIdentity( update_id=update_id, kind=IngressKind.CALLBACK, message_id=101, actor_id=123_456_789, chat_id=-100_123_456_789, topic_id=7, ), source=CursorIdentity(session_id, source_step, source_version), target=CursorIdentity(session_id, target_step, target_version), phase=phase, expires_at=expires_at, receipt_message_id=receipt_message_id, ) _SENTINEL_RAW_ANSWER = "SENTINEL_RAW_ANSWER" _CALLBACK_LIKE_CONTENT = "pc1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:calories:1:a0" def _replace_v2_identity_value( serialized: dict[str, object], identity_field: str, content: str, ) -> None: if identity_field == "receipt_message_id": serialized[identity_field] = content return container, field = identity_field.split(".", maxsplit=1) nested = cast(dict[str, object], serialized[container]) nested[field] = content _V2_IDENTITY_FIELDS = ( "ingress.update_id", "ingress.kind", "ingress.message_id", "ingress.actor_id", "ingress.chat_id", "ingress.topic_id", "source_cursor.session_id", "source_cursor.step", "source_cursor.version", "target_cursor.session_id", "target_cursor.step", "target_cursor.version", "receipt_message_id", ) def _content_as_int(content: str) -> int: return cast(int, cast(object, content)) def _construct_v2_identity_with_content(identity_field: str, content: str) -> object: if identity_field == "ingress.update_id": return IngressIdentity( _content_as_int(content), IngressKind.CALLBACK, 101, 123_456_789, -100_123_456_789, 7, ) if identity_field == "ingress.kind": return IngressIdentity( 41, cast(IngressKind, content), 101, 123_456_789, -100_123_456_789, 7, ) if identity_field == "ingress.message_id": return IngressIdentity( 41, IngressKind.CALLBACK, _content_as_int(content), 123_456_789, -100_123_456_789, 7, ) if identity_field == "ingress.actor_id": return IngressIdentity( 41, IngressKind.CALLBACK, 101, _content_as_int(content), -100_123_456_789, 7, ) if identity_field == "ingress.chat_id": return IngressIdentity( 41, IngressKind.CALLBACK, 101, 123_456_789, _content_as_int(content), 7, ) if identity_field == "ingress.topic_id": return IngressIdentity( 41, IngressKind.CALLBACK, 101, 123_456_789, -100_123_456_789, _content_as_int(content), ) if identity_field == "source_cursor.session_id": return CursorIdentity(content, "calories", 1) if identity_field == "source_cursor.step": return CursorIdentity(_PROJECTION_SESSION_ID, content, 1) if identity_field == "source_cursor.version": return CursorIdentity(_PROJECTION_SESSION_ID, "calories", _content_as_int(content)) if identity_field == "target_cursor.session_id": return CursorIdentity(content, "macros", 2) if identity_field == "target_cursor.step": return CursorIdentity(_PROJECTION_SESSION_ID, content, 2) if identity_field == "target_cursor.version": return CursorIdentity(_PROJECTION_SESSION_ID, "macros", _content_as_int(content)) if identity_field == "receipt_message_id": return _projection( phase=ProjectionPhase.DELIVERED, receipt_message_id=_content_as_int(content), ) raise AssertionError(f"unknown v2 identity field: {identity_field}") @pytest.mark.parametrize("identity_field", _V2_IDENTITY_FIELDS) @pytest.mark.parametrize("content", (_SENTINEL_RAW_ANSWER, _CALLBACK_LIKE_CONTENT)) def test_v2_identity_fields_reject_content_at_constructor_and_decode( identity_field: str, content: str, ) -> None: with pytest.raises(ValueError): _ = _construct_v2_identity_with_content(identity_field, content) serialized = _projection( phase=ProjectionPhase.DELIVERED, receipt_message_id=202, ).to_dict() _replace_v2_identity_value(serialized, identity_field, content) assert TelegramProjection.from_dict(serialized) is None @pytest.mark.parametrize("operation", ("load", "save")) def test_intermediate_directory_symlink_is_rejected_before_state_or_lock_mutation( tmp_path: Path, operation: str, ) -> None: external_parent = tmp_path / "external" / "private" external_parent.mkdir(parents=True) os.chmod(external_parent, 0o700) alias = tmp_path / "alias" alias.symlink_to(external_parent.parent, target_is_directory=True) store = BindingStore(alias / external_parent.name / "telegram-bindings.json") with pytest.raises(BindingStoreError): if operation == "load": _ = store.load("owner-1", "chat-1", "topic-1", _NOW) else: store.save({"session-1": _binding()}, "session-1") assert not tuple(external_parent.iterdir()) def test_parent_fsync_failure_surfaces_after_replace_and_cleans_temp( tmp_path: Path, monkeypatch: MonkeyPatch, ) -> None: path = tmp_path / "telegram-bindings.json" store = BindingStore(path) store.save({"session-1": _binding()}, "session-1") prior = path.read_bytes() original_fsync = os.fsync def fail_parent_fsync(descriptor: int) -> None: if stat.S_ISDIR(os.fstat(descriptor).st_mode): raise OSError("interrupted parent fsync") original_fsync(descriptor) monkeypatch.setattr(os, "fsync", fail_parent_fsync) with pytest.raises(BindingStoreError): store.save({"session-1": replace(_binding(), step="macros", version=2)}, "session-1") assert path.read_bytes() != prior assert not list(path.parent.glob("*.tmp")) @pytest.mark.parametrize( ("step", "awaiting_text", "accepted", "save_rejected"), ( ("launch", False, True, False), ("calories", False, False, True), ("launch", True, False, True), ("summary", False, False, True), ("macros", True, False, True), ), ) def test_empty_binding_message_is_limited_to_prebind_launch_state( tmp_path: Path, step: str, awaiting_text: bool, accepted: bool, save_rejected: bool, ) -> None: binding = replace( _binding(), step=step, message_id="", awaiting_text=awaiting_text, ) parsed = WizardBinding.from_dict(binding.to_dict()) if accepted: assert parsed == binding path = tmp_path / "telegram-bindings.json" store = BindingStore(path) store.save({binding.session_id: binding}, binding.session_id) persisted = cast(dict[str, object], json.loads(path.read_text(encoding="utf-8"))) record = cast(dict[str, object], cast(list[object], persisted["bindings"])[0]) assert record["step"] == "launch" assert record["message_id"] == "" assert record["awaiting_text"] is False else: assert parsed is None path = tmp_path / "telegram-bindings.json" store = BindingStore(path) if save_rejected: with pytest.raises(ValueError): store.save({binding.session_id: binding}, binding.session_id) else: store.save({binding.session_id: binding}, binding.session_id) assert not path.exists() def test_valid_v1_binding_load_is_lazy(tmp_path: Path) -> None: """A valid legacy private record remains byte-for-byte untouched when read.""" path = tmp_path / "telegram-bindings.json" expected = _write_private_json(path, _legacy_payload()) bindings, active_session_id = BindingStore(path).load( "owner-1", "chat-1", "topic-1", _NOW, ) assert active_session_id == "session-1" assert bindings["session-1"].step == "calories" assert path.read_text(encoding="utf-8") == expected def test_v1_read_then_real_mutation_writes_v2_and_preserves_binding(tmp_path: Path) -> None: path = tmp_path / "telegram-bindings.json" expected = _write_private_json(path, _legacy_payload()) store = BindingStore(path) bindings, active_session_id = store.load("owner-1", "chat-1", "topic-1", _NOW) assert path.read_text(encoding="utf-8") == expected store.save(bindings, active_session_id) persisted = cast(dict[str, object], json.loads(path.read_text(encoding="utf-8"))) assert persisted["version"] == 2 assert persisted["bindings"] == [_binding().to_dict()] assert persisted["projections"] == [] assert store.load("owner-1", "chat-1", "topic-1", _NOW)[0] == bindings def test_projection_schema_uses_typed_phase_ingress_and_cursor_identities(tmp_path: Path) -> None: projection = _projection() assert type(projection) is TelegramProjection assert type(projection.ingress) is IngressIdentity assert type(projection.source) is CursorIdentity assert type(projection.target) is CursorIdentity assert projection.ingress.kind is IngressKind.CALLBACK assert projection.phase is ProjectionPhase.PREPARED store = BindingStore(tmp_path / "telegram-bindings.json") assert store.record_projection(projection, now_epoch=_NOW) is True assert store.load_projections(_NOW) == (projection,) @pytest.mark.parametrize( ("from_phase", "to_phase", "receipt_message_id"), ( (ProjectionPhase.PREPARED, ProjectionPhase.DOMAIN_COMMITTED, None), (ProjectionPhase.DOMAIN_COMMITTED, ProjectionPhase.SEND_STARTED, None), (ProjectionPhase.DOMAIN_COMMITTED, ProjectionPhase.DELIVERY_FAILED, None), (ProjectionPhase.DOMAIN_COMMITTED, ProjectionPhase.DELIVERY_UNCERTAIN, None), (ProjectionPhase.SEND_STARTED, ProjectionPhase.DELIVERED, 202), (ProjectionPhase.SEND_STARTED, ProjectionPhase.DELIVERY_FAILED, None), (ProjectionPhase.SEND_STARTED, ProjectionPhase.DELIVERY_UNCERTAIN, None), ), ) def test_every_legal_projection_phase_transition_is_durable( tmp_path: Path, from_phase: ProjectionPhase, to_phase: ProjectionPhase, receipt_message_id: int | None, ) -> None: store = BindingStore(tmp_path / "telegram-bindings.json") projection = _projection(phase=from_phase) assert store.record_projection(projection, now_epoch=_NOW) is True advanced = replace( projection, phase=to_phase, receipt_message_id=receipt_message_id, ) assert store.record_projection(advanced, now_epoch=_NOW) is True assert store.load_projections(_NOW) == (advanced,) def test_duplicate_ingress_is_a_noop_but_conflicting_cursor_fails_closed(tmp_path: Path) -> None: store = BindingStore(tmp_path / "telegram-bindings.json") projection = _projection() assert store.record_projection(projection, now_epoch=_NOW) is True assert store.record_projection(projection, now_epoch=_NOW) is False with pytest.raises(BindingStoreConflict): _ = store.record_projection( replace(projection, target=replace(projection.target, step="meals")), now_epoch=_NOW, ) def test_only_one_nonterminal_transition_is_allowed_per_session(tmp_path: Path) -> None: store = BindingStore(tmp_path / "telegram-bindings.json") first = _projection() assert store.record_projection(first, now_epoch=_NOW) is True with pytest.raises(BindingStoreConflict): _ = store.record_projection( _projection( update_id=42, source_step="macros", source_version=2, target_step="meals", target_version=3, ), now_epoch=_NOW, ) def test_terminal_projection_is_retained_until_a_chained_projection_supersedes_it( tmp_path: Path, ) -> None: store = BindingStore(tmp_path / "telegram-bindings.json") first = _projection() assert store.record_projection(first, now_epoch=_NOW) is True committed = replace(first, phase=ProjectionPhase.DOMAIN_COMMITTED) assert store.record_projection(committed, now_epoch=_NOW) is True send_started = replace(committed, phase=ProjectionPhase.SEND_STARTED) assert store.record_projection(send_started, now_epoch=_NOW) is True delivered = replace(send_started, phase=ProjectionPhase.DELIVERED, receipt_message_id=202) assert store.record_projection(delivered, now_epoch=_NOW) is True next_projection = _projection( update_id=42, source_step="macros", source_version=2, target_step="meals", target_version=3, ) assert store.record_projection(next_projection, now_epoch=_NOW) is True assert store.load_projections(_NOW) == (delivered, next_projection) def test_expired_projection_does_not_block_a_fresh_session_transition(tmp_path: Path) -> None: store = BindingStore(tmp_path / "telegram-bindings.json") stale = _projection(expires_at=_NOW - 1) assert store.record_projection(stale, now_epoch=_NOW - 2) is True fresh = _projection( update_id=42, source_step="macros", source_version=2, target_step="meals", target_version=3, ) assert store.record_projection(fresh, now_epoch=_NOW) is True assert store.load_projections(_NOW) == (fresh,) def test_private_modes_and_atomic_fsync_replace_parent_fsync( tmp_path: Path, monkeypatch: MonkeyPatch, ) -> None: path = tmp_path / "private" / "telegram-bindings.json" lock_path = path.with_suffix(path.suffix + ".lock") store = BindingStore(path) fsync_calls: list[int] = [] replace_calls: list[tuple[str, str, int | None, int | None]] = [] original_fsync = os.fsync original_replace = os.replace def fsync_spy(descriptor: int) -> None: fsync_calls.append(descriptor) original_fsync(descriptor) def replace_spy( source: str, target: str, *, src_dir_fd: int | None = None, dst_dir_fd: int | None = None, ) -> None: replace_calls.append((source, target, src_dir_fd, dst_dir_fd)) original_replace(source, target, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) monkeypatch.setattr(os, "fsync", fsync_spy) monkeypatch.setattr(os, "replace", replace_spy) store.save({"session-1": _binding()}, "session-1") assert stat.S_IMODE(path.stat().st_mode) == 0o600 assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700 assert stat.S_IMODE(lock_path.stat().st_mode) == 0o600 assert len(fsync_calls) >= 2 # durable file, then durable directory entry assert len(replace_calls) == 1 source, target, source_fd, target_fd = replace_calls[0] assert source.endswith(".tmp") assert target == path.name assert source_fd == target_fd assert not list(path.parent.glob("*.tmp")) @pytest.mark.parametrize("boundary", ("fsync", "replace")) def test_interruption_before_replace_keeps_prior_state_and_cleans_temp( tmp_path: Path, monkeypatch: MonkeyPatch, boundary: str, ) -> None: path = tmp_path / "telegram-bindings.json" store = BindingStore(path) store.save({"session-1": _binding()}, "session-1") prior = path.read_bytes() if boundary == "fsync": def interrupted_fsync(_descriptor: int) -> None: raise OSError("interrupted durable write") monkeypatch.setattr(os, "fsync", interrupted_fsync) else: def interrupted_replace( _source: str, _target: str, *, src_dir_fd: int | None = None, dst_dir_fd: int | None = None, ) -> None: _ = (src_dir_fd, dst_dir_fd) raise OSError("interrupted replacement") monkeypatch.setattr(os, "replace", interrupted_replace) with pytest.raises(BindingStoreError): store.save({"session-1": replace(_binding(), step="macros", version=2)}, "session-1") assert path.read_bytes() == prior assert not list(path.parent.glob("*.tmp")) def test_symlinked_state_or_lock_is_rejected_without_following_it(tmp_path: Path) -> None: target = tmp_path / "target.json" assert target.write_text("{}", encoding="utf-8") == 2 path = tmp_path / "telegram-bindings.json" path.symlink_to(target) with pytest.raises(BindingStoreError): _ = BindingStore(path).load("owner-1", "chat-1", "topic-1", _NOW) assert target.read_text(encoding="utf-8") == "{}" path.unlink() lock = path.with_suffix(path.suffix + ".lock") lock.unlink() lock.symlink_to(target) with pytest.raises(BindingStoreError): BindingStore(path).save({"session-1": _binding()}, "session-1") assert target.read_text(encoding="utf-8") == "{}" @pytest.mark.parametrize( "payload", ( "{\"version\":1,\"bindings\":[", {"version": 1, "active_session_id": None, "bindings": [{"session_id": "only"}]}, {"version": 2, "active_session_id": None, "bindings": [], "projections": [{"phase": "prepared"}]}, {"version": 2, "active_session_id": None, "bindings": [], "projections": [], "unexpected": True}, ), ) def test_malformed_or_truncated_v1_v2_state_fails_closed( tmp_path: Path, payload: object, ) -> None: path = tmp_path / "telegram-bindings.json" if isinstance(payload, str): assert path.write_text(payload, encoding="utf-8") == len(payload) os.chmod(path, 0o600) else: _ = _write_private_json(path, payload) with pytest.raises(BindingStoreCorruption): _ = BindingStore(path).load("owner-1", "chat-1", "topic-1", _NOW) def test_conflicting_duplicate_update_identity_fails_closed(tmp_path: Path) -> None: store = BindingStore(tmp_path / "telegram-bindings.json") first = _projection() assert store.record_projection(first, now_epoch=_NOW) is True conflicting_ingress = IngressIdentity( update_id=first.ingress.update_id, kind=IngressKind.TEXT, message_id=102, actor_id=123_456_789, chat_id=-100_123_456_789, topic_id=7, ) with pytest.raises(BindingStoreConflict): _ = store.record_projection(replace(first, ingress=conflicting_ingress), now_epoch=_NOW) def test_v2_contains_only_content_free_binding_and_projection_keys(tmp_path: Path) -> None: path = tmp_path / "telegram-bindings.json" store = BindingStore(path) store.save({"session-1": _binding()}, "session-1") assert store.record_projection(_projection(), now_epoch=_NOW) is True payload: object = cast(object, json.loads(path.read_text(encoding="utf-8"))) keys: set[str] = set() def collect(value: object) -> None: record = cast(dict[str, object], value) if isinstance(value, dict) else None values = cast(list[object], value) if isinstance(value, list) else None if record is not None: keys.update(record) for child in record.values(): collect(child) elif values is not None: for child in values: collect(child) collect(payload) forbidden = { "answer", "answers", "raw_text", "text", "callback", "callback_data", "callback_payload", "payload", "customer_hash", "customer_value_hash", "value_hash", } assert not keys & forbidden