Add staged MCP release imports

dev/test_athena_operator.py

@@ -29,7 +29,8 @@ class OperatorTests(unittest.TestCase):
         root = Path(self.temporary.name)
         self.repo, self.stack = root / "repository", root / "stack"
         self.models, self.state = root / "models", root / "state"
-        for path in (self.repo, self.stack, self.models, self.state):
+        self.staging = root / "staging"
+        for path in (self.repo, self.stack, self.models, self.state, self.staging):
             path.mkdir()
         (self.repo / ".git").mkdir()
         (self.repo / "docs").mkdir()
@@ -40,6 +41,7 @@ class OperatorTests(unittest.TestCase):
         self.module.STACK = self.stack.resolve()
         self.module.MODELS = self.models.resolve()
         self.module.STATE = self.state.resolve()
+        self.module.STAGING_ROOTS = (self.staging.resolve(),)
         self.module.audit = lambda *args, **kwargs: None
         self.module.run = lambda argv, **kwargs: {"argv": argv, "exit_code": 0, "output": "ok"}
 
@@ -124,6 +126,40 @@ class OperatorTests(unittest.TestCase):
         self.assertEqual(payload["paths"], ["docs/test.md"])
         self.assertIn("ONE MCP RELEASE", preview)
 
+    def test_mcp_release_imports_reviewed_staging_file_by_sha(self):
+        source = self.staging / "server.py"
+        source.write_text("print('reviewed')\n", encoding="utf-8")
+        digest = self.module.sha(source.read_bytes())
+        payload, preview = self.module.normalise_operation("mcp_release", {
+            "imports": [{"source": str(source), "path": "server.py", "expected_source_sha256": digest}],
+            "services": ["mcp-example"], "message": "Import reviewed MCP source",
+            "checks": ["operator-tests"], "create_recovery": False, "hermes_sync": True,
+        })
+        self.assertEqual(payload["files"][0]["content"], "print('reviewed')\n")
+        self.assertTrue(payload["hermes_sync"])
+        self.assertIn(f"sha256={digest}", preview)
+
+    def test_staged_import_rejects_wrong_sha_and_unapproved_path(self):
+        source = self.staging / "server.py"
+        source.write_text("safe\n", encoding="utf-8")
+        with self.assertRaises(RuntimeError):
+            self.module.staged_file({"source": str(source), "path": "server.py", "expected_source_sha256": "0" * 64})
+        outside = Path(self.temporary.name) / "outside.py"
+        outside.write_text("unsafe\n", encoding="utf-8")
+        with self.assertRaises(PermissionError):
+            self.module.staged_file({"source": str(outside), "path": "server.py", "expected_source_sha256": self.module.sha(outside.read_bytes())})
+
+    def test_search_source_has_python_fallback_without_rg(self):
+        (self.repo / "docs" / "needle.md").write_text("alpha\nneedle here\nomega\n", encoding="utf-8")
+        original = self.module.shutil.which
+        self.module.shutil.which = lambda name: None
+        try:
+            result = self.module.search_source({"query": "needle"})
+        finally:
+            self.module.shutil.which = original
+        self.assertEqual(result["engine"], "python")
+        self.assertIn("docs/needle.md:2:needle here", result["matches"])
+
     def test_structured_power_operations_do_not_exist(self):
         self.assertNotIn("shell", self.module.ALLOWED_OPERATIONS)
         for operation in ("shutdown", "reboot", "ssh", "network", "command"):