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

129 lines
4.5 KiB
Python

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