Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity
Files
c2fc2683b97e3ba5dc7eaf4ac9798af76bd08dfc
Gitea-Tools/tests/test_issue_702_review_findings_f1_f6.py
T
sysadminandClaude Fable 5 c2fc2683b9 fix(recovery): address PR #703 review findings F1-F6 (#702) 
Fixes the six defects from the formal REQUEST_CHANGES review at
889931d553:

- F1: run sanctioned stale-binding recovery in
  gitea_resolve_task_capability BEFORE the terminal launcher probe, so a
  worktree removed mid-session can no longer wedge mutation resolution on
  'missing cwd' before recovery runs. The fail-closed probe block now also
  carries the binding classification evidence.
- F2: _resolve_preflight_workspace_path resolves with the same
  verify_paths existence checks as the canonical mutation context, and
  _get_workspace_porcelain returns a synthetic tracked-dirty sentinel when
  the workspace is missing or git fails — an uninspectable workspace can
  never read as clean/empty porcelain.
- F3: the orphaned_expired_superseded_head cleanup matrix now requires
  affirmative safe worktree evidence (worktree_clean=true or
  worktree_exists=false), aligned with the sibling orphaned_owner_missing
  gate and the diagnosis path; unknown evidence fails closed.
- F4: reviewer-session-lease shadows and stale-binding audit records are
  recovery-critical kinds exempt from the 4h session-state TTL; they
  persist until a sanctioned clear terminally reconciles them.
- F5: session-lease shadows are keyed by lease session id (collision-safe
  identity) with a list_states enumeration API; concurrent same-profile
  sessions can no longer overwrite or misattribute each other's crash
  evidence. Legacy single-slot records remain readable.
- F6: the durable audit record is persisted (status=pending) BEFORE the
  environment binding is cleared; if audit persistence fails the clear
  does not happen and the failure is reported explicitly.

30 new regression tests cover deleted-worktree recovery ordering, missing/
deleted/symlinked/mid-evaluation path changes, unknown-evidence cleanup,
TTL boundary (before/at/after), interleaved and reconnected concurrent
sessions, and audit-write failure/retry/idempotence/ordering.

Issue #704 dotenv load-path prevention is intentionally NOT included.

Full suite: 2695 passed, 6 skipped, 161 subtests passed.

Co-Authored-By: Claude Fable 5 <[email protected]>
2026-07-13 16:40:58 -04:00

