guanghu-ice-heart/server-tools/zhulan-remote-cell/tests/test_validation_executor.py

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()