from __future__ import annotations from dataclasses import replace from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from pathlib import Path from threading import Barrier import pytest from gateway.platforms.telegram_customer_bootstrap import ( BootstrapError, BootstrapState, CustomerDraft, RoomBootstrapStore, ) def _draft() -> CustomerDraft: return CustomerDraft( customer_key="pilot_001", display_name="Pilot Customer", starts_on="2026-08-21", daily_time="08:00", weekly_weekday=0, monthly_day=1, calories_kcal=2200, protein_g=140, meals=("breakfast", "lunch", "dinner"), customer_user_id="10", ) def test_unclaimed_invite_expires_at_24_hour_boundary(tmp_path: Path) -> None: issued_at = datetime(2026, 8, 21, tzinfo=timezone.utc) current_time = [issued_at] store = RoomBootstrapStore( tmp_path / "bootstrap", now=lambda: current_time[0], ) prepared = store.prepare_rehearsal_customer_invite( _draft(), bot_username="dualcoachtestbot", owner_id="12", ) token = prepared.customer_link.rsplit("=", 1)[1] assert prepared.expires_at == issued_at + timedelta(hours=24) current_time[0] = issued_at + timedelta(hours=23, minutes=59, seconds=59) assert store.rehearsal_customer_invite_session(token) == prepared.session current_time[0] = issued_at + timedelta(hours=24) with pytest.raises(BootstrapError, match="unavailable"): _ = store.rehearsal_customer_invite_session(token) assert store.get(prepared.session.session_id).state is BootstrapState.EXPIRED def test_invite_claimed_before_24_hours_remains_single_use(tmp_path: Path) -> None: issued_at = datetime(2026, 8, 21, tzinfo=timezone.utc) current_time = [issued_at] store = RoomBootstrapStore( tmp_path / "bootstrap", now=lambda: current_time[0], ) prepared = store.prepare_rehearsal_customer_invite( _draft(), bot_username="dualcoachtestbot", owner_id="12", ) token = prepared.customer_link.rsplit("=", 1)[1] current_time[0] = issued_at + timedelta(hours=23, minutes=59) claimed = store.claim_rehearsal_customer_invite( token, user_id="10", chat_id="10", message_id="100", ) assert claimed.state is BootstrapState.REGISTERING with pytest.raises(BootstrapError, match="unavailable"): _ = store.claim_rehearsal_customer_invite( token, user_id="10", chat_id="10", message_id="101", ) def test_invite_rejects_a_forwarded_link_before_consuming_it( tmp_path: Path, ) -> None: store = RoomBootstrapStore(tmp_path / "bootstrap") prepared = store.prepare_rehearsal_customer_invite( _draft(), bot_username="dualcoachtestbot", owner_id="12", ) token = prepared.customer_link.rsplit("=", 1)[1] with pytest.raises(BootstrapError, match="unavailable"): _ = store.claim_rehearsal_customer_invite( token, user_id="11", chat_id="11", message_id="100", ) assert store.rehearsal_customer_invite_session(token).state is BootstrapState.PREPARED claimed = store.claim_rehearsal_customer_invite( token, user_id="10", chat_id="10", message_id="101", ) assert claimed.state is BootstrapState.REGISTERING def test_first_private_claimant_atomically_binds_unbound_invite( tmp_path: Path, ) -> None: # Given: an intake-only invite has no predeclared Telegram identity. store = RoomBootstrapStore(tmp_path / "bootstrap") draft = replace( _draft(), customer_user_id=None, ) prepared = store.prepare_rehearsal_customer_invite( draft, bot_username="dualcoachtestbot", owner_id="12", first_claim=True, ) token = prepared.customer_link.rsplit("=", 1)[1] # When: the first private-DM user claims it. claimed = store.claim_rehearsal_customer_invite( token, user_id="11", chat_id="11", message_id="100", ) # Then: that identity is durably bound and every replay is denied. assert claimed.state is BootstrapState.REGISTERING assert claimed.customer_draft.customer_user_id == "11" assert claimed.customer_draft_digest == claimed.customer_draft.digest with pytest.raises(BootstrapError, match="unavailable"): _ = store.claim_rehearsal_customer_invite( token, user_id="10", chat_id="10", message_id="101", ) def test_first_claim_rejects_owner_without_consuming_link( tmp_path: Path, ) -> None: store = RoomBootstrapStore(tmp_path / "bootstrap") prepared = store.prepare_rehearsal_customer_invite( replace(_draft(), customer_user_id=None), bot_username="dualcoachtestbot", owner_id="12", first_claim=True, ) token = prepared.customer_link.rsplit("=", 1)[1] with pytest.raises(BootstrapError, match="unavailable"): _ = store.claim_rehearsal_customer_invite( token, user_id="12", chat_id="12", message_id="100", ) claimed = store.claim_rehearsal_customer_invite( token, user_id="11", chat_id="11", message_id="101", ) assert claimed.customer_draft.customer_user_id == "11" def test_concurrent_first_claimants_have_exactly_one_winner( tmp_path: Path, ) -> None: store = RoomBootstrapStore(tmp_path / "bootstrap") prepared = store.prepare_rehearsal_customer_invite( replace(_draft(), customer_user_id=None), bot_username="dualcoachtestbot", owner_id="12", first_claim=True, ) token = prepared.customer_link.rsplit("=", 1)[1] barrier = Barrier(2) def claim(user_id: str) -> str | None: _ = barrier.wait(timeout=5) try: claimed = store.claim_rehearsal_customer_invite( token, user_id=user_id, chat_id=user_id, message_id=user_id, ) except BootstrapError: return None return claimed.role_claims[0].user_id with ThreadPoolExecutor(max_workers=2) as executor: results = tuple(executor.map(claim, ("10", "11"))) assert sum(result is not None for result in results) == 1 winner = next(result for result in results if result is not None) current = store.get(prepared.session.session_id) assert current.customer_draft.customer_user_id == winner assert current.role_claims[0].user_id == winner def test_store_rejects_intermediate_data_symlink(tmp_path: Path) -> None: # Given profile = tmp_path / "profile" profile.mkdir(mode=0o700) outside = tmp_path / "outside" outside.mkdir(mode=0o700) (profile / "data").symlink_to(outside, target_is_directory=True) # When / Then with pytest.raises(BootstrapError, match="directory chain"): _ = RoomBootstrapStore( profile / "data/onboarding/telegram-room-bootstrap-v1" ) assert not (outside / "onboarding").exists()