694 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Regression tests for the PR #703 REQUEST_CHANGES findings F1–F6 (#702).
Formal review at head 889931d independently reproduced six defects:
F1 — stale-binding recovery ran only AFTER the terminal launcher probe in
``gitea_resolve_task_capability``; a worktree removed mid-session wedged
every mutation-task resolution on 'missing cwd' before recovery.
F2 — ``_resolve_preflight_workspace_path`` skipped the existence checks the
canonical mutation context applies; a dead binding could produce empty
porcelain and read as a clean workspace.
F3 — ``orphaned_expired_superseded_head`` cleanup accepted unknown
``worktree_exists``/``worktree_clean`` evidence.
F4 — the 4-hour session-state TTL silently discarded recovery-critical
crash-orphan shadow evidence.
F5 — the single same-profile shadow slot let concurrent sessions overwrite
each other's crash evidence.
F6 — the environment was cleared BEFORE the audit record was persisted, and
audit-write failure was swallowed.
"""
from __future__ import annotations
import json
import os
import shutil
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import mcp_session_state as mss # noqa: E402
import namespace_workspace_binding as nwb # noqa: E402
import reviewer_pr_lease as leases # noqa: E402
import stale_binding_recovery as sbr # noqa: E402
import mcp_server # noqa: E402
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
OLD_HEAD = "b" * 40
NEW_HEAD = "6" * 40
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
"gitea.pr.create", "gitea.branch.push",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
},
"rules": {"allow_runtime_switching": False},
}
def _now() -> datetime:
return datetime.now(timezone.utc)
class _CapabilityHarness(unittest.TestCase):
"""Shared config/env plumbing for resolve-capability integration tests."""
def setUp(self):
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {
"host": "gitea.example.com",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
self._dir.cleanup()
def _env(self, **extra):
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": "prgs-author",
"GITEA_TOKEN_AUTHOR": "author-pass",
}
env.update(extra)
return env
def _deleted_worktree(self) -> str:
path = tempfile.mkdtemp(prefix="gitea-removed-worktree-")
shutil.rmtree(path)
return path
class TestF1RecoveryBeforeTerminalProbe(_CapabilityHarness):
"""F1: recovery must run before the terminal cwd probe can wedge."""
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_removed_worktree_recovers_before_probe(self, _auth, _api):
missing = self._deleted_worktree()
env = self._env(
GITEA_ACTIVE_WORKTREE=missing,
GITEA_FORCE_TERMINAL_PROBE="1",
GITEA_FORCE_STALE_BINDING_RECOVERY="1",
)
with patch.dict(os.environ, env):
res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs"
)
# The provably-stale binding was cleared before the probe ran.
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
self.assertFalse(res.get("terminal_launcher_unhealthy"))
report = res.get("stale_binding_recovery") or {}
self.assertEqual(
report.get("classification"),
sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
)
self.assertTrue(report.get("recovery_performed"))
self.assertTrue(res.get("allowed_in_current_session"))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_probe_block_still_carries_recovery_report(self, _auth, _api):
# Recovery disabled (test mode without the force flag) preserves the
# fail-closed probe block, but the block must now carry the
# classification evidence instead of hiding it.
missing = self._deleted_worktree()
env = self._env(
GITEA_ACTIVE_WORKTREE=missing,
GITEA_FORCE_TERMINAL_PROBE="1",
)
with patch.dict(os.environ, env):
res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs"
)
self.assertIn("GITEA_ACTIVE_WORKTREE", os.environ)
self.assertTrue(res.get("terminal_launcher_unhealthy"))
report = res.get("stale_binding_recovery") or {}
self.assertEqual(
report.get("classification"),
sbr.CLASSIFICATION_PROVABLY_STALE_MISSING_PATH,
)
self.assertFalse(report.get("recovery_performed"))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_unverified_binding_is_never_cleared_by_reordering(self, _auth, _api):
# F1 must not weaken authorization: an existing, uncorroborated
# binding survives resolution untouched even with recovery forced.
existing = tempfile.mkdtemp(prefix="gitea-existing-worktree-")
self.addCleanup(shutil.rmtree, existing, ignore_errors=True)
env = self._env(
GITEA_ACTIVE_WORKTREE=existing,
GITEA_FORCE_TERMINAL_PROBE="1",
GITEA_FORCE_STALE_BINDING_RECOVERY="1",
)
with patch.dict(os.environ, env):
res = mcp_server.gitea_resolve_task_capability(
task="create_pr", remote="prgs"
)
self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), existing)
report = res.get("stale_binding_recovery") or {}
self.assertFalse(report.get("recovery_performed"))
class TestF2PreflightPathVerification(_CapabilityHarness):
"""F2: preflight and mutation contexts verify resolved paths alike."""
def test_preflight_and_mutation_context_agree_on_dead_binding(self):
missing = self._deleted_worktree()
env = self._env(GITEA_ACTIVE_WORKTREE=missing)
with patch.dict(os.environ, env):
preflight = mcp_server._resolve_preflight_workspace_path(None)
mutation = mcp_server._resolve_namespace_mutation_context(None)
self.assertEqual(preflight, mutation["workspace_path"])
self.assertNotEqual(preflight, os.path.realpath(missing))
def test_missing_explicit_worktree_never_reads_clean(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._env()):
porcelain = mcp_server._get_workspace_porcelain(worktree_path=missing)
self.assertEqual(
porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
)
# The sentinel parses as a tracked dirty entry — never clean.
self.assertTrue(mcp_server._parse_porcelain_entries(porcelain))
def test_workspace_deleted_during_evaluation_never_reads_clean(self):
# TOCTOU: resolution succeeded, then the path vanished before git ran.
missing = self._deleted_worktree()
with patch.dict(os.environ, self._env()):
with patch.object(
mcp_server,
"_resolve_preflight_workspace_path",
return_value=missing,
):
porcelain = mcp_server._get_workspace_porcelain()
self.assertEqual(
porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
)
def test_git_failure_never_reads_clean(self):
class _Failed:
returncode = 128
stdout = ""
stderr = "fatal: not a git repository"
with patch.dict(os.environ, self._env()):
with patch.object(
mcp_server.subprocess, "run", return_value=_Failed()
):
porcelain = mcp_server._get_workspace_porcelain()
self.assertEqual(
porcelain, mcp_server.PREFLIGHT_WORKSPACE_UNAVAILABLE_PORCELAIN
)
def test_symlinked_binding_to_deleted_target_is_demoted(self):
target = tempfile.mkdtemp(prefix="gitea-symlink-target-")
holder = tempfile.mkdtemp(prefix="gitea-symlink-holder-")
self.addCleanup(shutil.rmtree, holder, ignore_errors=True)
link = os.path.join(holder, "worktree-link")
os.symlink(target, link)
shutil.rmtree(target)
demotions: list[str] = []
_path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root=os.getcwd(),
env={"GITEA_ACTIVE_WORKTREE": link},
demotions=demotions,
verify_paths=True,
)
self.assertEqual(source, "MCP server process root (default)")
self.assertEqual(len(demotions), 1)
def test_dead_binding_falls_through_to_live_role_binding(self):
# Path changes during evaluation: the dead candidate demotes at
# verification time and never resurrects over the live fallback.
alive = tempfile.mkdtemp(prefix="gitea-alive-worktree-")
self.addCleanup(shutil.rmtree, alive, ignore_errors=True)
missing = self._deleted_worktree()
path, source = nwb.resolve_namespace_workspace(
role_kind="reviewer",
process_project_root=os.getcwd(),
env={
"GITEA_ACTIVE_WORKTREE": missing,
"GITEA_REVIEWER_WORKTREE": alive,
},
verify_paths=True,
)
self.assertEqual(path, os.path.realpath(alive))
self.assertEqual(
source, "GITEA_REVIEWER_WORKTREE environment variable"
)
class TestF3AffirmativeWorktreeEvidence(unittest.TestCase):
"""F3: unknown worktree evidence must fail closed for crash orphans."""
def _comments(self) -> list[dict]:
stale = _now() - timedelta(hours=3)
body = leases.format_lease_body(
repo=REPO,
pr_number=701,
issue_number=699,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="65664-4f248d213d60",
worktree="branches/review-pr-654",
phase="validation-start",
candidate_head=OLD_HEAD,
target_branch="master",
target_branch_sha="2" * 40,
last_activity=stale,
expires_at=stale + timedelta(minutes=120),
)
return [{
"id": 11131,
"body": body,
"user": {"login": "sysadmin"},
"author": "sysadmin",
"created_at": stale.isoformat(),
"updated_at": stale.isoformat(),
}]
def _assess(self, **kwargs):
defaults = dict(
pr_number=701,
current_head_sha=NEW_HEAD,
formal_reviews=[],
repo=REPO,
expected_repo=REPO,
requesting_session_id="fresh-controller",
controller_recovery_authorized=True,
owner_process_alive=False,
)
defaults.update(kwargs)
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
self._comments(), **defaults
)
def test_unknown_worktree_evidence_fails_closed(self):
result = self._assess(worktree_exists=None, worktree_clean=None)
self.assertEqual(
result["classification"], "orphaned_expired_superseded_head"
)
self.assertFalse(result["cleanup_allowed"])
self.assertIn(
"affirmative safe evidence",
" ".join(result["fail_closed_reasons"]),
)
def test_unknown_cleanliness_with_existing_worktree_fails_closed(self):
result = self._assess(worktree_exists=True, worktree_clean=None)
self.assertFalse(result["cleanup_allowed"])
def test_affirmative_clean_worktree_is_eligible(self):
result = self._assess(worktree_exists=True, worktree_clean=True)
self.assertEqual(
result["classification"], "orphaned_expired_superseded_head"
)
self.assertTrue(result["cleanup_allowed"])
def test_affirmative_absent_worktree_is_eligible(self):
result = self._assess(worktree_exists=False, worktree_clean=None)
self.assertTrue(result["cleanup_allowed"])
def test_matrix_matches_diagnosis_gate(self):
# The cleanup matrix and the diagnosis path must agree: unknown
# evidence yields acquire-only, affirmative evidence yields cleanup.
diagnosis_unknown = leases.diagnose_reviewer_pr_lease_handoff(
self._comments(),
pr_number=701,
current_session_id=None,
current_reviewer_identity="fresh-reviewer",
current_head_sha=NEW_HEAD,
formal_reviews=[],
owner_process_alive=False,
worktree_exists=None,
worktree_clean=None,
)
assess_unknown = self._assess(worktree_exists=None, worktree_clean=None)
self.assertNotEqual(
diagnosis_unknown["next_action"],
"cleanup_obsolete_reviewer_comment_lease",
)
self.assertFalse(assess_unknown["cleanup_allowed"])
class TestF4CrashShadowDurability(unittest.TestCase):
"""F4: crash-orphan evidence survives the former 4h TTL boundary."""
SID = "82984-abcabcabcabc"
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self._env = patch.dict(
os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name}
)
self._env.start()
leases._SESSION_LEASE = None
def tearDown(self):
self._env.stop()
self._dir.cleanup()
leases._SESSION_LEASE = None
def _age_record(self, kind: str, hours: float, instance_id: str | None = None):
path = mss.state_file_path(kind=kind, instance_id=instance_id)
with open(path, encoding="utf-8") as fh:
envelope = json.load(fh)
stamp = (
(_now() - timedelta(hours=hours)).isoformat().replace("+00:00", "Z")
)
envelope["recorded_at"] = stamp
envelope["updated_at"] = stamp
envelope["payload"]["recorded_at"] = stamp
envelope["payload"]["updated_at"] = stamp
with open(path, "w", encoding="utf-8") as fh:
json.dump(envelope, fh)
def _save_shadow(self) -> None:
mss.save_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE,
instance_id=self.SID,
payload={"session_id": self.SID, "owner_pid": os.getpid()},
)
def _load_shadow(self):
return mss.load_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID
)
def test_shadow_loads_before_former_ttl_boundary(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=3.9, instance_id=self.SID
)
self.assertIsNotNone(self._load_shadow())
def test_shadow_loads_at_former_ttl_boundary(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=4.0, instance_id=self.SID
)
self.assertIsNotNone(self._load_shadow())
def test_shadow_loads_long_after_former_ttl_boundary(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
)
record = self._load_shadow()
self.assertIsNotNone(record)
self.assertEqual(record.get("session_id"), self.SID)
def test_stale_binding_audit_record_is_also_durable(self):
mss.save_state(
kind=mss.KIND_STALE_BINDING_RECOVERY,
payload={"cleared_env": "GITEA_ACTIVE_WORKTREE"},
)
self._age_record(mss.KIND_STALE_BINDING_RECOVERY, hours=27.0)
self.assertIsNotNone(
mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY)
)
def test_non_recovery_kinds_keep_ttl(self):
mss.save_state(
kind=mss.KIND_WORKFLOW_LOAD, payload={"workflow_hash": "abc"}
)
self._age_record(mss.KIND_WORKFLOW_LOAD, hours=5.0)
self.assertIsNone(mss.load_state(kind=mss.KIND_WORKFLOW_LOAD))
def test_sanctioned_clear_is_the_terminal_reconciliation(self):
self._save_shadow()
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
)
mss.clear_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE, instance_id=self.SID
)
self.assertIsNone(self._load_shadow())
self.assertEqual(
mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE), []
)
def test_crash_orphan_recovery_works_after_former_ttl(self):
# End to end: a crashed session's shadow still proves owner exit a
# day later.
leases.record_session_lease(
{
"pr_number": 701,
"session_id": self.SID,
"worktree": "branches/review-pr-654",
"candidate_head": OLD_HEAD,
"repo": REPO,
"comment_id": 11131,
"profile": "prgs-reviewer",
"reviewer_identity": "sysadmin",
"phase": "claimed",
},
lease_provenance={"source": "acquire", "comment_id": 11131},
)
leases._SESSION_LEASE = None # simulated crash: no sanctioned clear
self._age_record(
mss.KIND_REVIEWER_SESSION_LEASE, hours=27.0, instance_id=self.SID
)
shadow = leases.load_session_lease_shadow(session_id=self.SID)
self.assertIsNotNone(shadow)
verdict = leases.assess_crashed_session_lease_orphan(
shadow,
lease={"session_id": self.SID},
current_pid=os.getpid() + 1,
pid_alive=False,
)
self.assertTrue(verdict["orphan_evidence"])
self.assertIs(verdict["owner_process_alive"], False)
class TestF5ConcurrentSessionShadows(unittest.TestCase):
"""F5: concurrent same-profile sessions keep distinct shadow records."""
SID_A = "11111-aaaaaaaaaaaa"
SID_B = "22222-bbbbbbbbbbbb"
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self._env = patch.dict(
os.environ, {"GITEA_MCP_SESSION_STATE_DIR": self._dir.name}
)
self._env.start()
leases._SESSION_LEASE = None
def tearDown(self):
self._env.stop()
self._dir.cleanup()
leases._SESSION_LEASE = None
def _lease(self, session_id: str, pr_number: int, head: str) -> dict:
return {
"pr_number": pr_number,
"session_id": session_id,
"worktree": f"branches/review-pr-{pr_number}-{session_id[:5]}",
"candidate_head": head,
"repo": REPO,
"comment_id": 11221,
"profile": "prgs-reviewer",
"reviewer_identity": "sysadmin",
"phase": "claimed",
}
def _record(self, lease: dict) -> None:
leases.record_session_lease(
lease,
lease_provenance={"source": "acquire", "comment_id": 11221},
)
def test_interleaved_sessions_do_not_overwrite_each_other(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self._record(self._lease(self.SID_B, 701, NEW_HEAD))
# Session A heartbeats again after B wrote.
updated_a = self._lease(self.SID_A, 703, OLD_HEAD)
updated_a["phase"] = "validation-start"
self._record(updated_a)
shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A)
shadow_b = leases.load_session_lease_shadow(session_id=self.SID_B)
self.assertEqual(shadow_a.get("session_id"), self.SID_A)
self.assertEqual(shadow_a.get("pr_number"), 703)
self.assertEqual(shadow_a.get("phase"), "validation-start")
self.assertEqual(shadow_b.get("session_id"), self.SID_B)
self.assertEqual(shadow_b.get("pr_number"), 701)
self.assertEqual(shadow_b.get("candidate_head"), NEW_HEAD)
def test_shadow_lookup_never_misattributes(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self.assertIsNone(
leases.load_session_lease_shadow(session_id=self.SID_B)
)
verdict = leases.assess_crashed_session_lease_orphan(
leases.load_session_lease_shadow(session_id=self.SID_A),
lease={"session_id": self.SID_B},
pid_alive=False,
)
self.assertFalse(verdict["orphan_evidence"])
def test_clearing_one_session_preserves_the_other(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self._record(self._lease(self.SID_B, 701, NEW_HEAD))
# Sanctioned clear of B (the currently recorded in-memory lease).
leases.clear_session_lease()
self.assertIsNone(
leases.load_session_lease_shadow(session_id=self.SID_B)
)
shadow_a = leases.load_session_lease_shadow(session_id=self.SID_A)
self.assertIsNotNone(shadow_a)
self.assertEqual(shadow_a.get("session_id"), self.SID_A)
def test_reconnected_process_recovers_by_session_id(self):
self._record(self._lease(self.SID_A, 703, OLD_HEAD))
self._record(self._lease(self.SID_B, 701, NEW_HEAD))
# Simulated crash + reconnect: in-memory state is gone, durable
# records remain enumerable and individually attributable.
leases._SESSION_LEASE = None
records = mss.list_states(kind=mss.KIND_REVIEWER_SESSION_LEASE)
self.assertEqual(
sorted(r.get("session_id") for r in records),
sorted([self.SID_A, self.SID_B]),
)
shadow = leases.load_session_lease_shadow(session_id=self.SID_B)
self.assertEqual(shadow.get("pr_number"), 701)
def test_legacy_single_slot_record_still_resolves_by_session_id(self):
# Upgrade path: a record written by pre-F5 code (no instance key)
# remains usable when its payload names the requested session.
mss.save_state(
kind=mss.KIND_REVIEWER_SESSION_LEASE,
payload={"session_id": self.SID_A, "owner_pid": os.getpid()},
)
shadow = leases.load_session_lease_shadow(session_id=self.SID_A)
self.assertIsNotNone(shadow)
self.assertIsNone(
leases.load_session_lease_shadow(session_id=self.SID_B)
)
class TestF6AuditBeforeClear(_CapabilityHarness):
"""F6: the durable audit record is persisted before the env clears."""
def _recovery_env(self, missing: str) -> dict:
return self._env(
GITEA_ACTIVE_WORKTREE=missing,
GITEA_FORCE_STALE_BINDING_RECOVERY="1",
)
def test_audit_write_failure_blocks_the_clear(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._recovery_env(missing)):
with patch.object(
mss, "save_state", side_effect=OSError("disk full")
):
report = mcp_server._assess_stale_active_binding(
auto_recover=True
)
self.assertEqual(os.environ.get("GITEA_ACTIVE_WORKTREE"), missing)
self.assertFalse(report["recovery_performed"])
self.assertIs(report.get("audit_persisted"), False)
self.assertTrue(report.get("audit_write_failed"))
self.assertIn("NOT cleared", " ".join(report.get("reasons") or []))
def test_retry_after_audit_failure_recovers(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._recovery_env(missing)):
with patch.object(
mss, "save_state", side_effect=OSError("disk full")
):
first = mcp_server._assess_stale_active_binding(
auto_recover=True
)
self.assertFalse(first["recovery_performed"])
# Persistence recovered; the retry clears with a durable audit.
second = mcp_server._assess_stale_active_binding(auto_recover=True)
self.assertTrue(second["recovery_performed"])
self.assertIs(second.get("audit_persisted"), True)
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
audit = mss.load_state(kind=mss.KIND_STALE_BINDING_RECOVERY)
self.assertIsNotNone(audit)
self.assertEqual(audit.get("recovery_status"), "performed")
self.assertEqual(audit.get("cleared_value"), missing)
def test_audit_record_exists_before_env_clear(self):
missing = self._deleted_worktree()
observed: list[tuple[str | None, str | None]] = []
real_save = mss.save_state
def _spy(**kwargs):
observed.append(
(
(kwargs.get("payload") or {}).get("recovery_status"),
os.environ.get("GITEA_ACTIVE_WORKTREE"),
)
)
return real_save(**kwargs)
with patch.dict(os.environ, self._recovery_env(missing)):
with patch.object(mss, "save_state", side_effect=_spy):
report = mcp_server._assess_stale_active_binding(
auto_recover=True
)
self.assertTrue(report["recovery_performed"])
self.assertGreaterEqual(len(observed), 2)
pending_status, env_at_pending = observed[0]
performed_status, env_at_performed = observed[-1]
# Ordering proof: the binding was still present while the pending
# audit persisted, and gone by the time the performed status wrote.
self.assertEqual(pending_status, "pending")
self.assertEqual(env_at_pending, missing)
self.assertEqual(performed_status, "performed")
self.assertIsNone(env_at_performed)
def test_recovery_is_idempotent_after_success(self):
missing = self._deleted_worktree()
with patch.dict(os.environ, self._recovery_env(missing)):
first = mcp_server._assess_stale_active_binding(auto_recover=True)
second = mcp_server._assess_stale_active_binding(auto_recover=True)
self.assertTrue(first["recovery_performed"])
self.assertEqual(second["classification"], sbr.CLASSIFICATION_UNBOUND)
self.assertFalse(second["recovery_performed"])
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink