Merge remote-tracking branch 'prgs/master' into feat/issue-551-codex-gitea-workflow

# Conflicts: # gitea_mcp_server.py
2026-07-09 08:44:51 -04:00
parent 1ed1539acf 0631e2e155
commit 9c599bf054
14 changed files with 1035 additions and 11 deletions
@@ -0,0 +1,142 @@
+"""Tests for live PR head re-pin before conflict-fix classification (#522)."""
+
+from __future__ import annotations
+
+import unittest
+
+from conflict_fix_classification import (
+    CLASSIFICATION_CONFLICT_FIX_NEEDED,
+    CLASSIFICATION_INCOMPLETE,
+    CLASSIFICATION_LIVE_MERGEABLE,
+    CLASSIFICATION_STALE_INVENTORY_SKIP,
+    assess_conflict_fix_classification,
+    assess_conflict_fix_classification_final_report,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+class TestConflictFixClassification(unittest.TestCase):
+    def test_stale_inventory_mergeable_skip(self):
+        """PR #508 style: inventory mergeable:false/stale head, live mergeable:true."""
+        result = assess_conflict_fix_classification(
+            pr_number=508,
+            inventory_head_sha=_sha("dad1dc8"),
+            inventory_mergeable=False,
+            live_head_sha=_sha("3f3d6cb"),
+            live_mergeable=True,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_STALE_INVENTORY_SKIP)
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["inventory_stale_head"])
+        self.assertTrue(result["inventory_stale_mergeable"])
+        self.assertEqual(result["pinned_head_sha"], _sha("3f3d6cb"))
+
+    def test_stale_inventory_head_only_live_mergeable_true(self):
+        """PR #493 style: head moved; live still mergeable."""
+        result = assess_conflict_fix_classification(
+            pr_number=493,
+            inventory_head_sha=_sha("685e627"),
+            inventory_mergeable=True,
+            live_head_sha=_sha("4561d7a"),
+            live_mergeable=True,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_STALE_INVENTORY_SKIP)
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["inventory_stale_head"])
+        self.assertEqual(result["pinned_head_sha"], _sha("4561d7a"))
+
+    def test_live_conflict_allows_worktree_on_pinned_head(self):
+        result = assess_conflict_fix_classification(
+            pr_number=99,
+            inventory_head_sha=_sha("aaaaaaa"),
+            inventory_mergeable=False,
+            live_head_sha=_sha("bbbbbbb"),
+            live_mergeable=False,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_CONFLICT_FIX_NEEDED)
+        self.assertTrue(result["worktree_allowed"])
+        self.assertFalse(result["skip_author_mutation"])
+        self.assertTrue(result["inventory_stale_head"])
+        self.assertEqual(result["pinned_head_sha"], _sha("bbbbbbb"))
+
+    def test_matching_inventory_live_mergeable_skip(self):
+        head = _sha("cccccccc")
+        result = assess_conflict_fix_classification(
+            pr_number=10,
+            inventory_head_sha=head,
+            inventory_mergeable=True,
+            live_head_sha=head,
+            live_mergeable=True,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_LIVE_MERGEABLE)
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertFalse(result["inventory_stale_head"])
+
+    def test_missing_live_head_incomplete(self):
+        result = assess_conflict_fix_classification(
+            pr_number=1,
+            inventory_head_sha=_sha("ddddddd"),
+            inventory_mergeable=False,
+            live_head_sha=None,
+            live_mergeable=False,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_INCOMPLETE)
+        self.assertFalse(result["worktree_allowed"])
+        self.assertTrue(result["skip_author_mutation"])
+        self.assertTrue(any("live PR head" in r for r in result["reasons"]))
+
+    def test_short_sha_rejected(self):
+        result = assess_conflict_fix_classification(
+            pr_number=1,
+            inventory_head_sha="abc1234",
+            inventory_mergeable=False,
+            live_head_sha="def5678",
+            live_mergeable=False,
+        )
+        self.assertEqual(result["classification"], CLASSIFICATION_INCOMPLETE)
+        self.assertFalse(result["worktree_allowed"])
+
+
+class TestConflictFixClassificationFinalReport(unittest.TestCase):
+    def test_non_conflict_report_not_applicable(self):
+        result = assess_conflict_fix_classification_final_report(
+            "Implemented feature without merge issues."
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["applicable"])
+
+    def test_conflict_report_requires_tool_live_head_classification(self):
+        result = assess_conflict_fix_classification_final_report(
+            "Did conflict-fix work on PR #99."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["applicable"])
+        joined = " ".join(result["reasons"])
+        self.assertIn("gitea_assess_conflict_fix_classification", joined)
+        self.assertIn("Live head SHA", joined)
+        self.assertIn("classification", joined.lower())
+
+    def test_good_conflict_report_passes(self):
+        live = _sha("3f3d6cb")
+        inv = _sha("dad1dc8")
+        text = f"""
+Conflict-fix classification: stale_inventory_skip
+Called gitea_assess_conflict_fix_classification before worktree creation.
+Inventory head SHA: {inv}
+Live head SHA: {live}
+Use live head only; skip author mutation.
+"""
+        result = assess_conflict_fix_classification_final_report(text)
+        self.assertTrue(result["proven"], msg=result.get("reasons"))
+        self.assertEqual(result["live_head_sha"], live)
+        self.assertEqual(result["inventory_head_sha"], inv)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,88 @@
+"""Tests for web UI runtime health dashboard (#430)."""
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from starlette.testclient import TestClient
+
+from webui.app import create_app
+from webui.runtime_health import _role_kind, load_runtime_snapshot, snapshot_to_dict
+
+
+def _mock_identity(_host: str):
+    return "jcwalker3", None
+
+
+def _mock_git_sync(_repo: Path, _remote: str, _branch: str):
+    return "localsha123456", "remoteshaabcdef", 2
+
+
+class TestRuntimeHealthLoader(unittest.TestCase):
+    def test_role_kind_author(self):
+        allowed = ["gitea.pr.create", "gitea.branch.push", "gitea.read"]
+        forbidden = ["gitea.pr.merge", "gitea.pr.approve"]
+        self.assertEqual(_role_kind(allowed, forbidden), "author")
+
+    def test_snapshot_with_mocks(self):
+        snapshot = load_runtime_snapshot(
+            resolve_username=_mock_identity,
+            git_sync=_mock_git_sync,
+        )
+        self.assertEqual(snapshot.project_id, "gitea-tools")
+        self.assertEqual(snapshot.authenticated_username, "jcwalker3")
+        self.assertEqual(snapshot.commits_behind_master, 2)
+        self.assertIsNotNone(snapshot.stale_runtime_warning)
+        self.assertGreaterEqual(len(snapshot.workflow_hashes), 3)
+        self.assertGreaterEqual(len(snapshot.schema_hashes), 2)
+        self.assertIn("shell_use_allowed", snapshot.shell_health)
+
+    def test_snapshot_dict_export(self):
+        snapshot = load_runtime_snapshot(
+            resolve_username=_mock_identity,
+            git_sync=_mock_git_sync,
+        )
+        data = snapshot_to_dict(snapshot)
+        self.assertEqual(data["authenticated_username"], "jcwalker3")
+        self.assertEqual(data["commits_behind_master"], 2)
+        self.assertTrue(data["stale_runtime_warning"])
+
+
+class TestRuntimeRoutes(unittest.TestCase):
+    def setUp(self):
+        self.client = TestClient(create_app())
+        self.snapshot = load_runtime_snapshot(
+            resolve_username=_mock_identity,
+            git_sync=_mock_git_sync,
+        )
+        self._patch = mock.patch(
+            "webui.app.load_runtime_snapshot",
+            return_value=self.snapshot,
+        )
+        self._patch.start()
+
+    def tearDown(self):
+        self._patch.stop()
+
+    def test_runtime_page_renders_profile_and_warning(self):
+        response = self.client.get("/runtime")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Runtime health", response.text)
+        self.assertIn("Active profile", response.text)
+        self.assertIn("Stale runtime warning", response.text)
+        self.assertIn("Workflow hashes", response.text)
+        self.assertNotIn("child issue", response.text.lower())
+
+    def test_api_runtime_json(self):
+        response = self.client.get("/api/runtime")
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertEqual(data["project_id"], "gitea-tools")
+        self.assertEqual(data["role_kind"], self.snapshot.role_kind)
+        self.assertEqual(data["commits_behind_master"], 2)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -29,8 +29,13 @@ class TestWebuiSkeleton(unittest.TestCase):
        self.assertIn("Operator console", response.text)
        self.assertIn("Read-only MVP", response.text)

+    def test_runtime_is_implemented(self):
+        response = self.client.get("/runtime")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Runtime health", response.text)
+
    def test_route_stubs_render(self):
-        for path in ("/runtime", "/audit"):
+        for path in ("/audit",):
            with self.subTest(path=path):
                response = self.client.get(path)
                self.assertEqual(response.status_code, 200)