Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

Merge master into feat/issue-204

Browse Source
This commit is contained in:
sysadmin
2026-07-05 16:29:58 -04:00
parent 88deed4e69 c6fd0fd963
commit 4fb08a012e
9 changed files with 931 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_author_proofs.py
+234
View File
@@ -0,0 +1,234 @@
"""Tests for author-side branch-identity proofs (Issue #177).
Issue #177 (author-side counterpart of the #173 reviewer proofs) requires
author workflows to *prove* local git state before staging, committing, or
pushing, instead of discovering drift after the fact:
1. The current branch equals the intended feature branch and is never a
protected branch (master/main/develop/development/dev).
2. Branch or HEAD drift between validation and commit — including external
branch switches in a shared worktree — stops the workflow.
3. A push requires local branch, remote target branch, and intended issue
branch to all match.
4. An accidental commit on a protected branch must not be pushed and its
repair must be reported, never silently continued.
These are the harness assertions from the issue's Required behavior 5.
"""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from author_proofs import ( # noqa: E402
PROTECTED_BRANCHES,
assess_protected_branch_commit,
build_commit_push_report,
detect_branch_drift,
verify_branch_for_commit,
verify_push_target,
)
FEATURE = "feat/issue-177-branch-drift-proofs"
HEAD_1 = "64dc334a92685b7b6a1fdb7ffe363f02a69f5dbd"
HEAD_2 = "ccc5ef79dfe629853e144763238593bd808d57e0"
class TestProtectedBranches(unittest.TestCase):
def test_known_protected_names(self):
for name in ("master", "main", "develop", "development", "dev"):
self.assertIn(name, PROTECTED_BRANCHES)
class TestVerifyBranchForCommit(unittest.TestCase):
"""Required behavior 1: prove the branch before staging/committing."""
def test_on_intended_feature_branch_is_proven(self):
proof = verify_branch_for_commit(FEATURE, FEATURE)
self.assertTrue(proof["proven"])
self.assertFalse(proof["block"])
def test_commit_attempted_while_on_master_is_blocked(self):
# Harness assertion (behavior 5, bullet 1).
proof = verify_branch_for_commit("master", FEATURE)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
self.assertTrue(any("master" in r for r in proof["reasons"]))
def test_every_protected_branch_is_blocked_as_current(self):
for name in PROTECTED_BRANCHES:
proof = verify_branch_for_commit(name, FEATURE)
self.assertTrue(proof["block"], name)
def test_intended_branch_may_not_be_protected(self):
proof = verify_branch_for_commit("master", "master")
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_wrong_feature_branch_is_blocked(self):
proof = verify_branch_for_commit("feat/issue-178-other-work", FEATURE)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_missing_current_branch_fails_closed(self):
proof = verify_branch_for_commit("", FEATURE)
self.assertTrue(proof["block"])
def test_missing_intended_branch_fails_closed(self):
proof = verify_branch_for_commit(FEATURE, None)
self.assertTrue(proof["block"])
class TestBranchDrift(unittest.TestCase):
"""Required behaviors 2 + 3: drift between validation and commit stops
the workflow."""
def test_no_drift_when_branch_and_head_unchanged(self):
drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1)
self.assertFalse(drift["drifted"])
self.assertFalse(drift["block"])
def test_branch_drift_between_validation_and_commit_is_blocked(self):
# Harness assertion (behavior 5, bullet 2).
drift = detect_branch_drift(FEATURE, HEAD_1, "feat/other", HEAD_1)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
def test_shared_worktree_branch_switch_is_detected(self):
# Harness assertion (behavior 5, bullet 4): an external session
# switching the shared checkout to another branch (e.g. master)
# must be detected as drift, not treated as exceptional noise.
drift = detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
self.assertTrue(any("switch" in r.lower() for r in drift["reasons"]))
def test_head_moved_since_validation_is_blocked(self):
drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_2)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
self.assertTrue(any("HEAD" in r for r in drift["reasons"]))
def test_missing_state_fails_closed(self):
drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, None)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
class TestVerifyPushTarget(unittest.TestCase):
"""Required behavior 1 (push leg) + acceptance: push needs proof that
local, remote, and intended branches all match."""
def test_matching_local_remote_and_intended_is_proven(self):
proof = verify_push_target(FEATURE, FEATURE, FEATURE)
self.assertTrue(proof["proven"])
self.assertFalse(proof["block"])
def test_push_target_mismatch_is_blocked(self):
# Harness assertion (behavior 5, bullet 3).
proof = verify_push_target(FEATURE, "feat/issue-178-other-work", FEATURE)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_local_branch_differs_from_intended_is_blocked(self):
proof = verify_push_target("feat/other", FEATURE, FEATURE)
self.assertTrue(proof["block"])
def test_pushing_a_protected_branch_is_blocked(self):
proof = verify_push_target("master", "master", "master")
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_missing_remote_target_fails_closed(self):
proof = verify_push_target(FEATURE, "", FEATURE)
self.assertTrue(proof["block"])
class TestProtectedBranchAccident(unittest.TestCase):
"""Required behavior 4: accidental protected-branch commits must not be
pushed and their repair must be reported."""
def test_feature_branch_commit_is_not_an_accident(self):
result = assess_protected_branch_commit(FEATURE)
self.assertFalse(result["accident"])
self.assertEqual(result["violations"], [])
def test_commit_on_master_is_an_accident_and_must_not_push(self):
result = assess_protected_branch_commit(
"master", pushed=False, repair_reported=True
)
self.assertTrue(result["accident"])
self.assertTrue(result["must_not_push"])
self.assertEqual(result["violations"], [])
self.assertTrue(result["repair_required"])
def test_pushing_the_accident_is_a_violation(self):
result = assess_protected_branch_commit(
"master", pushed=True, repair_reported=True
)
self.assertTrue(any("push" in v.lower() for v in result["violations"]))
def test_silent_repair_is_a_violation(self):
# Harness assertion (behavior 5, bullet 5): the repair path must not
# silently continue without reporting.
result = assess_protected_branch_commit(
"master", pushed=False, repair_reported=False
)
self.assertTrue(any("report" in v.lower() for v in result["violations"]))
class TestCommitPushReport(unittest.TestCase):
"""Acceptance criteria: the final report includes branch proof before
commit and before push, and blocks instead of continuing."""
def _report(self, **overrides):
kwargs = {
"commit_proof": verify_branch_for_commit(FEATURE, FEATURE),
"drift": detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1),
"push_proof": verify_push_target(FEATURE, FEATURE, FEATURE),
"accident": assess_protected_branch_commit(FEATURE),
}
kwargs.update(overrides)
return build_commit_push_report(**kwargs)
def test_fully_proven_report_is_ok(self):
report = self._report()
self.assertEqual(report["status"], "ok")
self.assertTrue(report["branch_proof_before_commit"])
self.assertTrue(report["branch_proof_before_push"])
self.assertFalse(report["drift_detected"])
self.assertEqual(report["violations"], [])
def test_commit_proof_failure_blocks(self):
report = self._report(
commit_proof=verify_branch_for_commit("master", FEATURE)
)
self.assertEqual(report["status"], "blocked")
self.assertFalse(report["branch_proof_before_commit"])
def test_drift_blocks(self):
report = self._report(
drift=detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
)
self.assertEqual(report["status"], "blocked")
self.assertTrue(report["drift_detected"])
def test_push_proof_failure_blocks(self):
report = self._report(
push_proof=verify_push_target(FEATURE, "feat/other", FEATURE)
)
self.assertEqual(report["status"], "blocked")
self.assertFalse(report["branch_proof_before_push"])
def test_accident_violations_block(self):
report = self._report(
accident=assess_protected_branch_commit(
"master", pushed=False, repair_reported=False
)
)
self.assertEqual(report["status"], "blocked")
self.assertTrue(report["violations"])
if __name__ == "__main__":
unittest.main()
tests/test_operator_guide.py
+15 -4
View File
@@ -46,8 +46,8 @@ EXPECTED_SKILLS = [
"gitea-resolve-task-capability",
"profile-switching",
"redaction-security-review",
"jenkins-readonly",
"glitchtip-readonly",
"jenkins-mcp",
"glitchtip-mcp",
"release-operator",
]
@@ -234,8 +234,8 @@ class TestProjectSkills(GuideTestBase):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
r = mcp_list_project_skills()
by_name = {s["name"]: s for s in r["skills"]}
self.assertNotEqual(by_name["jenkins-readonly"]["status"], "available")
self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available")
self.assertNotEqual(by_name["jenkins-mcp"]["status"], "available")
self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
def test_no_urls_in_registry(self):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
@@ -245,6 +245,17 @@ class TestProjectSkills(GuideTestBase):
self.assertNotIn("http://", blob)
self.assertNotIn("keychain:", blob)
def test_enabled_but_no_usable_tools_negative_assertion(self):
"""Negative assertion for 'enabled but no usable tools' (per issue #146)."""
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
r = mcp_list_project_skills()
by_name = {s["name"]: s for s in r["skills"]}
# jenkins-mcp is designed-not-implemented; even if "enabled" in config,
# it should not be usable/available to current profile without tools.
self.assertIn("jenkins-mcp", by_name)
self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented")
self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
# ---------------------------------------------------------------------------
# mcp_get_skill_guide
tests/test_review_proofs.py
+200
View File
@@ -23,9 +23,11 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
from review_proofs import ( # noqa: E402
assess_controller_handoff,
assess_inventory_completeness,
assess_role_boundary,
assess_self_review_contamination,
assess_validation_report,
build_final_report,
pr_inventory_trust_gate,
resolve_repos_from_user_reference,
verify_pinned_head_checkout,
)
@@ -99,6 +101,21 @@ def _good_contamination():
)
def _good_role_boundary():
return assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": False,
"author_mutations": [],
"review_mutations": [],
"operator_authorized_author_work": False,
"scratch_evidence_claimed": False,
}
)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -368,6 +385,74 @@ class TestSelfReviewContamination(unittest.TestCase):
self.assertEqual(result["status"], "unknown")
class TestRoleBoundary(unittest.TestCase):
"""Issue #175: reviewer queue tasks must not pivot into author work."""
def test_reviewer_queue_without_author_mutations_is_clean(self):
result = _good_role_boundary()
self.assertEqual(result["status"], "clean")
self.assertEqual(result["violations"], [])
def test_reviewer_queue_author_mutation_without_authorization_violates(self):
result = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": True,
"author_mutations": ["claim issue #171", "push branch"],
"operator_authorized_author_work": False,
"mixed_namespace_justification": (
"author namespace was used for implementation"
),
}
)
self.assertEqual(result["status"], "violation")
self.assertTrue(any("pivot" in r for r in result["violations"]))
def test_mixed_namespace_use_without_justification_is_warning(self):
result = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": True,
"author_mutations": [],
}
)
self.assertEqual(result["status"], "warning")
self.assertTrue(
any("mixed" in r.lower() for r in result["reasons"])
)
def test_author_task_cannot_perform_review_mutations(self):
result = assess_role_boundary(
{
"task_role": "author",
"reviewer_namespace_used": False,
"author_namespace_used": True,
"review_mutations": ["approve PR"],
}
)
self.assertEqual(result["status"], "violation")
self.assertTrue(
any("reviewer-only" in r for r in result["violations"])
)
def test_scratch_only_notes_are_not_durable_evidence(self):
result = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"scratch_evidence_claimed": True,
"scratch_evidence_durable": False,
}
)
self.assertEqual(result["status"], "warning")
self.assertTrue(any("scratch-only" in r for r in result["reasons"]))
class TestFinalReport(unittest.TestCase):
"""Required behavior 6 + acceptance criteria: the report must
distinguish each proof, and only a fully proven run earns an "A"."""
@@ -381,6 +466,7 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"role_boundary": _good_role_boundary(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -393,6 +479,7 @@ class TestFinalReport(unittest.TestCase):
self.assertTrue(report["identity_eligible"])
self.assertTrue(report["pr_author_distinct_from_reviewer"])
self.assertEqual(report["session_contamination"], "clean")
self.assertEqual(report["role_boundary"], "clean")
self.assertTrue(report["validated_on_pinned_head"])
self.assertFalse(report["merge_performed"])
self.assertTrue(report["issue_status_verified"])
@@ -465,6 +552,37 @@ class TestFinalReport(unittest.TestCase):
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
def test_missing_role_boundary_downgrades_and_blocks_merge(self):
kwargs = {
"checkout_proof": _good_checkout(),
"inventory": _good_inventory(),
"validation": _good_validation(),
"contamination": _good_contamination(),
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
}
report = build_final_report(**kwargs)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
self.assertEqual(report["role_boundary"], "warning")
def test_role_boundary_violation_blocks_report(self):
boundary = assess_role_boundary(
{
"task_role": "reviewer",
"task_kind": "blind_pr_queue_review",
"reviewer_namespace_used": True,
"author_namespace_used": True,
"author_mutations": ["create PR"],
"operator_authorized_author_work": False,
"mixed_namespace_justification": "implementation pivot",
}
)
report = self._report(role_boundary=boundary)
self.assertEqual(report["grade"], "blocked")
self.assertFalse(report["merge_allowed"])
class TestStdoutIsolation(unittest.TestCase):
"""Regression test for #178: tests must not close or corrupt stdout/stderr
@@ -713,5 +831,87 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("issue #182", skill)
class TestPRInventoryTrustGate(unittest.TestCase):
"""Issue #194: unit tests for the PR inventory trust gate."""
def setUp(self):
self.profile = {
"profile_name": "prgs-reviewer",
"allowed_operations": ["read", "gitea.read", "gitea.pr.approve"],
}
self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
def test_trusted_nonempty(self):
res = pr_inventory_trust_gate([{"number": 1}])
self.assertEqual(res["status"], "trusted_nonempty")
self.assertFalse(res["corroborated"])
def test_inventory_error_none_or_not_list(self):
self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error")
self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error")
def test_untrusted_empty_no_pagination_or_corroboration(self):
res = pr_inventory_trust_gate(
[], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
state="open", authenticated_profile=self.profile,
local_remote_url=self.local_url, user_context=None,
corroboration_open_pr_counter=None, has_finality_metadata=False
)
self.assertEqual(res["status"], "untrusted_empty")
self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"])
def test_untrusted_empty_profile_permission_mismatch(self):
bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]}
res = pr_inventory_trust_gate(
[], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
state="open", authenticated_profile=bad_profile,
local_remote_url=self.local_url, user_context=None,
corroboration_open_pr_counter=0, has_finality_metadata=False
)
self.assertEqual(res["status"], "untrusted_empty")
self.assertIn("authenticated profile lacks read permissions", res["reasons"])
def test_untrusted_empty_remote_url_mismatch(self):
bad_url = "https://gitea.prgs.cc/other-org/other-repo.git"
res = pr_inventory_trust_gate(
[], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
state="open", authenticated_profile=self.profile,
local_remote_url=bad_url, user_context=None,
corroboration_open_pr_counter=0, has_finality_metadata=False
)
self.assertEqual(res["status"], "untrusted_empty")
self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"])
def test_untrusted_empty_user_context_indicates_prs(self):
res = pr_inventory_trust_gate(
[], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
state="open", authenticated_profile=self.profile,
local_remote_url=self.local_url, user_context="please check open PR #181",
corroboration_open_pr_counter=0, has_finality_metadata=False
)
self.assertEqual(res["status"], "untrusted_empty")
self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"]))
def test_trusted_empty_with_corroboration(self):
res = pr_inventory_trust_gate(
[], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
state="open", authenticated_profile=self.profile,
local_remote_url=self.local_url, user_context=None,
corroboration_open_pr_counter=0, has_finality_metadata=False
)
self.assertEqual(res["status"], "trusted_empty")
self.assertTrue(res["corroborated"])
def test_trusted_empty_with_finality_metadata(self):
res = pr_inventory_trust_gate(
[], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools",
state="open", authenticated_profile=self.profile,
local_remote_url=self.local_url, user_context=None,
corroboration_open_pr_counter=None, has_finality_metadata=True
)
self.assertEqual(res["status"], "trusted_empty")
self.assertTrue(res["corroborated"])
if __name__ == "__main__":
unittest.main()