"""Line-exact allocation ownership proof for weekly reminder bootstrap.""" from __future__ import annotations import ast import sys from pathlib import Path from types import FrameType, TracebackType from typing import Protocol, TypeAlias from gateway.platforms.telegram import TelegramAdapter from tests.gateway._nutrition_weekly_reminder_support import reminder_owner_fixture from tests.gateway.test_nutrition_weekly_reminder_bootstrap_boundary_r7 import ( platform_config, ) _FACTORY_NAMES = frozenset( { "adopt", "callback", "open_fd", "_open_root", "open_relative", "open_pinned_profile_registry", "verify_registered", "acquire_parent_authority_descriptor_owned", "open_authority_root_owned", "_open_child_owned", "_open_registered_files_owned", "acquire_unregistered_authority_owned", "open_registered_weekly_reminder_ledger_authority_owned", "construct_owned_customers", } ) _SOURCES = ( "gateway/platforms/nutrition_weekly_reminder_resources.py", "gateway/platforms/nutrition_weekly_operations_registry_identity.py", "gateway/platforms/nutrition_weekly_reminder_bootstrap.py", "gateway/platforms/nutrition_weekly_reminder_bootstrap_customers.py", "dualcoach/profile/checkin_cli/weekly_operations_parent_owned.py", "dualcoach/profile/checkin_cli/weekly_operations_authority_owned.py", "dualcoach/profile/checkin_cli/weekly_operations_customer_authority_owned.py", "dualcoach/profile/checkin_cli/weekly_reminder_ledger_open_owned.py", ) _TraceArgument: TypeAlias = ( None | int | str | tuple[type[BaseException], BaseException, TracebackType] ) class _Trace(Protocol): def __call__( self, frame: FrameType, event: str, argument: _TraceArgument, / ) -> _Trace | None: ... _REQUIRES_REGISTRAR = frozenset( { "_open_root", "open_relative", "open_pinned_profile_registry", "verify_registered", "acquire_parent_authority_descriptor_owned", "open_authority_root_owned", "_open_child_owned", "_open_registered_files_owned", "acquire_unregistered_authority_owned", "open_registered_weekly_reminder_ledger_authority_owned", "construct_owned_customers", } ) def _fds() -> frozenset[int]: return frozenset(int(entry.name) for entry in Path("/proc/self/fd").iterdir()) def _bytes(root: Path) -> dict[str, bytes]: return { str(path.relative_to(root)): path.read_bytes() for path in root.rglob("*") if path.is_file() } def _traceable(frame: FrameType) -> bool: return ( frame.f_code.co_name in _FACTORY_NAMES and any(frame.f_code.co_filename.endswith(source) for source in _SOURCES) ) def _executed_factory_lines(root: Path) -> tuple[tuple[str, str, int, int], ...]: root.mkdir() fixture = reminder_owner_fixture(root) config = platform_config(root, fixture.extra) lines: list[tuple[str, str, int, int]] = [] occurrences: dict[tuple[str, str, int], int] = {} def record( frame: FrameType, event: str, _argument: _TraceArgument ) -> _Trace | None: if event == "line" and _traceable(frame): key = (frame.f_code.co_filename, frame.f_code.co_name, frame.f_lineno) occurrence = occurrences.get(key, 0) + 1 occurrences[key] = occurrence lines.append((*key, occurrence)) return record try: sys.settrace(record) adapter = TelegramAdapter(config) sys.settrace(None) adapter.close_weekly_reminder_capabilities() finally: sys.settrace(None) fixture.owner.close() indexes = tuple(round(index * (len(lines) - 1) / 428) for index in range(429)) return tuple(lines[index] for index in indexes) def test_bootstrap_allocating_factories_require_registrar_and_no_post_return_adopt() -> None: found: set[str] = set() for relative in _SOURCES: source = Path(relative).read_text() tree = ast.parse(source) for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): if node.name in _REQUIRES_REGISTRAR: found.add(node.name) parameters = {argument.arg for argument in node.args.args} parameters.update(argument.arg for argument in node.args.kwonlyargs) assert "registrar" in parameters if node.name not in _FACTORY_NAMES: continue allocated: set[str] = set() for child in ast.walk(node): if ( isinstance(child, ast.Assign) and isinstance(child.value, ast.Call) ): allocated.update( target.id for target in child.targets if isinstance(target, ast.Name) ) if not isinstance(child, ast.Call): continue function = child.func if not ( isinstance(function, ast.Attribute) and function.attr in {"callback", "own"} and child.args ): continue close = child.args[0] if node.name in {"adopt", "open_fd", "callback"}: continue assert not ( isinstance(close, ast.Attribute) and close.attr == "close" and isinstance(close.value, ast.Call) ) assert not ( isinstance(close, ast.Attribute) and close.attr == "close" and isinstance(close.value, ast.Name) and close.value.id in allocated ) resources = ast.parse(Path(_SOURCES[0]).read_text()) open_fd = next( node for node in ast.walk(resources) if isinstance(node, ast.FunctionDef) and node.name == "open_fd" ) guarded = next(node for node in ast.walk(open_fd) if isinstance(node, ast.Try)) assert any( isinstance(handler.type, ast.Name) and handler.type.id == "BaseException" for handler in guarded.handlers ) calls = [ child for statement in guarded.body for child in ast.walk(statement) if isinstance(child, ast.Call) ] assert any(isinstance(call.func, ast.Name) and call.func.id == "opener" for call in calls) assert any( isinstance(call.func, ast.Attribute) and call.func.attr == "_register_guard" for call in calls ) def test_success_allocation_deltas_and_idempotent_reverse_close(tmp_path: Path) -> None: fixture = reminder_owner_fixture(tmp_path) config = platform_config(tmp_path, fixture.extra) baseline = _fds() observed: list[tuple[str, int]] = [] expected_events = { "nutricoach.weekly_reminder.registry_pinned", "nutricoach.weekly_reminder.authority_fd_registered", "nutricoach.weekly_reminder.parent_registered", "nutricoach.weekly_reminder.weekly_authority_registered", "nutricoach.weekly_reminder.source_registered", "nutricoach.weekly_reminder.ledger_registered", } active = [True] def audit(event: str, _arguments: tuple[str]) -> None: if active[0] and event in expected_events: observed.append((event, len(_fds()) - len(baseline))) sys.addaudithook(audit) try: adapter = TelegramAdapter(config) assert observed == [ ("nutricoach.weekly_reminder.registry_pinned", 2), ("nutricoach.weekly_reminder.authority_fd_registered", 3), ("nutricoach.weekly_reminder.parent_registered", 4), ("nutricoach.weekly_reminder.weekly_authority_registered", 5), ("nutricoach.weekly_reminder.source_registered", 11), ("nutricoach.weekly_reminder.ledger_registered", 15), ] adapter.close_weekly_reminder_capabilities() adapter.close_weekly_reminder_capabilities() assert _fds() == baseline finally: active[0] = False fixture.owner.close() def test_every_executed_allocation_factory_line_unwinds_exactly(tmp_path: Path) -> None: targets = _executed_factory_lines(tmp_path / "discovery") for target_index, target in enumerate(targets): for interruption in (KeyboardInterrupt, SystemExit): for attempt in (1, 2): root = tmp_path / f"case-{target_index}-{interruption.__name__}-{attempt}" root.mkdir() fixture = reminder_owner_fixture(root) config = platform_config(root, fixture.extra) baseline_fds = _fds() baseline_bytes = _bytes(root) hit = [False] occurrences: dict[tuple[str, str, int], int] = {} created: list[TelegramAdapter] = [] def interrupt( frame: FrameType, event: str, _argument: _TraceArgument ) -> _Trace | None: if event == "line" and _traceable(frame): key = ( frame.f_code.co_filename, frame.f_code.co_name, frame.f_lineno, ) occurrence = occurrences.get(key, 0) + 1 occurrences[key] = occurrence current = (*key, occurrence) if not hit[0] and current == target: hit[0] = True raise interruption return interrupt try: sys.settrace(interrupt) caught = False try: created.append(TelegramAdapter(config)) except interruption: caught = True finally: sys.settrace(None) assert hit[0] and caught, target assert created == [] assert _fds() == baseline_fds, target assert len(_fds()) == len(baseline_fds), target assert _bytes(root) == baseline_bytes, target finally: sys.settrace(None) fixture.owner.close()