...............................F.F...................................... [ 94%] .... [100%] =================================== FAILURES =================================== _________ test_claimed_customer_registers_disabled_without_other_role __________ tmp_path = PosixPath('/tmp/pytest-of-cube/pytest-8786/test_claimed_customer_register0') def test_claimed_customer_registers_disabled_without_other_role( tmp_path: Path, ) -> None: profile_root = tmp_path / "profile" registry_path = profile_root / "customers" / "registry.json" registry_path.parent.mkdir(parents=True) registry_path.write_text( json.dumps( { "version": 1, "owner": { "user_id": "12", "chat_id": "-100", "topic_id": "22", }, "customers": [], } ), encoding="utf-8", ) store = RoomBootstrapStore(profile_root / "bootstrap") prepared = store.prepare_rehearsal_customer_invite( draft(), bot_username="dualcoachtestbot", owner_id="12", ) claimed = store.claim_rehearsal_customer_invite( prepared.customer_link.rsplit("=", 1)[1], user_id="10", chat_id="10", message_id="100", ) result = TelegramCustomerBootstrapRegistration( profile_root, store, package_root=PACKAGE_ROOT, > ).handoff_rehearsal_customer(claimed) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ tests/gateway/test_telegram_customer_bootstrap.py:244: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ gateway/platforms/telegram_customer_bootstrap_registration.py:93: in handoff_rehearsal_customer return self._handoff_private_customer(live) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ gateway/platforms/telegram_customer_bootstrap_registration.py:99: in _handoff_private_customer admin, coaching = self._profile_modules() ^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = def _profile_modules(self) -> tuple[ModuleType, ModuleType]: package = self.package_root / "checkin_cli" if not package.is_dir() or not (package / "customer_admin.py").is_file(): raise BootstrapError("profile-local checkin_cli is unavailable") root_text = str(self.package_root) if root_text not in sys.path: sys.path.insert(0, root_text) admin = importlib.import_module("checkin_cli.customer_admin") coaching = importlib.import_module("checkin_cli.customer_coaching") for module in (admin, coaching): source = Path(str(module.__file__)).resolve() if not source.is_relative_to(package.resolve()): > raise BootstrapError("loaded checkin_cli is not profile-local") E gateway.platforms.telegram_customer_bootstrap.BootstrapError: loaded checkin_cli is not profile-local gateway/platforms/telegram_customer_bootstrap_registration.py:166: BootstrapError ______________ test_restart_recovery_registers_and_publishes_once ______________ self = async def _recover_room_bootstrap_waiting_states(self) -> None: """Reconcile every durable bootstrap card state after adapter restart.""" lock = getattr(self, "_room_bootstrap_recovery_lock", None) if lock is None: lock = asyncio.Lock() self._room_bootstrap_recovery_lock = lock async with lock: transport = self._get_room_bootstrap_transport() if transport is None: return try: sessions = transport.store.list_sessions() except (OSError, ValueError): logger.warning("room bootstrap recovery ledger is unavailable") return for session in sessions: state = str(getattr(getattr(session, "state", None), "value", "")) if state == "AWAITING_ACTIVATION": runtime = self._get_nutrition_onboarding_runtime() recover = getattr(runtime, "recover_waiting_session", None) if callable(recover): try: await recover(session) except (OSError, RuntimeError, TypeError, ValueError): logger.warning( "[%s] onboarding card recovery failed for bootstrap session", self.name, ) continue if ( state in {"REGISTERING", "AWAITING_CONSENT"} and getattr(session, "chat_id", None) is None ): from gateway.platforms.telegram_customer_bootstrap import Role claims = getattr(session, "role_claims", ()) owner_v1 = ( len(claims) == 1 and session.role_claim(Role.CUSTOMER) is not None ) if not owner_v1 or getattr(session, "recovery_attempts", ()): continue nutrition = self._get_nutrition_coaching() refresher = getattr(nutrition, "refresh_live_registry", None) if nutrition is None or not callable(refresher): continue try: current = session if state == "REGISTERING": from gateway.platforms.telegram_customer_bootstrap_registration import ( TelegramCustomerBootstrapRegistration, ) current = TelegramCustomerBootstrapRegistration( nutrition.profile_root, transport.store, > ).handoff_rehearsal_customer(session).session ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ gateway/platforms/telegram.py:5312: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ gateway/platforms/telegram_customer_bootstrap_registration.py:93: in handoff_rehearsal_customer return self._handoff_private_customer(live) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ gateway/platforms/telegram_customer_bootstrap_registration.py:99: in _handoff_private_customer admin, coaching = self._profile_modules() ^^^^^^^^^^^^^^^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = def _profile_modules(self) -> tuple[ModuleType, ModuleType]: package = self.package_root / "checkin_cli" if not package.is_dir() or not (package / "customer_admin.py").is_file(): raise BootstrapError("profile-local checkin_cli is unavailable") root_text = str(self.package_root) if root_text not in sys.path: sys.path.insert(0, root_text) admin = importlib.import_module("checkin_cli.customer_admin") coaching = importlib.import_module("checkin_cli.customer_coaching") for module in (admin, coaching): source = Path(str(module.__file__)).resolve() if not source.is_relative_to(package.resolve()): > raise BootstrapError("loaded checkin_cli is not profile-local") E gateway.platforms.telegram_customer_bootstrap.BootstrapError: loaded checkin_cli is not profile-local gateway/platforms/telegram_customer_bootstrap_registration.py:166: BootstrapError During handling of the above exception, another exception occurred: tmp_path = PosixPath('/tmp/pytest-of-cube/pytest-8786/test_restart_recovery_register0') monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7a17da0b57c0> def test_restart_recovery_registers_and_publishes_once( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile_root = tmp_path / "profile" profile_root.mkdir(mode=0o700) registry_path = profile_root / "customers" / "registry.json" registry_path.parent.mkdir(mode=0o700) registry_path.write_text( json.dumps( { "version": 1, "owner": { "user_id": "12", "chat_id": "-100", "topic_id": "22", }, "customers": [], } ), encoding="utf-8", ) registry_path.chmod(0o600) monkeypatch.setenv("DUALCOACH_PROFILE_PACKAGE", str(PACKAGE_ROOT)) store = RoomBootstrapStore( room_bootstrap_state_dir(profile_root), now=lambda: datetime(2026, 8, 14, tzinfo=timezone.utc), ) prepared = store.prepare_rehearsal_customer_invite( draft(), bot_username="dualcoachtestbot", owner_id="12", ) claimed = store.claim_rehearsal_customer_invite( prepared.customer_link.rsplit("=", 1)[1], user_id="10", chat_id="10", message_id="100", ) transport = RoomBootstrapTransport(store, owner_id="12") nutrition = SimpleNamespace( profile_root=profile_root, owner=SimpleNamespace(user_id="12"), refresh_live_registry=Mock(return_value=True), resolve=Mock(return_value=None), open_customer_onboarding=Mock( return_value=SimpleNamespace(text="consent", buttons=()) ), ) adapter = object.__new__(TelegramAdapter) adapter._room_bootstrap_transport = transport adapter._get_nutrition_coaching = lambda: nutrition adapter._get_nutrition_onboarding_runtime = lambda: None adapter._nutrition_onboarding_markup = lambda _card: None adapter._send_nutrition_topic = AsyncMock( return_value=SimpleNamespace(message_id=500) ) > asyncio.run(adapter._recover_room_bootstrap_waiting_states()) tests/gateway/test_telegram_customer_bootstrap.py:343: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ ../../../miniconda3/lib/python3.12/asyncio/runners.py:195: in run return runner.run(main) ^^^^^^^^^^^^^^^^ ../../../miniconda3/lib/python3.12/asyncio/runners.py:118: in run return self._loop.run_until_complete(task) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ../../../miniconda3/lib/python3.12/asyncio/base_events.py:691: in run_until_complete return future.result() ^^^^^^^^^^^^^^^ gateway/platforms/telegram.py:5323: in _recover_room_bootstrap_waiting_states self.name, ^^^^^^^^^ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ self = @property def name(self) -> str: """Human-readable name for this adapter.""" > return self.platform.value.title() ^^^^^^^^^^^^^ E AttributeError: 'TelegramAdapter' object has no attribute 'platform' gateway/platforms/base.py:2237: AttributeError =========================== short test summary info ============================ FAILED tests/gateway/test_telegram_customer_bootstrap.py::test_claimed_customer_registers_disabled_without_other_role FAILED tests/gateway/test_telegram_customer_bootstrap.py::test_restart_recovery_registers_and_publishes_once 2 failed, 74 passed in 1.24s