feat(zhulan): add restricted remote development cell

server-tools/zhulan-remote-cell/tests/test_validation_executor.py

@@ -0,0 +1,129 @@
+import json
+import sqlite3
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from importlib.util import module_from_spec, spec_from_file_location
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = spec_from_file_location(
+    "zhulan_validation_executor",
+    ROOT / "runtime" / "zhulan_validation_executor.py",
+)
+MODULE = module_from_spec(SPEC)
+assert SPEC and SPEC.loader
+SPEC.loader.exec_module(MODULE)
+
+
+class ValidationExecutorTest(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        root = Path(self.temp.name)
+        workspace_root = root / "workspaces"
+        self.workspace = workspace_root / "DEV-20260813-099" / "bingshuo__guanghu-ice-heart"
+        (self.workspace / ".git").mkdir(parents=True)
+        self.policy = root / "policy.json"
+        self.policy.write_text(
+            json.dumps(
+                {
+                    "repositories": {
+                        "bingshuo/guanghu-ice-heart": {
+                            "allowed_path_prefixes": ["server-tools/zhulan-remote-cell"],
+                            "validation_commands": [["python3", "-m", "unittest"]],
+                        }
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        self.bwrap = root / "zhulan-bwrap"
+        self.bwrap.write_text("test", encoding="utf-8")
+        self.state_db = root / "state.sqlite3"
+        with sqlite3.connect(self.state_db) as db:
+            db.execute(
+                """CREATE TABLE requests (
+                     id TEXT PRIMARY KEY, state TEXT, picked_up_at INTEGER,
+                     capability_expires_at INTEGER, development_id TEXT,
+                     repository TEXT, paths_json TEXT, actions_json TEXT
+                   )"""
+            )
+            db.execute(
+                "INSERT INTO requests VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    "ZLR-20260813-AAAAAAAAAA",
+                    "APPROVED",
+                    int(time.time()),
+                    int(time.time()) + 3600,
+                    "DEV-20260813-099",
+                    "bingshuo/guanghu-ice-heart",
+                    json.dumps(["server-tools/zhulan-remote-cell"]),
+                    json.dumps(["read", "test"]),
+                ),
+            )
+        with patch.dict(
+            "os.environ",
+            {
+                "ZHULAN_WORKSPACE_ROOT": str(workspace_root),
+                "ZHULAN_POLICY": str(self.policy),
+                "ZHULAN_BWRAP": str(self.bwrap),
+                "ZHULAN_DB": str(self.state_db),
+            },
+        ):
+            self.executor = MODULE.ValidationExecutor()
+
+    def tearDown(self):
+        self.temp.cleanup()
+
+    def valid(self):
+        return {
+            "operation": "run",
+            "request_id": "ZLR-20260813-AAAAAAAAAA",
+            "repository": "bingshuo/guanghu-ice-heart",
+            "development_id": "DEV-20260813-099",
+            "workspace": str(self.workspace),
+            "paths": ["server-tools/zhulan-remote-cell"],
+            "command": ["python3", "-m", "unittest"],
+        }
+
+    def test_accepts_only_exact_registered_request(self):
+        workspace, paths, command = self.executor.validate(self.valid())
+        self.assertEqual(workspace, self.workspace.resolve())
+        self.assertEqual(paths, ["server-tools/zhulan-remote-cell"])
+        self.assertEqual(command, ["python3", "-m", "unittest"])
+
+    def test_rejects_extra_fields_arbitrary_command_and_path(self):
+        for key, value in [
+            ("extra", True),
+            ("command", ["sh", "-c", "id"]),
+            ("paths", ["deployment"]),
+        ]:
+            request = self.valid()
+            request[key] = value
+            with self.assertRaises(PermissionError):
+                self.executor.validate(request)
+
+    def test_rejects_workspace_repository_mismatch(self):
+        request = self.valid()
+        wrong = self.workspace.parent / "bingshuo__other"
+        (wrong / ".git").mkdir(parents=True)
+        request["workspace"] = str(wrong)
+        with self.assertRaises(PermissionError):
+            self.executor.validate(request)
+
+    def test_rejects_inactive_or_mismatched_approval_binding(self):
+        for key, value in [
+            ("request_id", "ZLR-20260813-BBBBBBBBBB"),
+            ("development_id", "DEV-20260813-098"),
+        ]:
+            request = self.valid()
+            request[key] = value
+            with self.assertRaises(PermissionError):
+                self.executor.validate(request)
+
+
+if __name__ == "__main__":
+    unittest.main()