"""Regression: the author bootstrap wrapper's runtime-context helpers (#943). ``gitea_bootstrap_author_issue_worktree`` passed three values down to ``author_issue_bootstrap.bootstrap_author_issue_worktree``:: active_identity=_active_username(), active_profile=_active_profile_name(), owner_session=_current_session_id(), None of those three names was ever defined. Commit ``a942afe`` (#850) introduced the references and no definition, so every call — dry-run included — raised ``NameError: name '_active_username' is not defined`` while evaluating the arguments, before the bootstrap service was entered. The defect was unreachable until PR #942 (#941) wired the bootstrap scope into ``workflow_scope_guard``: until then ``verify_preflight_purity`` refused first with ``missing_issue_worktree``, so the guard fix is what exposed this. ``test_every_global_referenced_by_the_wrapper_resolves`` is the test that would have caught the original defect: it resolves every global name the wrapper's body references. Asserting only that the three known helpers now exist would not generalise to the next missing reference. """ from __future__ import annotations import ast import builtins import os import re import subprocess import tempfile import unittest from unittest import mock import author_issue_bootstrap as aib import create_issue_bootstrap as cib import gitea_mcp_server as gms import workflow_scope_guard BOOTSTRAP_TASK = "bootstrap_author_issue_worktree" WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree" RUNTIME_HELPERS = ("_active_username", "_active_profile_name", "_current_session_id") # "--", the shape the pre-existing lease call sites mint. SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+-\d+-[0-9a-f]{8}$") def _make_control_repo(tmp: str) -> tuple[str, str]: """Create a clean control checkout on master and return (path, head).""" repo = os.path.join(tmp, "repo") os.makedirs(os.path.join(repo, "branches")) subprocess.check_call( ["git", "init", "-b", "master", repo], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) subprocess.check_call( [ "git", "-C", repo, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "--allow-empty", "-m", "init", ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) head = subprocess.check_output( ["git", "-C", repo, "rev-parse", "HEAD"], text=True ).strip() return repo, head def _wrapper_ast() -> ast.FunctionDef: """Return the AST of the bootstrap wrapper as it exists on disk. Read from source rather than ``inspect``: the tool decorator may replace the callable, and the defect lived in the *source* argument expressions. """ path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "gitea_mcp_server.py") with open(path, encoding="utf-8") as fh: tree = ast.parse(fh.read()) for node in ast.walk(tree): if isinstance(node, ast.FunctionDef) and node.name == WRAPPER_NAME: return node raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py") class RuntimeHelperResolutionTests(unittest.TestCase): """AC: every runtime helper the wrapper references is defined and callable.""" def test_three_named_helpers_are_defined_and_callable(self): for name in RUNTIME_HELPERS: with self.subTest(helper=name): self.assertTrue( hasattr(gms, name), f"{name} is referenced but not defined" ) self.assertTrue(callable(getattr(gms, name)), f"{name} not callable") def test_helpers_accept_zero_arguments_as_called(self): """The wrapper calls each with no arguments; the signature must allow it.""" prior = gms._ACTIVE_SESSION_ID self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior) for name in RUNTIME_HELPERS: with self.subTest(helper=name): with mock.patch.object(gms, "get_profile", return_value={}), \ mock.patch.object( gms.session_ctx, "get_session_context", return_value=None): gms._ACTIVE_SESSION_ID = None getattr(gms, name)() # must not raise TypeError def test_every_global_referenced_by_the_wrapper_resolves(self): """The generalised form of this defect: an unresolvable global name. Collects every ``Name`` load in the wrapper body, subtracts locals (arguments, assignments, comprehension targets, imports), and asserts the remainder resolves against module globals or builtins. """ fn = _wrapper_ast() bound: set[str] = {a.arg for a in fn.args.args} bound |= {a.arg for a in fn.args.kwonlyargs} if fn.args.vararg: bound.add(fn.args.vararg.arg) if fn.args.kwarg: bound.add(fn.args.kwarg.arg) for node in ast.walk(fn): if isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)): bound.add(node.id) elif isinstance(node, (ast.Import, ast.ImportFrom)): for alias in node.names: bound.add((alias.asname or alias.name).split(".")[0]) elif isinstance(node, ast.ExceptHandler) and node.name: bound.add(node.name) unresolved = sorted( node.id for node in ast.walk(fn) if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load) and node.id not in bound and not hasattr(gms, node.id) and not hasattr(builtins, node.id) ) self.assertEqual( unresolved, [], f"{WRAPPER_NAME} references undefined globals: {unresolved}" ) def test_wrapper_still_passes_all_three_runtime_values(self): """Guard the wiring itself: the fix must not be 'stop passing them'.""" fn = _wrapper_ast() called = { node.func.id for node in ast.walk(fn) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) } for name in RUNTIME_HELPERS: with self.subTest(helper=name): self.assertIn(name, called) class ActiveUsernameTests(unittest.TestCase): """AC: identity comes from the authoritative pin, and fails closed.""" def test_returns_identity_from_bound_session_context(self): with mock.patch.object( gms.session_ctx, "get_session_context", return_value={"identity": "jcwalker3", "profile_name": "prgs-author"}, ): self.assertEqual(gms._active_username(), "jcwalker3") def test_unbound_context_returns_none_so_callers_fail_closed(self): with mock.patch.object( gms.session_ctx, "get_session_context", return_value=None ): self.assertIsNone(gms._active_username()) def test_blank_identity_is_not_treated_as_an_identity(self): for blank in ("", " ", None): with self.subTest(identity=blank): with mock.patch.object( gms.session_ctx, "get_session_context", return_value={"identity": blank}, ): self.assertIsNone(gms._active_username()) def test_identity_is_not_fabricated_from_the_profile(self): """A profile's expected username must never stand in for a real identity.""" with mock.patch.object( gms.session_ctx, "get_session_context", return_value={"identity": None, "expected_username": "jcwalker3"}, ): self.assertIsNone(gms._active_username()) class ActiveProfileNameTests(unittest.TestCase): """AC: profile name comes from the live profile, context only as fallback.""" def test_prefers_the_live_profile(self): with mock.patch.object( gms, "get_profile", return_value={"profile_name": "prgs-author"} ), mock.patch.object( gms.session_ctx, "get_session_context", return_value={"profile_name": "prgs-reviewer"}, ): self.assertEqual(gms._active_profile_name(), "prgs-author") def test_falls_back_to_session_context_when_profile_unreadable(self): with mock.patch.object(gms, "get_profile", side_effect=RuntimeError("no cfg")), \ mock.patch.object( gms.session_ctx, "get_session_context", return_value={"profile_name": "prgs-author"}): self.assertEqual(gms._active_profile_name(), "prgs-author") def test_returns_none_when_neither_source_knows(self): with mock.patch.object(gms, "get_profile", return_value={}), \ mock.patch.object( gms.session_ctx, "get_session_context", return_value=None): self.assertIsNone(gms._active_profile_name()) class CurrentSessionIdTests(unittest.TestCase): """AC: a real, stable session identifier — never a fresh owner per call.""" def setUp(self): self._prior = gms._ACTIVE_SESSION_ID gms._ACTIVE_SESSION_ID = None self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", self._prior) def test_shape_matches_the_existing_lease_call_sites(self): with mock.patch.object( gms, "get_profile", return_value={"profile_name": "prgs-author"} ): sid = gms._current_session_id() self.assertRegex(sid, SESSION_ID_RE) self.assertTrue(sid.startswith("prgs-author-")) self.assertIn(str(os.getpid()), sid) def test_stable_across_calls_within_one_process(self): """A new id per call would make lease-ownership checks unsatisfiable.""" with mock.patch.object( gms, "get_profile", return_value={"profile_name": "prgs-author"} ): first = gms._current_session_id() second = gms._current_session_id() third = gms._current_session_id() self.assertEqual(first, second) self.assertEqual(second, third) def test_returns_none_when_profile_undeterminable(self): with mock.patch.object(gms, "get_profile", return_value={}), \ mock.patch.object( gms.session_ctx, "get_session_context", return_value=None): self.assertIsNone(gms._current_session_id()) def test_none_result_is_not_memoised_as_a_session(self): with mock.patch.object(gms, "get_profile", return_value={}), \ mock.patch.object( gms.session_ctx, "get_session_context", return_value=None): self.assertIsNone(gms._current_session_id()) with mock.patch.object( gms, "get_profile", return_value={"profile_name": "prgs-author"} ): self.assertIsNotNone(gms._current_session_id()) class BootstrapServiceReachedTests(unittest.TestCase): """AC: the values the helpers produce carry a dry-run into the service.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.tmp = self._tmp.name self.repo, self.head = _make_control_repo(self.tmp) self.journals = os.path.join(self.tmp, "journals") os.makedirs(self.journals) def _bootstrap(self, **over): kwargs = dict( issue_number=943, canonical_repo_root=self.repo, expected_base_sha=self.head, branch_name="fix/issue-943-runtime-context-helpers", remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", active_identity="jcwalker3", active_profile="prgs-author", owner_session="prgs-author-4242-abcdef12", lock_dir=self.journals, idempotency_key="test-943", dry_run=True, ) kwargs.update(over) return aib.bootstrap_author_issue_worktree(**kwargs) def test_dry_run_succeeds_with_helper_produced_bindings(self): """Feed the service exactly what the live helpers return.""" prior = gms._ACTIVE_SESSION_ID self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior) with mock.patch.object( gms, "get_profile", return_value={"profile_name": "prgs-author"} ), mock.patch.object( gms.session_ctx, "get_session_context", return_value={"identity": "jcwalker3", "profile_name": "prgs-author"}, ): gms._ACTIVE_SESSION_ID = None identity = gms._active_username() profile = gms._active_profile_name() session = gms._current_session_id() res = self._bootstrap( active_identity=identity, active_profile=profile, owner_session=session ) self.assertTrue(res.get("success"), res) self.assertTrue(res.get("dry_run")) self.assertEqual(res.get("issue_number"), 943) self.assertEqual(res.get("base_sha"), self.head) def test_dry_run_creates_no_branch_worktree_or_lease(self): res = self._bootstrap() self.assertTrue(res.get("success"), res) branches = subprocess.check_output( ["git", "-C", self.repo, "branch", "--list"], text=True ) self.assertNotIn("issue-943", branches) worktrees = subprocess.check_output( ["git", "-C", self.repo, "worktree", "list"], text=True ) self.assertNotIn("issue-943", worktrees) self.assertFalse( os.path.exists(os.path.join(self.repo, "branches", "fix-issue-943-runtime-context-helpers")) ) journal = res.get("phase_journal") or {} self.assertFalse(journal.get("completed")) self.assertFalse(any((journal.get("artifacts_created") or {}).values())) self.assertIsNone(journal.get("lease_id")) self.assertIsNone(journal.get("assignment_id")) def test_missing_identity_fails_closed(self): res = self._bootstrap(active_identity=None) self.assertFalse(res.get("success")) self.assertEqual(res.get("reason_code"), "missing_active_identity") def test_missing_profile_fails_closed(self): res = self._bootstrap(active_profile=" ") self.assertFalse(res.get("success")) self.assertEqual(res.get("reason_code"), "missing_active_profile") def test_missing_session_fails_closed(self): res = self._bootstrap(owner_session=None) self.assertFalse(res.get("success")) self.assertEqual(res.get("reason_code"), "missing_owner_session") def test_expected_base_mismatch_fails_closed(self): res = self._bootstrap(expected_base_sha="0" * 40) self.assertFalse(res.get("success")) self.assertEqual(res.get("reason_code"), "stale_concurrency_pin") def test_unbound_runtime_context_cannot_reach_the_service(self): """With nothing bound, the helpers yield None and the service refuses.""" prior = gms._ACTIVE_SESSION_ID self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior) with mock.patch.object(gms, "get_profile", return_value={}), \ mock.patch.object( gms.session_ctx, "get_session_context", return_value=None): gms._ACTIVE_SESSION_ID = None res = self._bootstrap( active_identity=gms._active_username(), active_profile=gms._active_profile_name(), owner_session=gms._current_session_id(), ) self.assertFalse(res.get("success")) self.assertIn( res.get("reason_code"), {"missing_owner_session", "missing_active_identity", "missing_active_profile"}, ) def test_apply_reaches_the_intended_transition(self): res = self._bootstrap(dry_run=False) self.assertTrue(res.get("success"), res) self.assertNotEqual(res.get("dry_run"), True) branches = subprocess.check_output( ["git", "-C", self.repo, "branch", "--list"], text=True ) self.assertIn("issue-943", branches) self.assertTrue(os.path.isdir(res.get("worktree_path") or "")) class Issue941ScopeGuardNotRegressedTests(unittest.TestCase): """AC: PR #942's bootstrap-scope wiring still holds.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.repo, self.head = _make_control_repo(self._tmp.name) def _assessment(self, task: str = BOOTSTRAP_TASK) -> dict: return aib.assess_author_issue_bootstrap( workspace_path=self.repo, canonical_repo_root=self.repo, current_branch="master", head_sha=self.head, porcelain_status="", remote_master_sha=self.head, task=task, ) def test_bootstrap_task_still_permitted_from_clean_control_checkout(self): res = workflow_scope_guard.assess_root_source_mutation( workspace_path=self.repo, canonical_repo_root=self.repo, role_kind="author", mutation_task=BOOTSTRAP_TASK, porcelain_status="", bootstrap_assessment=self._assessment(), ) self.assertFalse(res.get("block"), res) self.assertNotEqual( res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE ) def test_bootstrap_task_still_blocked_without_evidence(self): res = workflow_scope_guard.assess_root_source_mutation( workspace_path=self.repo, canonical_repo_root=self.repo, role_kind="author", mutation_task=BOOTSTRAP_TASK, porcelain_status="", ) self.assertTrue(res.get("block")) self.assertEqual( res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE ) def test_ordinary_author_mutation_still_blocked_from_control_checkout(self): res = workflow_scope_guard.assess_root_source_mutation( workspace_path=self.repo, canonical_repo_root=self.repo, role_kind="author", mutation_task="commit_files", porcelain_status="", bootstrap_assessment=self._assessment(), ) self.assertTrue(res.get("block")) self.assertEqual( res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE ) def test_create_issue_bootstrap_unchanged(self): self.assertTrue(cib.is_create_issue_task("create_issue")) self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK)) if __name__ == "__main__": unittest.main()