test: align duplicate-gate and artifact tests with keyed lock store (#443)

Update lock-issue and duplicate-recheck tests to mock branch listing,
use GITEA_ISSUE_LOCK_DIR session binding, and satisfy create_pr
provenance/worktree guards after the keyed persistent lock store landed.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

tests/test_agent_temp_artifacts.py

@@ -87,6 +87,7 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
         "mcp_server.issue_duplicate_context_fetcher",
         return_value=([], [], {"status": "not_claimed"}),
     )
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server._auth", return_value="token x")
     @patch("mcp_server._resolve", return_value=("h", "o", "r"))
     @patch("issue_lock_worktree.read_worktree_git_state")

tests/test_issue_work_duplicate_gate.py

@@ -1,5 +1,4 @@
 """Tests for early duplicate-work detection (#400)."""
-import json
 import os
 import sys
 import tempfile
@@ -9,6 +8,8 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
+import issue_lock_provenance
+import issue_lock_store
 import issue_work_duplicate_gate as dup_gate
 import mcp_server
 from issue_work_duplicate_gate import (
@@ -122,8 +123,9 @@ class TestDuplicateReportOutcome(unittest.TestCase):
 
 
 class TestInjectableDuplicateFetcher(unittest.TestCase):
+    @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value="token x")
-    def test_lock_issue_uses_injected_fetcher(self, _auth):
+    def test_lock_issue_uses_injected_fetcher(self, _auth, _api):
         seen = {}
 
         def fetcher(h, o, r, auth, issue_number):
@@ -140,26 +142,29 @@ class TestInjectableDuplicateFetcher(unittest.TestCase):
                 "porcelain_status": "",
                 "base_equivalent": True,
             },
-        ), patch.dict(os.environ, {
+        ):
-            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
+            with tempfile.TemporaryDirectory() as lock_dir:
-        }, clear=True):
+                with patch.dict(os.environ, {
-            with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
+                    "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
-                mcp_server.gitea_lock_issue(
+                    "GITEA_ISSUE_LOCK_DIR": lock_dir,
-                    issue_number=400,
+                }, clear=True):
-                    branch_name="feat/issue-400-duplicate-work-preflight",
+                    mcp_server.gitea_lock_issue(
-                    remote="prgs",
+                        issue_number=400,
-                )
+                        branch_name="feat/issue-400-duplicate-work-preflight",
+                        remote="prgs",
+                    )
         self.assertEqual(seen["issue_number"], 400)
 
 
 class TestMcpDuplicateRecheck(unittest.TestCase):
     def setUp(self):
         self._dir = tempfile.TemporaryDirectory()
-        self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
+        self._env_patch = patch.dict(
-        self._lock_patch = patch.object(
+            os.environ,
-            mcp_server, "ISSUE_LOCK_FILE", self.lock_path
+            {"GITEA_ISSUE_LOCK_DIR": self._dir.name},
+            clear=False,
         )
-        self._lock_patch.start()
+        self._env_patch.start()
         self._remotes = patch.dict(mcp_server.REMOTES, {
             "prgs": {"host": "gitea.example.com", "org": "Example-Org",
                      "repo": "Example-Repo"},
@@ -172,12 +177,27 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
         self._dir.cleanup()
 
     def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
-        with open(self.lock_path, "w", encoding="utf-8") as fh:
+        worktree_path = os.path.realpath(os.getcwd())
-            json.dump({
+        work_lease = {
-                "issue_number": issue_number,
+            "operation_type": "author_issue_work",
-                "branch_name": branch,
+            "issue_number": issue_number,
-                "remote": "prgs",
+            "branch": branch,
-            }, fh)
+            "claimant": {"username": "test-user", "profile": "test-author"},
+            "expires_at": "2999-01-01T00:00:00Z",
+        }
+        issue_lock_store.bind_session_lock({
+            "issue_number": issue_number,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": "Example-Org",
+            "repo": "Example-Repo",
+            "worktree_path": worktree_path,
+            "work_lease": work_lease,
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=work_lease.get("claimant"),
+            ),
+        })
 
     @patch("mcp_server._assess_issue_duplicate_gate")
     @patch("mcp_server.get_profile", return_value={
@@ -241,6 +261,7 @@ class TestMcpDuplicateRecheck(unittest.TestCase):
                 base="master",
                 body="Closes #400",
                 remote="prgs",
+                worktree_path=os.path.realpath(os.getcwd()),
             )
         self.assertFalse(result["success"])
         self.assertIsNone(result.get("number"))