62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
"""生命周期隔离：调用方的 AGENT_SYSTEM_* 不得进入 serve，serve.env 为 0600。"""
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import shutil
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
from _port_alloc import free_test_port # noqa: E402
|
||
|
||
SOURCE = Path(__file__).resolve().parents[1]
|
||
_spec = importlib.util.spec_from_file_location("tsr", SOURCE / "tests" / "test_svc_recover.py")
|
||
tsr = importlib.util.module_from_spec(_spec)
|
||
_spec.loader.exec_module(tsr)
|
||
|
||
|
||
class LifecycleIsolationTests(tsr.SvcAutoRecoverTests):
|
||
def setUp(self):
|
||
super().setUp()
|
||
shutil.copy2(SOURCE / "scripts" / "svc_manifest.py",
|
||
self.root / "scripts" / "svc_manifest.py")
|
||
|
||
def test_caller_identity_env_does_not_reach_serve(self):
|
||
port = free_test_port()
|
||
self.seed_agent("iso", port)
|
||
env = dict(self.env)
|
||
env["AGENT_SYSTEM_DSN"] = "postgres://leak-user:leak-pass@127.0.0.1/leak"
|
||
env["AGENT_SYSTEM_DATA"] = "/tmp/leak-data"
|
||
env["AGENT_SYSTEM_INSTANCE_ID"] = "svc-foreign-leak"
|
||
self.svc("iso", "start", env=env)
|
||
pid = self.pid_of("iso")
|
||
self.assertIsNotNone(pid)
|
||
raw = Path("/proc/%d/environ" % pid).read_bytes()
|
||
self.assertNotIn(b"leak-pass", raw)
|
||
self.assertNotIn(b"/tmp/leak-data", raw)
|
||
self.assertNotIn(b"svc-foreign-leak", raw)
|
||
serve_env = self.runtime("iso") / "serve.env"
|
||
self.assertTrue(serve_env.is_file())
|
||
self.assertEqual(serve_env.stat().st_mode & 0o777, 0o600)
|
||
text = serve_env.read_text()
|
||
self.assertNotIn("AGENT_SYSTEM_", text)
|
||
self.assertNotIn("leak-pass", text)
|
||
self.svc("iso", "stop")
|
||
|
||
def test_two_bare_restarts_keep_identity(self):
|
||
port = free_test_port()
|
||
self.seed_agent("iso2", port)
|
||
self.svc("iso2", "start")
|
||
manifest = json.loads((self.runtime("iso2") / "manifest.json").read_text())
|
||
iid = manifest["instance_id"]
|
||
data = manifest["identity"]["data_root"]
|
||
self.svc("iso2", "stop")
|
||
self.svc("iso2", "start")
|
||
self.svc("iso2", "stop")
|
||
self.svc("iso2", "start")
|
||
again = json.loads((self.runtime("iso2") / "manifest.json").read_text())
|
||
self.assertEqual(again["instance_id"], iid)
|
||
self.assertEqual(again["identity"]["data_root"], data)
|
||
self.assertEqual(again["identity"].get("db_marker"), "established")
|
||
self.svc("iso2", "stop")
|