Merge pull request 'fix(author bootstrap): restore missing runtime identity and session helpers (Closes #943)' (#944) from fix/issue-943-runtime-context-helpers into master
This commit was merged in pull request #944.
This commit is contained in:
@@ -0,0 +1,747 @@
|
||||
"""Regression: author bootstrap runtime authority and session ownership (#943).
|
||||
|
||||
Two rounds of defects live here.
|
||||
|
||||
**Round 1 (#943 as filed).** ``gitea_bootstrap_author_issue_worktree`` passed
|
||||
four values down to the bootstrap service that were never defined:
|
||||
``_active_username``, ``_active_profile_name``, ``_current_session_id`` and
|
||||
``_author_mutation_block``. Every call — dry-run included — raised
|
||||
``NameError`` while evaluating the arguments, before the service was entered.
|
||||
|
||||
**Round 2 (review 622 on PR #944).** The first fix defined all four but made
|
||||
``_current_session_id`` mint ``<profile>-<pid>-<hex>`` once per process. The MCP
|
||||
daemon outlives every task it serves, so that value conflates sequential author
|
||||
tasks and can never equal the control-plane session that owns an
|
||||
allocator-created lease: ``_verify_assignment_and_lease_ids`` refused the whole
|
||||
allocated path with ``lease_session_mismatch``. The reviewed round also read the
|
||||
identity from the pinned session context while reading the profile from the live
|
||||
profile, so a rebind could produce a mixed claimant pair, and it swallowed every
|
||||
``get_profile()`` exception.
|
||||
|
||||
These tests therefore drive real state, not mocks of internals: a temporary
|
||||
control-plane SQLite database and a temporary issue-lock directory, both
|
||||
redirected through the same environment variables production uses
|
||||
(``GITEA_CONTROL_PLANE_DB``, ``GITEA_ISSUE_LOCK_DIR``). The ownership gate that
|
||||
runs is the real one.
|
||||
|
||||
``test_every_global_referenced_by_the_wrapper_resolves`` remains: it is what
|
||||
found ``_author_mutation_block``, and it generalises to the next missing
|
||||
reference. It supplements the runtime coverage below rather than standing in for
|
||||
it.
|
||||
"""
|
||||
|
||||
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 control_plane_db
|
||||
import create_issue_bootstrap as cib
|
||||
import gitea_mcp_server as gms
|
||||
import issue_lock_store
|
||||
import workflow_scope_guard
|
||||
|
||||
BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
|
||||
WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree"
|
||||
RUNTIME_HELPERS = (
|
||||
"_active_mutation_authority",
|
||||
"_active_username",
|
||||
"_active_profile_name",
|
||||
"_resolve_owner_workflow_session",
|
||||
"_author_mutation_block",
|
||||
)
|
||||
ORG = "Scaled-Tech-Consulting"
|
||||
REPO = "Gitea-Tools"
|
||||
IDENTITY = "jcwalker3"
|
||||
PROFILE = "prgs-author"
|
||||
|
||||
# A per-task ownership key must carry no process identifier (#790).
|
||||
TASK_KEY_RE = re.compile(r"^author_issue_work-[0-9a-f]{16}$")
|
||||
|
||||
|
||||
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."""
|
||||
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 _IsolatedControlPlane(unittest.TestCase):
|
||||
"""Temp control-plane DB and temp issue-lock dir, via production env vars."""
|
||||
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
self.tmp = self._tmp.name
|
||||
self.db_path = os.path.join(self.tmp, "control-plane.sqlite3")
|
||||
self.lock_dir = os.path.join(self.tmp, "issue-locks")
|
||||
self.journals = os.path.join(self.tmp, "journals")
|
||||
os.makedirs(self.lock_dir)
|
||||
os.makedirs(self.journals)
|
||||
env = mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
control_plane_db.DB_PATH_ENV: self.db_path,
|
||||
issue_lock_store.LOCK_DIR_ENV: self.lock_dir,
|
||||
},
|
||||
)
|
||||
env.start()
|
||||
self.addCleanup(env.stop)
|
||||
self.db = control_plane_db.ControlPlaneDB(self.db_path)
|
||||
|
||||
def _allocate(self, session_id: str, *, issue: int = 943):
|
||||
"""Create a real assignment + lease owned by *session_id*."""
|
||||
self.db.upsert_session(
|
||||
session_id=session_id, role="author", profile=PROFILE, pid=os.getpid()
|
||||
)
|
||||
res = self.db.assign_and_lease(
|
||||
session_id=session_id, role="author", remote="prgs",
|
||||
org=ORG, repo=REPO, kind="issue", number=issue,
|
||||
)
|
||||
self.assertEqual(res.outcome, "assigned", res)
|
||||
return res.assignment_id, res.lease_id
|
||||
|
||||
def _authority(self):
|
||||
"""A resolved authority pair, as the wrapper would compute it."""
|
||||
return {"ok": True, "identity": IDENTITY, "profile_name": PROFILE}
|
||||
|
||||
def _resolve_session(self, **over):
|
||||
kwargs = dict(
|
||||
issue_number=943,
|
||||
assignment_id=None,
|
||||
lease_id=None,
|
||||
session_id=None,
|
||||
identity=IDENTITY,
|
||||
profile_name=PROFILE,
|
||||
remote="prgs",
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
)
|
||||
kwargs.update(over)
|
||||
return gms._resolve_owner_workflow_session(**kwargs)
|
||||
|
||||
|
||||
class OwnershipGateTests(_IsolatedControlPlane):
|
||||
"""B2: the allocator-driven ownership path, against a real control plane."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.repo, self.head = _make_control_repo(self.tmp)
|
||||
|
||||
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=ORG,
|
||||
repo=REPO,
|
||||
active_identity=IDENTITY,
|
||||
active_profile=PROFILE,
|
||||
lock_dir=self.journals,
|
||||
idempotency_key="test-943",
|
||||
dry_run=True,
|
||||
)
|
||||
kwargs.update(over)
|
||||
return aib.bootstrap_author_issue_worktree(**kwargs)
|
||||
|
||||
def test_true_owning_session_passes_the_ownership_gate(self):
|
||||
"""The canonical owner reaches and completes the service."""
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
|
||||
)
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertTrue(res.get("dry_run"))
|
||||
self.assertEqual(res.get("base_sha"), self.head)
|
||||
|
||||
def test_different_session_is_refused(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id,
|
||||
lease_id=lease_id,
|
||||
owner_session="prgs-author-task-b",
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
|
||||
|
||||
def test_process_derived_session_would_be_refused(self):
|
||||
"""The reviewed round-1 value shape can never own an allocated lease."""
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
round_one_value = f"{PROFILE}-{os.getpid()}-deadbeef"
|
||||
self.assertNotEqual(round_one_value, session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id,
|
||||
lease_id=lease_id,
|
||||
owner_session=round_one_value,
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
|
||||
|
||||
def test_unknown_lease_fails_closed(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, _ = self._allocate(session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id,
|
||||
lease_id="lease-does-not-exist",
|
||||
owner_session=session,
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "unknown_lease_id")
|
||||
|
||||
def test_released_lease_fails_closed(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
self.db.release_lease(lease_id, session_id=session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "lease_not_live")
|
||||
|
||||
def test_force_expired_lease_fails_closed(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
self.db.force_expire_lease(lease_id, reason="test")
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "lease_not_live")
|
||||
|
||||
def test_replacement_lease_does_not_inherit_prior_ownership(self):
|
||||
"""A second task's lease is not ownable by the first task's session."""
|
||||
first = "prgs-author-task-a"
|
||||
assignment_a, lease_a = self._allocate(first)
|
||||
self.db.release_lease(lease_a, session_id=first)
|
||||
second = "prgs-author-task-b"
|
||||
assignment_b, lease_b = self._allocate(second)
|
||||
self.assertNotEqual(lease_a, lease_b)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_b, lease_id=lease_b, owner_session=first
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "lease_session_mismatch")
|
||||
|
||||
def test_assignment_lease_identifier_mismatch_fails_closed(self):
|
||||
session = "prgs-author-task-a"
|
||||
_, lease_id = self._allocate(session)
|
||||
res = self._bootstrap(
|
||||
assignment_id="asn-not-the-recorded-one",
|
||||
lease_id=lease_id,
|
||||
owner_session=session,
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "assignment_lease_mismatch")
|
||||
|
||||
def test_lease_id_without_assignment_id_fails_closed(self):
|
||||
session = "prgs-author-task-a"
|
||||
_, lease_id = self._allocate(session)
|
||||
res = self._bootstrap(lease_id=lease_id, owner_session=session)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "incomplete_assignment_lease_ids")
|
||||
|
||||
def test_dry_run_with_valid_allocator_bindings_leaves_no_durable_state(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id, lease_id=lease_id, owner_session=session
|
||||
)
|
||||
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()))
|
||||
# The dry run must not have created an issue lock in the isolated dir.
|
||||
self.assertEqual(os.listdir(self.lock_dir), [])
|
||||
|
||||
def test_apply_reaches_the_intended_transition_with_valid_bindings(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
res = self._bootstrap(
|
||||
assignment_id=assignment_id,
|
||||
lease_id=lease_id,
|
||||
owner_session=session,
|
||||
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 WorkflowSessionResolutionTests(_IsolatedControlPlane):
|
||||
"""B1: the wrapper resolves the owning session, never a process identifier."""
|
||||
|
||||
def test_declared_session_is_verified_against_the_control_plane(self):
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
res = self._resolve_session(
|
||||
session_id=session, assignment_id=assignment_id, lease_id=lease_id
|
||||
)
|
||||
self.assertTrue(res.get("ok"), res)
|
||||
self.assertEqual(res.get("session_id"), session)
|
||||
self.assertEqual(res.get("session_source"), "declared")
|
||||
|
||||
def test_unknown_declared_session_is_refused_not_trusted(self):
|
||||
res = self._resolve_session(session_id="prgs-author-not-a-session")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
|
||||
|
||||
def test_declared_session_for_another_role_is_refused(self):
|
||||
self.db.upsert_session(
|
||||
session_id="prgs-reviewer-x", role="reviewer", profile="prgs-reviewer",
|
||||
pid=os.getpid(),
|
||||
)
|
||||
res = self._resolve_session(session_id="prgs-reviewer-x")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
|
||||
|
||||
def test_declared_session_for_another_profile_is_refused(self):
|
||||
self.db.upsert_session(
|
||||
session_id="other-profile-session", role="author",
|
||||
profile="prgs-controller", pid=os.getpid(),
|
||||
)
|
||||
res = self._resolve_session(session_id="other-profile-session")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "workflow_session_unverified")
|
||||
|
||||
def test_allocated_work_without_a_session_is_refused(self):
|
||||
"""Supplying a lease is not itself evidence of ownership."""
|
||||
session = "prgs-author-task-a"
|
||||
assignment_id, lease_id = self._allocate(session)
|
||||
res = self._resolve_session(assignment_id=assignment_id, lease_id=lease_id)
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(
|
||||
res.get("reason_code"), "workflow_session_required_for_allocated_work"
|
||||
)
|
||||
|
||||
def test_existing_issue_lock_supplies_its_per_task_session(self):
|
||||
lock_session = issue_lock_store.mint_task_session_id(
|
||||
issue_lock_store.AUTHOR_ISSUE_WORK_LEASE
|
||||
)
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote="prgs", org=ORG, repo=REPO, issue_number=943,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
issue_lock_store.write_lock_file(
|
||||
path,
|
||||
{
|
||||
"issue_number": 943,
|
||||
"branch_name": "fix/issue-943-runtime-context-helpers",
|
||||
"work_lease": {
|
||||
"task_session_id": lock_session,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
},
|
||||
},
|
||||
) if hasattr(issue_lock_store, "write_lock_file") else _write_json(
|
||||
path,
|
||||
{
|
||||
"issue_number": 943,
|
||||
"branch_name": "fix/issue-943-runtime-context-helpers",
|
||||
"work_lease": {
|
||||
"task_session_id": lock_session,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
},
|
||||
},
|
||||
)
|
||||
res = self._resolve_session()
|
||||
self.assertTrue(res.get("ok"), res)
|
||||
self.assertEqual(res.get("session_id"), lock_session)
|
||||
self.assertEqual(res.get("session_source"), "issue_lock")
|
||||
|
||||
def test_issue_lock_owned_by_another_identity_is_refused(self):
|
||||
path = issue_lock_store.lock_file_path(
|
||||
remote="prgs", org=ORG, repo=REPO, issue_number=943,
|
||||
lock_dir=self.lock_dir,
|
||||
)
|
||||
_write_json(
|
||||
path,
|
||||
{
|
||||
"issue_number": 943,
|
||||
"work_lease": {
|
||||
"task_session_id": "author_issue_work-" + "0" * 16,
|
||||
"claimant": {"username": "someone-else", "profile": PROFILE},
|
||||
},
|
||||
},
|
||||
)
|
||||
res = self._resolve_session()
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "issue_lock_owner_mismatch")
|
||||
|
||||
def test_unallocated_bootstrap_mints_a_per_task_key(self):
|
||||
res = self._resolve_session()
|
||||
self.assertTrue(res.get("ok"), res)
|
||||
self.assertEqual(res.get("session_source"), "minted_task_key")
|
||||
self.assertRegex(res["session_id"], TASK_KEY_RE)
|
||||
|
||||
def test_minted_key_contains_no_process_identifier(self):
|
||||
res = self._resolve_session()
|
||||
self.assertNotIn(str(os.getpid()), res["session_id"])
|
||||
self.assertNotIn(PROFILE, res["session_id"])
|
||||
|
||||
def test_sequential_tasks_on_one_daemon_do_not_share_ownership(self):
|
||||
"""The round-1 defect: one identifier per process for every task."""
|
||||
first = self._resolve_session()["session_id"]
|
||||
second = self._resolve_session()["session_id"]
|
||||
third = self._resolve_session()["session_id"]
|
||||
self.assertNotEqual(first, second)
|
||||
self.assertNotEqual(second, third)
|
||||
self.assertEqual(len({first, second, third}), 3)
|
||||
|
||||
def test_no_process_lifetime_cache_remains(self):
|
||||
self.assertFalse(hasattr(gms, "_ACTIVE_SESSION_ID"))
|
||||
self.assertFalse(hasattr(gms, "_current_session_id"))
|
||||
|
||||
|
||||
class MutationAuthorityTests(unittest.TestCase):
|
||||
"""F3/F4: one coherent authority pair, drift detected, no silent fallback."""
|
||||
|
||||
def _ctx(self, **over):
|
||||
base = {"identity": IDENTITY, "profile_name": PROFILE}
|
||||
base.update(over)
|
||||
return base
|
||||
|
||||
def test_matching_live_and_pinned_authority_resolves(self):
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": PROFILE}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value=IDENTITY), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=self._ctx()):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertTrue(res.get("ok"), res)
|
||||
self.assertEqual(res["identity"], IDENTITY)
|
||||
self.assertEqual(res["profile_name"], PROFILE)
|
||||
|
||||
def test_identity_drift_fails_closed(self):
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": PROFILE}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value="someone-else"), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=self._ctx()):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "authority_identity_drift")
|
||||
self.assertEqual(res.get("expected"), IDENTITY)
|
||||
self.assertEqual(res.get("actual"), "someone-else")
|
||||
|
||||
def test_profile_drift_fails_closed(self):
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": "prgs-controller"}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value=IDENTITY), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=self._ctx()):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "authority_profile_drift")
|
||||
|
||||
def test_identity_and_profile_never_come_from_different_snapshots(self):
|
||||
"""Round 2's mixed pair: pinned identity plus live profile."""
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": "prgs-controller"}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value="new-identity"), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=self._ctx()):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertIsNone(gms._active_username("gitea.prgs.cc"))
|
||||
self.assertIsNone(gms._active_profile_name("gitea.prgs.cc"))
|
||||
|
||||
def test_unresolvable_profile_is_a_structured_refusal_not_a_fallback(self):
|
||||
"""F4: no bare-except fallback to a previously pinned profile name."""
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
side_effect=RuntimeError("profile disabled")), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value=IDENTITY), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=self._ctx()):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
|
||||
self.assertNotEqual(res.get("profile_name"), PROFILE)
|
||||
|
||||
def test_malformed_profile_without_name_fails_closed(self):
|
||||
with mock.patch.object(gms, "get_profile", return_value={}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value=IDENTITY), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=None):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "authority_profile_unresolved")
|
||||
|
||||
def test_unresolved_identity_fails_closed(self):
|
||||
for value in (None, "", " "):
|
||||
with self.subTest(identity=value):
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": PROFILE}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value=value), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=None):
|
||||
res = gms._active_mutation_authority("gitea.prgs.cc")
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(
|
||||
res.get("reason_code"), "authority_identity_unresolved"
|
||||
)
|
||||
|
||||
def test_missing_host_cannot_yield_an_identity(self):
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": PROFILE}), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=None):
|
||||
res = gms._active_mutation_authority(None)
|
||||
self.assertFalse(res.get("ok"))
|
||||
self.assertEqual(res.get("reason_code"), "authority_identity_unresolved")
|
||||
|
||||
def test_expected_username_is_never_substituted_for_authentication(self):
|
||||
with mock.patch.object(
|
||||
gms, "get_profile",
|
||||
return_value={"profile_name": PROFILE, "username": IDENTITY},
|
||||
), mock.patch.object(gms, "_authenticated_username", return_value=None), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value={"expected_username": IDENTITY}):
|
||||
self.assertIsNone(gms._active_username("gitea.prgs.cc"))
|
||||
|
||||
def test_accessors_share_one_snapshot(self):
|
||||
with mock.patch.object(gms, "get_profile",
|
||||
return_value={"profile_name": PROFILE}), \
|
||||
mock.patch.object(gms, "_authenticated_username",
|
||||
return_value=IDENTITY), \
|
||||
mock.patch.object(gms.session_ctx, "get_session_context",
|
||||
return_value=self._ctx()):
|
||||
self.assertEqual(gms._active_username("gitea.prgs.cc"), IDENTITY)
|
||||
self.assertEqual(gms._active_profile_name("gitea.prgs.cc"), PROFILE)
|
||||
|
||||
|
||||
class AuthorMutationBlockTests(unittest.TestCase):
|
||||
"""Preserved: the structured refusal shape review 622 confirmed correct."""
|
||||
|
||||
def test_matches_the_sibling_refusal_shape(self):
|
||||
res = gms._author_mutation_block(["stopped"])
|
||||
self.assertIs(res["success"], False)
|
||||
self.assertIs(res["performed"], False)
|
||||
self.assertEqual(res["outcome"], "REFUSED")
|
||||
self.assertEqual(res["reasons"], ["stopped"])
|
||||
|
||||
def test_carries_reason_code_and_transport_fields(self):
|
||||
res = gms._author_mutation_block(
|
||||
["nope"], reason_code="authority_identity_drift",
|
||||
retryable=False, transport_survives=True,
|
||||
expected="a", actual="b", issue_number=943,
|
||||
)
|
||||
self.assertEqual(res["reason_code"], "authority_identity_drift")
|
||||
self.assertIs(res["retryable"], False)
|
||||
self.assertIs(res["transport_survives"], True)
|
||||
self.assertEqual((res["expected"], res["actual"]), ("a", "b"))
|
||||
self.assertEqual(res["issue_number"], 943)
|
||||
self.assertIs(res["success"], False)
|
||||
|
||||
|
||||
class RuntimeHelperResolutionTests(unittest.TestCase):
|
||||
"""Every runtime helper the wrapper references is defined and callable.
|
||||
|
||||
Supplements the runtime coverage above; it does not replace it.
|
||||
"""
|
||||
|
||||
def test_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 not defined")
|
||||
self.assertTrue(callable(getattr(gms, name)))
|
||||
|
||||
def test_every_global_referenced_by_the_wrapper_resolves(self):
|
||||
"""The generalised form of the round-1 defect: an unresolvable global."""
|
||||
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_wires_the_authority_and_session_resolvers(self):
|
||||
fn = _wrapper_ast()
|
||||
called = {
|
||||
node.func.id
|
||||
for node in ast.walk(fn)
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
|
||||
}
|
||||
self.assertIn("_active_mutation_authority", called)
|
||||
self.assertIn("_resolve_owner_workflow_session", called)
|
||||
self.assertIn("_author_mutation_block", called)
|
||||
|
||||
def test_wrapper_accepts_an_optional_session_id(self):
|
||||
"""ABI addition stays backward compatible: optional, defaulting to None."""
|
||||
fn = _wrapper_ast()
|
||||
names = [a.arg for a in fn.args.args]
|
||||
self.assertIn("session_id", names)
|
||||
offset = len(names) - len(fn.args.defaults)
|
||||
default = fn.args.defaults[names.index("session_id") - offset]
|
||||
self.assertIsInstance(default, ast.Constant)
|
||||
self.assertIsNone(default.value)
|
||||
|
||||
|
||||
class Issue941ScopeGuardNotRegressedTests(unittest.TestCase):
|
||||
"""Preserved: 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))
|
||||
|
||||
|
||||
def _write_json(path: str, payload: dict) -> None:
|
||||
"""Write an issue-lock file directly, for lock-precedence tests."""
|
||||
import json
|
||||
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
with open(path, "w", encoding="utf-8") as fh:
|
||||
json.dump(payload, fh)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user