121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""法规知识库组合门禁：components.pin 只接受固定 SHA；run.sh 缺环境直接失败。"""
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
GATE = ROOT / "tests" / "integration" / "regulations-kb"
|
||
PIN = GATE / "components.pin"
|
||
RUN = GATE / "run.sh"
|
||
HEX40 = re.compile(r"^[0-9a-f]{40}$")
|
||
KEYS = {"rdm_sha", "kb_sha"}
|
||
REQUIRED = ("KBREG_TEST_DATABASE_URL", "SVC_RDM_SRC_DIR", "KB_SRC_DIR")
|
||
|
||
|
||
def load_pin(text):
|
||
out = {}
|
||
for line in text.splitlines():
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
if "=" not in line:
|
||
raise ValueError("bad pin line")
|
||
key, value = line.split("=", 1)
|
||
if key not in KEYS:
|
||
raise ValueError("unknown pin key")
|
||
if not HEX40.match(value):
|
||
raise ValueError("sha must be 40 hex")
|
||
if key in out:
|
||
raise ValueError("duplicate pin key")
|
||
out[key] = value
|
||
if set(out) != KEYS:
|
||
raise ValueError("pin keys incomplete")
|
||
return out
|
||
|
||
|
||
def run_gate(env, timeout=20):
|
||
base = {"PATH": os.environ.get("PATH", "/usr/bin:/bin")}
|
||
base.update(env)
|
||
return subprocess.run(
|
||
["bash", str(RUN)], cwd=ROOT, env=base, capture_output=True, text=True, timeout=timeout
|
||
)
|
||
|
||
|
||
class RegulationsKbPinTests(unittest.TestCase):
|
||
def test_pin_is_two_fixed_shas(self):
|
||
doc = load_pin(PIN.read_text())
|
||
self.assertEqual(set(doc), KEYS)
|
||
for value in doc.values():
|
||
self.assertRegex(value, HEX40)
|
||
|
||
def test_parser_rejects_url_short_sha_and_unknown_key(self):
|
||
kb = "kb_sha=" + "b" * 40 + "\n"
|
||
bad = [
|
||
"rdm_sha=https://example.invalid/repo.git\n" + kb,
|
||
"rdm_sha=abc\n" + kb,
|
||
"rdm_sha=" + "a" * 40 + "\n" + kb + "url=" + "c" * 40 + "\n",
|
||
]
|
||
for text in bad:
|
||
with self.assertRaises(ValueError):
|
||
load_pin(text)
|
||
|
||
def test_run_sh_parser_rejects_bad_pins_before_env_check(self):
|
||
kb = "kb_sha=" + "b" * 40 + "\n"
|
||
cases = {
|
||
"rdm_sha=https://example.invalid/repo.git\n" + kb: "不是 40 位 hex",
|
||
"rdm_sha=abc\n" + kb: "不是 40 位 hex",
|
||
"rdm_sha=" + "a" * 40 + "\n" + kb + "url=" + "c" * 40 + "\n": "未知键",
|
||
"rdm_sha=$(touch /tmp/regkb-pwned)\n" + kb: "不是 40 位 hex",
|
||
kb: "必须同时有",
|
||
}
|
||
for text, expect in cases.items():
|
||
with tempfile.NamedTemporaryFile("w", suffix=".pin", delete=False) as f:
|
||
f.write(text)
|
||
path = f.name
|
||
try:
|
||
proc = run_gate({"REGKB_PIN": path})
|
||
finally:
|
||
os.unlink(path)
|
||
self.assertNotEqual(proc.returncode, 0, text)
|
||
self.assertIn(expect, proc.stderr, text)
|
||
self.assertNotIn("=== stage", proc.stdout)
|
||
|
||
def test_run_sh_names_each_missing_variable_without_running_stages(self):
|
||
proc = run_gate({})
|
||
self.assertNotEqual(proc.returncode, 0)
|
||
for name in REQUIRED:
|
||
self.assertIn(name, proc.stderr)
|
||
self.assertNotIn("=== stage", proc.stdout)
|
||
|
||
for missing in REQUIRED:
|
||
env = {n: "/nonexistent-fake" for n in REQUIRED if n != missing}
|
||
proc = run_gate(env)
|
||
self.assertNotEqual(proc.returncode, 0, missing)
|
||
self.assertIn(missing + " 未设置", proc.stderr)
|
||
self.assertNotIn("=== stage", proc.stdout)
|
||
self.assertNotIn("/nonexistent-fake", proc.stdout + proc.stderr)
|
||
|
||
def test_scan_default_excludes_server_bin(self):
|
||
text = RUN.read_text()
|
||
loops = [ln.strip() for ln in text.splitlines() if ln.strip().startswith('for d in "$rdm_dir/')]
|
||
self.assertEqual(len(loops), 1)
|
||
self.assertIn('"$rdm_dir/.build/product/frontend/dist"', loops[0])
|
||
self.assertNotIn('"$rdm_dir/bin"', text)
|
||
|
||
with tempfile.TemporaryDirectory() as rdm_dir:
|
||
Path(rdm_dir, "bin").mkdir()
|
||
Path(rdm_dir, "bin", "rdm").write_text("KBREG_DATABASE_URL\n")
|
||
script = 'dirs=(); ' + loops[0] + ' [ -d "$d" ] && dirs+=("$d"); done; printf "%s\\n" "${dirs[@]}"'
|
||
proc = subprocess.run(
|
||
["bash", "-c", script], env={"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "rdm_dir": rdm_dir},
|
||
capture_output=True, text=True, timeout=10,
|
||
)
|
||
self.assertEqual(proc.returncode, 0, proc.stderr)
|
||
self.assertEqual(proc.stdout.strip(), "")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|