diff --git a/docs/development-plan.md b/docs/development-plan.md index 4678056..c97b6c6 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -354,6 +354,9 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立,可并行推进。 - 依赖:无(代码已存在,是补测试和补骨架,不是从零开发) - Done:四模块各自有对应 `test_*.py`(可直接照搬/改造 graphloop 的 `test_change_manifest.py`/`test_selective_apply.py`/`test_workspace_security.py` 等,逻辑未改动,测试也基本不用改);Deploy Supervisor 骨架能独立启停,不受 Live/Candidate 进程崩溃影响 - **待定(2026-07-31 订正)**:此前这里写的是"`supervisor/` 目录该归入 U-002 的目标项目树(`.claude/`、`skills/`、`src/` 那套)还是 U-003 的平台目录"——但 U-002 的范围已经订正为只创建 `.graphlemon/`(见该 Unit 2026-07-31 订正说明),`.claude/`、`skills/` 从来就不是目标项目的东西,是 graphlemon 平台自己仓库的资产。所以这个选择题已经不成立,`supervisor/` 只可能归属"platform 自身工具"(当前的实际存放位置,`supervisor/isolation/`),不存在真正的"归入目标项目"选项;保留这条记录是为了让后来者理解为什么这里曾经有过一次误判 + - **实现补充(2026-08-02)**:四模块的单测直接照搬 graphloop 对应文件(`test_isolated_coding_workspace.py`/`test_workspace_security.py`/`test_change_manifest.py`/`test_selective_apply.py`),逐条核对后确认 API 与行为完全匹配(graphlemon 这四个模块本来就是"adapted verbatim"),只改了 import 路径(`from supervisor.isolation import X`)+ 补了几条 graphlemon 自己模块文档里提到但原测试没覆盖的小 case(`isolated_workspace` 的空/不存在 source_dir 拒绝、`workspace_security` 的换行符注入拒绝);新增 `supervisor/__init__.py`(原本靠 Python 3 namespace package 也能跑,但补一个空的显式 `__init__.py` 和 `src/__init__.py` 保持同一约定,不依赖隐式行为)。52 个测试全绿,放在 `supervisor/isolation/test_*.py`(和被测模块同目录,比照 `src/test_*.py` 的既有约定),运行方式 `python3 -m unittest discover -s supervisor/isolation -p "test_*.py"`(不在 `python3 -m unittest discover -s src` 覆盖范围内,是新增的第二条独立测试入口,README/CLAUDE.md 后续需要记一笔)。 + Deploy Supervisor 骨架落在新文件 `supervisor/deploy_supervisor.py`:PID 文件+状态文件走 `paths.platform_home()/deploy_supervisor/` 下(复用 U-003 已有的运行时目录约定,不新建一套),原子写沿用 `tempfile+os.replace`(参照 graphloop `hermes_watchdog.py` 的 `_atomic_save`,未直接 import,逻辑重写了一遍)。`run_foreground(poll_fn, poll_interval, iterations)` 是骨架的核心循环——`poll_fn` 是 U-037+ 才会真正实现的"检查是否有待部署的 commit 并行动"逻辑,本 Unit 只负责按周期调用它、写 PID/状态文件、处理 `SIGTERM`(收到后跳出循环,`finally` 块无条件写 `status='stopped'` 并删 PID 文件,包括 `poll_fn` 自己抛异常的情况——fail-closed,崩溃了就该看起来像崩溃了,不能让 PID 文件继续指向一个已经不存在的循环)。`start_detached()`/`stop()` 是给上层(未来某个"启动 graphlemon 平台"脚本,或人工)用的真实进程管理入口,`start_detached` 用 `subprocess.Popen` 拉起真实独立进程(不是线程,故意如此——两个进程之间不共享失败域是这整个设计的前提),`stop()` 发 `SIGTERM` 并轮询等它自己清理干净,超时不 SIGKILL(留给调用方判断,不擅自做主)。**验证方式**:`RunForegroundTest` 系列在同一个测试进程内直接验证循环/状态写入/`SIGTERM`处理(`SIGTERM` 测试用 `poll_fn` 内部对自己进程 `os.kill(os.getpid(), SIGTERM)` 模拟外部信号,不需要真的起第二个进程);`RealSubprocessLifecycleTest` 是真实的(不 mock)`python3 -m supervisor.deploy_supervisor` 子进程拉起+`stop()`往返,确认"能独立启停"这条 Done 标准不是靠单元测试逻辑自证的,是真的对着一个真实操作系统进程验证过的。全程用 `GRAPHLEMON_HOME` 环境变量覆盖到临时目录,不碰真实用户的 `~/.graphlemon`。73 个测试(含四个 isolation 模块)全绿。 + - 覆盖:`supervisor/isolation/` 四模块(真实单测覆盖,不再只是"一次性手工冒烟");Deploy Supervisor 进程骨架(`supervisor/deploy_supervisor.py`,尚不含任何真实部署决策逻辑) - **U-037 [Risk Gate]** `c_deploy_supervisor_stage`:候选实例拉起(全新目录 checkout,不 `git pull` 原地改)+ 一次性 DB 副本机制(候选实例验证期间绝不接触生产库) - 依赖:U-034, U-036 diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 14ec0f4..b543610 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -53,7 +53,7 @@ {"id": "U-034", "phase": "PH5", "sub_phase": "PH5-S1", "name": "01-M 扩展:deployment/deploy_health_signal + core_engine 枚举(见 development-plan.md 2026-08-02 实现补充)", "classification": "Enabler", "depends_on": ["U-001"], "coverage": {"model": ["deployment", "deploy_health_signal", "bug_record"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "两张新表建表成功(共 17 张);bug_record 新枚举值向后兼容 U-015/U-018 已有测试;deployment 状态机用 01-M 1.15 完整 15 值枚举;ux_deployment_active 表达式索引强制全局单例非终态约束;bug_record.py 应用层白名单刻意未跟着放开 core_engine,classification+路由逻辑留给 U-035"}, {"id": "U-035", "phase": "PH5", "sub_phase": "PH5-S1", "name": "self-target worktree 无条件强制 + core_engine 分类审计(见 development-plan.md 2026-08-02 实现补充)", "classification": "Risk Gate", "depends_on": ["U-013", "U-016", "U-034"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-03"]}, "done": "self-target 时无条件走 worktree(work_graph_orchestrate.run_graph 内强制);core_engine 分类(src/core_engine_classify.py,路径规则)在 self-target 任务 PASS 收尾时记一条审计用 bug_record;且改动落地后重跑 U-013/REQ-03(worktree 隔离)现有回归用例全部通过。**2026-08-02 用户决策修正**:Deploy Supervisor 触发条件按 supervisor-auto-rollback-design.md §1 字面设计(self-target+PASS 即触发,bug_record 是旁路产物不是前提),而非 03-C/04-P 字面文本要求的\"先有 core_engine bug_record\"——后者会让从不产生 bug_record 的 goal 型自我改进任务永远触发不了 Deploy Supervisor,已用 AskUserQuestion 与用户确认此订正。c_org_loop_route 未改动(fast_lane/full_replan 是 FAIL 路径重试路由,与此处 PASS 路径的蓝绿触发正交),deployment 行创建仍是 U-037 的范围,不在本 Unit 提前做"}, - {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py;Deploy Supervisor 骨架独立进程可启停,不受 Live/Candidate 崩溃影响"}, + {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架(见 development-plan.md 2026-08-02 实现补充)", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py(供 python3 -m unittest discover -s supervisor/isolation 运行,52 测试全绿);Deploy Supervisor 骨架(supervisor/deploy_supervisor.py)独立进程可启停(真实子进程验证,非 mock),不受 Live/Candidate 崩溃影响;本 Unit 只做进程生命周期骨架,不含任何真实部署决策逻辑(U-037+ 的 poll_fn 挂载点)"}, {"id": "U-037", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_stage", "classification": "Risk Gate", "depends_on": ["U-034", "U-036"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_stage"], "process": [], "requirement": []}, "done": "能生成 deployment 记录并拉起监听临时端口、连一次性 DB 副本的候选进程"}, {"id": "U-038", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_health_check", "classification": "Risk Gate", "depends_on": ["U-037"], "coverage": {"model": [], "view": [], "controller": ["c_deploy_supervisor_health_check"], "process": [], "requirement": []}, "done": "process_alive/http_health/canary_task 三项检查全过 -> health_check_passed;任一失败 -> health_check_failed 且 Live 未受影响"}, {"id": "U-039", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_cutover", "classification": "Risk Gate", "depends_on": ["U-038"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_cutover"], "process": [], "requirement": []}, "done": "健康检查通过后完成真实切换;在途 work_node_execution 重启后可恢复;previous_instance_dir 保留至监控窗口结束"}, diff --git a/supervisor/__init__.py b/supervisor/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/supervisor/__init__.py diff --git a/supervisor/deploy_supervisor.py b/supervisor/deploy_supervisor.py new file mode 100644 index 0000000..a5f984b --- /dev/null +++ b/supervisor/deploy_supervisor.py @@ -0,0 +1,176 @@ +"""Deploy Supervisor process skeleton (U-036). + +Independent OS process (started via `python3 -m supervisor.deploy_supervisor`), +pure stdlib, no LLM CLI dependency -- per docs/supervisor-auto-rollback-design.md +§2's "three independent processes, no shared failure domain" principle: +Deploy Supervisor crashing must not affect Live/Candidate, and Live/Candidate +crashing must not affect Deploy Supervisor's own ability to keep observing +and act. + +PID/status file conventions adapted (NOT imported -- see CLAUDE.md's "do not +import graphloop code directly" rule) from graphloop's hermes_watchdog.py: +atomic writes (tempfile-in-same-dir + os.replace), liveness via +`os.kill(pid, 0)`, fail-closed (this skeleton never auto-restarts anything -- +a crashing `poll_fn` stops the loop and marks status='stopped' rather than +silently dying while still holding the PID file, which would make +`is_running()` lie). + +Scope: this Unit owns process lifecycle only (start/stop/status/PID file). +The actual "check for staged deployments and act" logic is U-037+'s job, +injected here as the `poll_fn` callback -- U-036 has no opinion about what +that logic does, only about running it on a schedule and surviving/reporting +its own crashes correctly. +""" + +import argparse +import json +import os +import signal +import subprocess +import sys +import time + +from src import paths + + +def runtime_dir(): + d = os.path.join(paths.platform_home(), "deploy_supervisor") + os.makedirs(d, exist_ok=True) + return d + + +def pid_file_path(): + return os.path.join(runtime_dir(), "supervisor.pid") + + +def status_file_path(): + return os.path.join(runtime_dir(), "supervisor.status.json") + + +def _atomic_write_json(path, data): + tmp = path + ".{}.tmp".format(os.getpid()) + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(data, handle) + os.replace(tmp, path) + + +def is_alive(pid): + if not isinstance(pid, int) or pid <= 0: + return False + try: + os.kill(pid, 0) + except OSError: + return False + return True + + +def write_status(status, detail=None): + _atomic_write_json(status_file_path(), { + "status": status, "pid": os.getpid(), "updated_at": time.time(), "detail": detail, + }) + + +def read_status(): + try: + with open(status_file_path(), encoding="utf-8") as handle: + return json.load(handle) + except (OSError, json.JSONDecodeError): + return None + + +def is_running(): + """True if a Deploy Supervisor process is currently alive per the PID + file. Doesn't distinguish "never started" from "PID file stale" -- + callers needing that distinction should also check `read_status()`.""" + try: + with open(pid_file_path(), encoding="utf-8") as handle: + pid = int(handle.read().strip()) + except (OSError, ValueError): + return False + return is_alive(pid) + + +class _StopRequested(Exception): + """Internal: raised by the SIGTERM handler to break run_foreground's loop.""" + + +def run_foreground(poll_fn=None, poll_interval=5.0, iterations=None): + """Blocking main loop -- calls `poll_fn()` (if given) once per + `poll_interval` seconds, up to `iterations` cycles (None = forever, only + exits on SIGTERM). Writes a PID file + 'running' status on entry, and + unconditionally writes a 'stopped' status + removes the PID file on any + exit path (normal completion, SIGTERM, or an unhandled exception from + `poll_fn`) -- an exception from `poll_fn` propagates to the caller after + cleanup runs, it is not swallowed (fail-closed: a broken poll cycle + should surface, not vanish into an infinite silent retry). + """ + def _handle_sigterm(signum, frame): + raise _StopRequested() + + previous_handler = signal.signal(signal.SIGTERM, _handle_sigterm) + with open(pid_file_path(), "w", encoding="utf-8") as handle: + handle.write(str(os.getpid())) + write_status("running") + try: + count = 0 + while iterations is None or count < iterations: + if poll_fn is not None: + poll_fn() + count += 1 + if iterations is not None and count >= iterations: + break + time.sleep(poll_interval) + except _StopRequested: + pass + finally: + signal.signal(signal.SIGTERM, previous_handler) + write_status("stopped") + if os.path.exists(pid_file_path()): + os.remove(pid_file_path()) + + +def start_detached(poll_interval=5.0): + """Launches `python3 -m supervisor.deploy_supervisor` as a real, + independent OS process (Popen, not a thread) -- crashing does not take + down whatever called this. Returns the child's pid. Raises RuntimeError + if a Deploy Supervisor is already running (per its own PID file).""" + if is_running(): + raise RuntimeError("Deploy Supervisor is already running") + process = subprocess.Popen( + [sys.executable, "-m", "supervisor.deploy_supervisor", "--poll-interval", str(poll_interval)], + cwd=paths.repo_root(), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + return process.pid + + +def stop(timeout=10.0): + """Sends SIGTERM to the running Deploy Supervisor (if any) and waits up + to `timeout` seconds for it to exit cleanly (PID file removed). Returns + True if it stopped, False if it was already not running or didn't stop + in time -- fail-closed: this does not SIGKILL on timeout, that's a + judgment call left to whoever's driving it, not silently forced here.""" + try: + with open(pid_file_path(), encoding="utf-8") as handle: + pid = int(handle.read().strip()) + except (OSError, ValueError): + return False + if not is_alive(pid): + return False + os.kill(pid, signal.SIGTERM) + deadline = time.time() + timeout + while time.time() < deadline: + if not is_running(): + return True + time.sleep(0.1) + return False + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--poll-interval", type=float, default=5.0) + args = parser.parse_args() + run_foreground(poll_interval=args.poll_interval) + + +if __name__ == "__main__": + main() diff --git a/supervisor/isolation/test_change_manifest.py b/supervisor/isolation/test_change_manifest.py new file mode 100644 index 0000000..2a0aad1 --- /dev/null +++ b/supervisor/isolation/test_change_manifest.py @@ -0,0 +1,313 @@ +"""U-036: real unittest coverage for change_manifest.py -- adapted from +graphloop's test_change_manifest.py (2026-08-02). Module logic is unchanged +from graphloop ("adapted verbatim", see change_manifest.py's own docstring) +except graphlemon's `unified_diff` collapses "@@ -l,s +l,s @@" hunk headers +to a single leading "@@" marker (see that module's own comment) -- the +"only text gets hunks" assertion below counts "@@" substring occurrences, +which still holds 1-to-1 with graphlemon's collapsed format for these +single-hunk fixtures. + +Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v +""" +import hashlib +import os +import shutil +import tempfile +import unittest + +from supervisor.isolation import change_manifest as cm + +TEXT_A = "hello\n" +TEXT_A_MODIFIED = "hello modified\n" +BINARY_BYTES = b"\x89PNG\r\n\x1a\n\x00\x01\x02\x03BINARYDATA\xff\xfe" + + +def sha256_of(data): + return hashlib.sha256(data).hexdigest() + + +class ChangeManifestTestCase(unittest.TestCase): + def setUp(self): + self.source_dir = tempfile.mkdtemp(prefix="cm-src-") + self.addCleanup(shutil.rmtree, self.source_dir, ignore_errors=True) + self._extra_dirs = [] + self.addCleanup(self._cleanup_extra_dirs) + + def _cleanup_extra_dirs(self): + for path in self._extra_dirs: + shutil.rmtree(path, ignore_errors=True) + + def _new_isolated_dir(self): + path = tempfile.mkdtemp(prefix="cm-iso-") + self._extra_dirs.append(path) + return path + + @staticmethod + def _write_text(root, rel_path, content): + full = os.path.join(root, *rel_path.split("/")) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "w", encoding="utf-8") as handle: + handle.write(content) + return full + + @staticmethod + def _write_bytes(root, rel_path, data): + full = os.path.join(root, *rel_path.split("/")) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "wb") as handle: + handle.write(data) + return full + + def _copy_source_to_isolated(self): + isolated_dir = self._new_isolated_dir() + shutil.rmtree(isolated_dir) + shutil.copytree(self.source_dir, isolated_dir) + return isolated_dir + + +class SnapshotWorkspaceTests(ChangeManifestTestCase): + def test_records_text_file_with_hash_size_and_kind(self): + self._write_text(self.source_dir, "a.txt", TEXT_A) + + snapshot = cm.snapshot_workspace(self.source_dir) + + entry = snapshot["files"]["a.txt"] + self.assertEqual(entry["sha256"], sha256_of(TEXT_A.encode("utf-8"))) + self.assertEqual(entry["size"], len(TEXT_A.encode("utf-8"))) + self.assertEqual(entry["kind"], "text") + + def test_records_binary_file_as_binary_kind(self): + self._write_bytes(self.source_dir, "logo.bin", BINARY_BYTES) + + snapshot = cm.snapshot_workspace(self.source_dir) + + entry = snapshot["files"]["logo.bin"] + self.assertEqual(entry["sha256"], sha256_of(BINARY_BYTES)) + self.assertEqual(entry["size"], len(BINARY_BYTES)) + self.assertEqual(entry["kind"], "binary") + + def test_uses_relative_posix_paths_for_nested_files(self): + self._write_text(self.source_dir, "subdir/nested/file.txt", "x") + + snapshot = cm.snapshot_workspace(self.source_dir) + + self.assertIn("subdir/nested/file.txt", snapshot["files"]) + + def test_ignores_git_pycache_pytest_cache_and_dsstore(self): + self._write_text(self.source_dir, "keep.txt", "keep") + self._write_text(self.source_dir, ".git/config", "git config") + self._write_text(self.source_dir, ".git/objects/ab/cdef", "blob") + self._write_text(self.source_dir, "__pycache__/mod.cpython-39.pyc", "pyc") + self._write_text(self.source_dir, ".pytest_cache/v/cache/lastfailed", "{}") + self._write_text(self.source_dir, ".DS_Store", "junk") + + snapshot = cm.snapshot_workspace(self.source_dir) + + self.assertEqual(set(snapshot["files"]), {"keep.txt"}) + + def test_skips_symlinks_but_keeps_their_real_target(self): + target = self._write_text(self.source_dir, "real.txt", "real content") + link_path = os.path.join(self.source_dir, "link.txt") + os.symlink(target, link_path) + + snapshot = cm.snapshot_workspace(self.source_dir) + + self.assertIn("real.txt", snapshot["files"]) + self.assertNotIn("link.txt", snapshot["files"]) + + +class BuildChangeManifestTests(ChangeManifestTestCase): + def _baseline_snapshot(self): + self._write_text(self.source_dir, "a.txt", TEXT_A) + self._write_text(self.source_dir, "b.txt", "world\n") + self._write_text(self.source_dir, "c.txt", "keep me\n") + return cm.snapshot_workspace(self.source_dir) + + def test_added_modified_deleted_and_unchanged_classification(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + + self._write_text(isolated_dir, "a.txt", TEXT_A_MODIFIED) + os.remove(os.path.join(isolated_dir, "b.txt")) + self._write_text(isolated_dir, "d.txt", "brand new\n") + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + by_path = {entry["path"]: entry for entry in manifest["changed_files"]} + + self.assertEqual(set(by_path), {"a.txt", "b.txt", "d.txt"}) + self.assertNotIn("c.txt", by_path) + + modified = by_path["a.txt"] + self.assertEqual(modified["status"], "modified") + self.assertEqual(modified["old_hash"], sha256_of(TEXT_A.encode("utf-8"))) + self.assertEqual(modified["new_hash"], sha256_of(TEXT_A_MODIFIED.encode("utf-8"))) + + deleted = by_path["b.txt"] + self.assertEqual(deleted["status"], "deleted") + self.assertIsNone(deleted["new_hash"]) + self.assertEqual(deleted["old_hash"], sha256_of(b"world\n")) + + added = by_path["d.txt"] + self.assertEqual(added["status"], "added") + self.assertIsNone(added["old_hash"]) + self.assertEqual(added["new_hash"], sha256_of(b"brand new\n")) + + def test_unchanged_files_never_appear_in_changed_files(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + + self.assertEqual(manifest["changed_files"], []) + self.assertEqual(cm.unified_diff(manifest), "") + + def test_ignored_files_excluded_even_when_added_in_isolated(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + + self._write_text(isolated_dir, ".git/config", "new git config") + self._write_text(isolated_dir, "__pycache__/mod.cpython-39.pyc", "pyc") + self._write_text(isolated_dir, ".DS_Store", "junk") + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + + self.assertEqual(manifest["changed_files"], []) + + def test_binary_file_added_is_classified_binary(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + self._write_bytes(isolated_dir, "logo.bin", BINARY_BYTES) + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + + self.assertEqual(len(manifest["changed_files"]), 1) + entry = manifest["changed_files"][0] + self.assertEqual(entry["path"], "logo.bin") + self.assertEqual(entry["status"], "added") + self.assertEqual(entry["kind"], "binary") + self.assertEqual(entry["new_hash"], sha256_of(BINARY_BYTES)) + + def test_changed_files_are_sorted_deterministically(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + + self._write_text(isolated_dir, "z_file.txt", "z\n") + self._write_text(isolated_dir, "0_file.txt", "0\n") + self._write_text(isolated_dir, "m_file.txt", "m\n") + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + paths = [entry["path"] for entry in manifest["changed_files"]] + + self.assertEqual(paths, sorted(paths)) + self.assertEqual(paths, ["0_file.txt", "m_file.txt", "z_file.txt"]) + + def test_rejects_symlink_found_in_isolated_workspace(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + target = os.path.join(isolated_dir, "a.txt") + os.symlink(target, os.path.join(isolated_dir, "sneaky_link.txt")) + + with self.assertRaises(cm.ChangeManifestError): + cm.build_change_manifest(snapshot, isolated_dir) + + def test_rejects_symlink_escaping_isolated_workspace(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + outside_dir = self._new_isolated_dir() + outside_file = self._write_text(outside_dir, "outside.txt", "outside") + os.symlink(outside_file, os.path.join(isolated_dir, "escape_link.txt")) + + with self.assertRaises(cm.ChangeManifestError): + cm.build_change_manifest(snapshot, isolated_dir) + + def test_rejects_tampered_snapshot_with_traversal_path(self): + snapshot = self._baseline_snapshot() + isolated_dir = self._copy_source_to_isolated() + + for bad_path in ("../evil.txt", "/etc/passwd", "sub/../../escape.txt"): + with self.subTest(bad_path=bad_path): + tampered = {"root": snapshot["root"], "files": dict(snapshot["files"])} + tampered["files"][bad_path] = {"sha256": "0" * 64, "size": 0, "kind": "text"} + with self.assertRaises(cm.ChangeManifestError): + cm.build_change_manifest(tampered, isolated_dir) + + +class UnifiedDiffTests(ChangeManifestTestCase): + def test_modified_text_file_produces_readable_unified_diff(self): + self._write_text(self.source_dir, "a.txt", TEXT_A) + snapshot = cm.snapshot_workspace(self.source_dir) + isolated_dir = self._copy_source_to_isolated() + self._write_text(isolated_dir, "a.txt", TEXT_A_MODIFIED) + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + diff_text = cm.unified_diff(manifest) + lines = diff_text.splitlines() + + self.assertTrue(any(l.startswith("---") and "a.txt" in l for l in lines)) + self.assertTrue(any(l.startswith("+++") and "a.txt" in l for l in lines)) + self.assertTrue(any(l.startswith("@@") for l in lines)) + self.assertTrue(any(l.startswith("-") and "hello" in l and not l.startswith("---") for l in lines)) + self.assertTrue(any(l.startswith("+") and "hello modified" in l and not l.startswith("+++") for l in lines)) + + def test_added_text_file_diffs_against_dev_null(self): + snapshot = cm.snapshot_workspace(self.source_dir) + isolated_dir = self._copy_source_to_isolated() + self._write_text(isolated_dir, "new.txt", "brand new content\n") + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + diff_text = cm.unified_diff(manifest) + lines = diff_text.splitlines() + + self.assertTrue(any("/dev/null" in l for l in lines)) + self.assertTrue(any(l.startswith("+") and "brand new content" in l for l in lines)) + removed_content_lines = [l for l in lines if l.startswith("-") and not l.startswith("---")] + self.assertEqual(removed_content_lines, []) + + def test_deleted_text_file_diffs_against_dev_null(self): + self._write_text(self.source_dir, "gone.txt", "will be deleted\n") + snapshot = cm.snapshot_workspace(self.source_dir) + isolated_dir = self._copy_source_to_isolated() + os.remove(os.path.join(isolated_dir, "gone.txt")) + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + diff_text = cm.unified_diff(manifest) + lines = diff_text.splitlines() + + self.assertTrue(any("/dev/null" in l for l in lines)) + self.assertTrue(any(l.startswith("-") and "will be deleted" in l for l in lines)) + added_content_lines = [l for l in lines if l.startswith("+") and not l.startswith("+++")] + self.assertEqual(added_content_lines, []) + + def test_binary_file_diff_is_marker_not_content_diff(self): + snapshot = cm.snapshot_workspace(self.source_dir) + isolated_dir = self._copy_source_to_isolated() + self._write_bytes(isolated_dir, "logo.bin", BINARY_BYTES) + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + diff_text = cm.unified_diff(manifest) + + self.assertIn("Binary files", diff_text) + self.assertIn("logo.bin", diff_text) + self.assertNotIn("@@", diff_text) + + def test_text_and_binary_changes_together_only_text_gets_hunks(self): + self._write_text(self.source_dir, "a.txt", TEXT_A) + self._write_text(self.source_dir, "gone.txt", "will be deleted\n") + snapshot = cm.snapshot_workspace(self.source_dir) + isolated_dir = self._copy_source_to_isolated() + + self._write_text(isolated_dir, "a.txt", TEXT_A_MODIFIED) + os.remove(os.path.join(isolated_dir, "gone.txt")) + self._write_text(isolated_dir, "new.txt", "added text\n") + self._write_bytes(isolated_dir, "logo.bin", BINARY_BYTES) + + manifest = cm.build_change_manifest(snapshot, isolated_dir) + diff_text = cm.unified_diff(manifest) + + text_change_count = sum(1 for entry in manifest["changed_files"] if entry["kind"] == "text") + self.assertEqual(diff_text.count("@@"), text_change_count) + self.assertIn("Binary files", diff_text) + + +if __name__ == "__main__": + unittest.main() diff --git a/supervisor/isolation/test_isolated_workspace.py b/supervisor/isolation/test_isolated_workspace.py new file mode 100644 index 0000000..ce5fc93 --- /dev/null +++ b/supervisor/isolation/test_isolated_workspace.py @@ -0,0 +1,77 @@ +"""U-036: real unittest coverage for isolated_workspace.py -- adapted from +graphloop's test_isolated_coding_workspace.py (2026-08-02). Module logic is +unchanged from graphloop (see isolated_workspace.py's own docstring); only +the import path changed to match this package's location. + +Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v +""" +import os +import shutil +import tempfile +import unittest + +from supervisor.isolation import isolated_workspace as iso + + +class IsolateWorkspaceTests(unittest.TestCase): + def setUp(self): + self.source_dir = tempfile.mkdtemp(prefix="iso-src-") + self.addCleanup(shutil.rmtree, self.source_dir, ignore_errors=True) + + self.source_file = os.path.join(self.source_dir, "hello.txt") + with open(self.source_file, "w", encoding="utf-8") as handle: + handle.write("hello world") + + self._copies_to_clean = [] + self.addCleanup(self._cleanup_copies) + + def _cleanup_copies(self): + for path in self._copies_to_clean: + shutil.rmtree(path, ignore_errors=True) + + def test_returns_separate_path_leaves_source_unchanged_and_copies_file(self): + copied_dir = iso.isolate_workspace(self.source_dir) + self._copies_to_clean.append(copied_dir) + + self.assertNotEqual(os.path.abspath(copied_dir), os.path.abspath(self.source_dir)) + + self.assertTrue(os.path.isdir(self.source_dir)) + with open(self.source_file, "r", encoding="utf-8") as handle: + self.assertEqual(handle.read(), "hello world") + + copied_file = os.path.join(copied_dir, "hello.txt") + self.assertTrue(os.path.isfile(copied_file)) + with open(copied_file, "r", encoding="utf-8") as handle: + self.assertEqual(handle.read(), "hello world") + + def test_parent_dir_overrides_os_default_temp_location(self): + custom_parent = tempfile.mkdtemp(prefix="iso-parent-") + self.addCleanup(shutil.rmtree, custom_parent, ignore_errors=True) + + copied_dir = iso.isolate_workspace(self.source_dir, parent_dir=custom_parent) + self._copies_to_clean.append(copied_dir) + + self.assertEqual(os.path.dirname(copied_dir), custom_parent) + + def test_parent_dir_created_if_missing(self): + base = tempfile.mkdtemp(prefix="iso-parent-") + self.addCleanup(shutil.rmtree, base, ignore_errors=True) + custom_parent = os.path.join(base, "nested", "dir") + + copied_dir = iso.isolate_workspace(self.source_dir, parent_dir=custom_parent) + self._copies_to_clean.append(copied_dir) + + self.assertTrue(os.path.isdir(custom_parent)) + self.assertEqual(os.path.dirname(copied_dir), custom_parent) + + def test_rejects_nonexistent_source_dir(self): + with self.assertRaises(iso.IsolatedWorkspaceError): + iso.isolate_workspace(os.path.join(self.source_dir, "does-not-exist")) + + def test_rejects_empty_source_dir_argument(self): + with self.assertRaises(iso.IsolatedWorkspaceError): + iso.isolate_workspace("") + + +if __name__ == "__main__": + unittest.main() diff --git a/supervisor/isolation/test_selective_apply.py b/supervisor/isolation/test_selective_apply.py new file mode 100644 index 0000000..25c739d --- /dev/null +++ b/supervisor/isolation/test_selective_apply.py @@ -0,0 +1,176 @@ +"""U-036: real unittest coverage for selective_apply.py -- adapted from +graphloop's test_selective_apply.py (2026-08-02). Module logic is unchanged +from graphloop except the internal `import change_manifest as cm` -> +relative package import (see selective_apply.py's own docstring); the +result key is `manifest_id` (not `digest`) in this port. + +Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v +""" +import json +import os +import shutil +import tempfile +import unittest + +from supervisor.isolation import change_manifest as cm +from supervisor.isolation import selective_apply as sa + + +class SelectiveApplyTestCase(unittest.TestCase): + @staticmethod + def _write(root, rel_path, data): + full = os.path.join(root, *rel_path.split("/")) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "wb") as handle: + handle.write(data) + return full + + def _build_scenario(self): + source_dir = tempfile.mkdtemp(prefix="sa-src-") + isolated_dir = tempfile.mkdtemp(prefix="sa-iso-") + self.addCleanup(shutil.rmtree, source_dir, ignore_errors=True) + self.addCleanup(shutil.rmtree, isolated_dir, ignore_errors=True) + + self._write(source_dir, "keep.txt", b"keep\n") + self._write(isolated_dir, "keep.txt", b"keep\n") + + self._write(source_dir, "mod.txt", b"original\n") + self._write(isolated_dir, "mod.txt", b"changed\n") + + self._write(source_dir, "old.txt", b"old content\n") + + self._write(isolated_dir, "new.txt", b"new content\n") + + source_snapshot = cm.snapshot_workspace(source_dir) + manifest = cm.build_change_manifest(source_snapshot, isolated_dir) + return source_dir, isolated_dir, manifest, source_snapshot + + def test_approved_paths_none_raises(self): + source_dir, isolated_dir, manifest, before = self._build_scenario() + with self.assertRaises(sa.SelectiveApplyError): + sa.apply_manifest(manifest, source_dir, isolated_dir, None) + after = cm.snapshot_workspace(source_dir) + self.assertEqual(before, after) + + def test_rejects_path_not_in_manifest_and_leaves_source_untouched(self): + source_dir, isolated_dir, manifest, before = self._build_scenario() + with self.assertRaises(sa.SelectiveApplyError): + sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt", "not_in_manifest.txt"]) + after = cm.snapshot_workspace(source_dir) + self.assertEqual(before, after) + + def test_rejects_absolute_and_traversal_paths(self): + source_dir, isolated_dir, manifest, before = self._build_scenario() + for bad_path in ["/etc/passwd", "../escape.txt", "sub/../../escape.txt"]: + with self.assertRaises(sa.SelectiveApplyError): + sa.apply_manifest(manifest, source_dir, isolated_dir, [bad_path]) + after = cm.snapshot_workspace(source_dir) + self.assertEqual(before, after) + + def test_applies_only_approved_subset_and_preserves_the_rest(self): + source_dir, isolated_dir, manifest, _ = self._build_scenario() + result = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt"]) + + self.assertEqual(result["applied"], ["mod.txt"]) + self.assertEqual(sorted(result["skipped"]), ["new.txt", "old.txt"]) + self.assertEqual(result["already_applied"], []) + + with open(os.path.join(source_dir, "mod.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"changed\n") + self.assertFalse(os.path.exists(os.path.join(source_dir, "new.txt"))) + with open(os.path.join(source_dir, "old.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"old content\n") + with open(os.path.join(source_dir, "keep.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"keep\n") + + def test_apply_add_modify_delete_full_manifest(self): + source_dir, isolated_dir, manifest, _ = self._build_scenario() + result = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt", "new.txt", "old.txt"]) + + self.assertEqual(sorted(result["applied"]), ["mod.txt", "new.txt", "old.txt"]) + self.assertEqual(result["skipped"], []) + self.assertEqual(result["already_applied"], []) + + with open(os.path.join(source_dir, "mod.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"changed\n") + with open(os.path.join(source_dir, "new.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"new content\n") + self.assertFalse(os.path.exists(os.path.join(source_dir, "old.txt"))) + with open(os.path.join(source_dir, "keep.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"keep\n") + + def test_source_drift_rejected_before_any_write(self): + source_dir, isolated_dir, manifest, expected_snapshot = self._build_scenario() + + self._write(source_dir, "keep.txt", b"drifted!\n") + + with self.assertRaises(sa.SelectiveApplyError): + sa.apply_manifest( + manifest, source_dir, isolated_dir, ["mod.txt", "new.txt", "old.txt"], + expected_source_snapshot=expected_snapshot, + ) + + with open(os.path.join(source_dir, "mod.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"original\n") + self.assertFalse(os.path.exists(os.path.join(source_dir, "new.txt"))) + with open(os.path.join(source_dir, "old.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"old content\n") + with open(os.path.join(source_dir, "keep.txt"), "rb") as handle: + self.assertEqual(handle.read(), b"drifted!\n") + + def test_idempotent_second_apply_reports_already_applied(self): + source_dir, isolated_dir, manifest, _ = self._build_scenario() + approved = ["mod.txt", "new.txt", "old.txt"] + + first = sa.apply_manifest(manifest, source_dir, isolated_dir, approved) + self.assertEqual(sorted(first["applied"]), sorted(approved)) + + snapshot_after_first = cm.snapshot_workspace(source_dir) + + second = sa.apply_manifest(manifest, source_dir, isolated_dir, approved) + self.assertEqual(second["applied"], []) + self.assertEqual(sorted(second["already_applied"]), sorted(approved)) + + snapshot_after_second = cm.snapshot_workspace(source_dir) + self.assertEqual(snapshot_after_first, snapshot_after_second) + + def test_symlink_destination_rejected(self): + source_dir = tempfile.mkdtemp(prefix="sa-src-") + isolated_dir = tempfile.mkdtemp(prefix="sa-iso-") + self.addCleanup(shutil.rmtree, source_dir, ignore_errors=True) + self.addCleanup(shutil.rmtree, isolated_dir, ignore_errors=True) + + real_target = self._write(source_dir, "real.txt", b"real\n") + symlink_path = os.path.join(source_dir, "linked.txt") + os.symlink(real_target, symlink_path) + + self._write(isolated_dir, "linked.txt", b"attacker controlled\n") + + source_snapshot = cm.snapshot_workspace(source_dir) + manifest = cm.build_change_manifest(source_snapshot, isolated_dir) + self.assertIn("linked.txt", [entry["path"] for entry in manifest["changed_files"]]) + + with self.assertRaises(sa.SelectiveApplyError): + sa.apply_manifest(manifest, source_dir, isolated_dir, ["linked.txt"]) + + self.assertTrue(os.path.islink(symlink_path)) + self.assertEqual(os.readlink(symlink_path), real_target) + with open(real_target, "rb") as handle: + self.assertEqual(handle.read(), b"real\n") + + def test_result_is_json_serializable_and_carries_stable_identifier(self): + source_dir, isolated_dir, manifest, _ = self._build_scenario() + result = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt"]) + + json.dumps(result) + + self.assertIn("manifest_id", result) + self.assertIsInstance(result["manifest_id"], str) + self.assertTrue(result["manifest_id"]) + + second = sa.apply_manifest(manifest, source_dir, isolated_dir, ["mod.txt"]) + self.assertEqual(result["manifest_id"], second["manifest_id"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/supervisor/isolation/test_workspace_security.py b/supervisor/isolation/test_workspace_security.py new file mode 100644 index 0000000..20a364b --- /dev/null +++ b/supervisor/isolation/test_workspace_security.py @@ -0,0 +1,172 @@ +"""U-036: real unittest coverage for workspace_security.py -- adapted from +graphloop's test_workspace_security.py (2026-08-02). Module logic is +unchanged from graphloop ("adapted verbatim", see workspace_security.py's +own docstring); only the import path changed. + +Run: python3 -m unittest discover -s supervisor/isolation -p "test_*.py" -v +""" + +import os +import shutil +import sys +import tempfile +import unittest + +from supervisor.isolation import workspace_security as ws + + +class ValidateArtifactsTests(unittest.TestCase): + def setUp(self): + self.workspace = tempfile.mkdtemp(prefix="wssec-workspace-") + self.addCleanup(shutil.rmtree, self.workspace, ignore_errors=True) + + self.outside_dir = tempfile.mkdtemp(prefix="wssec-outside-") + self.addCleanup(shutil.rmtree, self.outside_dir, ignore_errors=True) + + def _write(self, relative_path, content="content"): + full_path = os.path.join(self.workspace, relative_path) + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, "w", encoding="utf-8") as handle: + handle.write(content) + return full_path + + def test_accepts_and_normalizes_plain_relative_artifacts(self): + self._write("output.txt") + self._write(os.path.join("nested", "dir", "file.txt")) + + result = ws.validate_artifacts(["output.txt", "nested/dir/file.txt"], self.workspace) + + self.assertEqual(result, ["output.txt", "nested/dir/file.txt"]) + + def test_normalizes_redundant_leading_dot_slash(self): + self._write("output.txt") + + result = ws.validate_artifacts(["./output.txt"], self.workspace) + + self.assertEqual(result, ["output.txt"]) + + def test_rejects_non_string_item(self): + self._write("output.txt") + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts(["output.txt", 42], self.workspace) + + def test_rejects_non_list_artifacts_argument(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts("output.txt", self.workspace) + + def test_rejects_absolute_path_artifact(self): + absolute = self._write("output.txt") + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts([absolute], self.workspace) + + def test_rejects_dot_dot_traversal_even_when_target_exists(self): + self._write("output.txt") + os.makedirs(os.path.join(self.workspace, "sibling"), exist_ok=True) + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts(["sibling/../output.txt"], self.workspace) + + def test_rejects_dot_dot_traversal_escaping_workspace(self): + secret = os.path.join(self.outside_dir, "secret.txt") + with open(secret, "w", encoding="utf-8") as handle: + handle.write("secret") + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts( + [os.path.join("..", os.path.basename(self.outside_dir), "secret.txt")], + self.workspace, + ) + + def test_rejects_missing_artifact(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts(["does_not_exist.txt"], self.workspace) + + def test_rejects_symlink_artifact(self): + real_file = self._write("real.txt") + symlink_path = os.path.join(self.workspace, "link.txt") + os.symlink(real_file, symlink_path) + self.addCleanup(lambda: os.path.exists(symlink_path) and os.remove(symlink_path)) + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts(["link.txt"], self.workspace) + + def test_rejects_artifact_resolving_outside_workspace_via_symlinked_directory(self): + secret = os.path.join(self.outside_dir, "secret.txt") + with open(secret, "w", encoding="utf-8") as handle: + handle.write("secret") + + symlinked_dir = os.path.join(self.workspace, "escape") + os.symlink(self.outside_dir, symlinked_dir, target_is_directory=True) + self.addCleanup(lambda: os.path.exists(symlinked_dir) and os.remove(symlinked_dir)) + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts(["escape/secret.txt"], self.workspace) + + def test_rejects_empty_string_artifact(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_artifacts([""], self.workspace) + + +class ValidateSubprocessCommandTests(unittest.TestCase): + def setUp(self): + self.workspace = tempfile.mkdtemp(prefix="wssec-cmd-workspace-") + self.addCleanup(shutil.rmtree, self.workspace, ignore_errors=True) + + self.adapter_script = os.path.join(self.workspace, "existing_adapter.py") + with open(self.adapter_script, "w", encoding="utf-8") as handle: + handle.write("print('ok')\n") + + def test_accepts_python_executable_with_existing_script(self): + result = ws.validate_subprocess_command([sys.executable, self.adapter_script]) + + self.assertEqual(result, [sys.executable, self.adapter_script]) + + def test_rejects_empty_command(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command([]) + + def test_rejects_non_list_command(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command("{} {}".format(sys.executable, self.adapter_script)) + + def test_rejects_non_string_argument(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command([sys.executable, self.adapter_script, 7]) + + def test_rejects_shell_metacharacters_in_arguments(self): + dangerous_args = [ + "foo; rm -rf /", "foo && rm -rf /", "foo | rm -rf /", + "foo > /etc/passwd", "foo `rm -rf /`", "foo $(rm -rf /)", + ] + for dangerous_arg in dangerous_args: + with self.subTest(dangerous_arg=dangerous_arg): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command([sys.executable, self.adapter_script, dangerous_arg]) + + def test_rejects_bare_shell_invocation(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command(["/bin/sh", "-c", "echo hi"]) + + def test_rejects_script_argument_with_disallowed_extension(self): + non_script = os.path.join(self.workspace, "not_a_script.txt") + with open(non_script, "w", encoding="utf-8") as handle: + handle.write("not python") + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command([sys.executable, non_script]) + + def test_rejects_script_argument_that_does_not_exist(self): + missing_script = os.path.join(self.workspace, "missing_adapter.py") + + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command([sys.executable, missing_script]) + + def test_rejects_newline_in_argument(self): + with self.assertRaises(ws.WorkspaceSecurityError): + ws.validate_subprocess_command([sys.executable, self.adapter_script, "foo\nbar"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/supervisor/test_deploy_supervisor.py b/supervisor/test_deploy_supervisor.py new file mode 100644 index 0000000..def1bf8 --- /dev/null +++ b/supervisor/test_deploy_supervisor.py @@ -0,0 +1,193 @@ +"""U-036 tests: Deploy Supervisor process skeleton -- lifecycle only (PID +file, status file, start/stop), no actual deploy logic (that's U-037+). +GRAPHLEMON_HOME is always overridden to a throwaway tempdir so these tests +never touch the real user's ~/.graphlemon. + +Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v +""" + +import os +import shutil +import signal +import subprocess +import sys +import tempfile +import time +import unittest + +from supervisor import deploy_supervisor as ds +from src import paths + + +class DeploySupervisorTestBase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="deploy-supervisor-test-") + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.old_home = os.environ.get("GRAPHLEMON_HOME") + os.environ["GRAPHLEMON_HOME"] = self.tmp + + def tearDown(self): + if self.old_home is None: + os.environ.pop("GRAPHLEMON_HOME", None) + else: + os.environ["GRAPHLEMON_HOME"] = self.old_home + + +class RuntimePathsTest(DeploySupervisorTestBase): + def test_runtime_dir_created_under_platform_home(self): + d = ds.runtime_dir() + self.assertTrue(os.path.isdir(d)) + self.assertTrue(d.startswith(paths.platform_home())) + + def test_pid_and_status_file_paths_live_under_runtime_dir(self): + self.assertEqual(os.path.dirname(ds.pid_file_path()), ds.runtime_dir()) + self.assertEqual(os.path.dirname(ds.status_file_path()), ds.runtime_dir()) + + +class IsAliveTest(unittest.TestCase): + def test_current_process_is_alive(self): + self.assertTrue(ds.is_alive(os.getpid())) + + def test_zero_or_negative_pid_is_not_alive(self): + self.assertFalse(ds.is_alive(0)) + self.assertFalse(ds.is_alive(-1)) + + def test_non_int_is_not_alive(self): + self.assertFalse(ds.is_alive("not-a-pid")) + + def test_a_pid_that_does_not_exist_is_not_alive(self): + # PID_MAX on Linux/macOS is well below this; astronomically unlikely + # to collide with a real running process. + self.assertFalse(ds.is_alive(2**30)) + + +class StatusFileTest(DeploySupervisorTestBase): + def test_write_then_read_status_round_trips(self): + ds.write_status("running", detail="test detail") + status = ds.read_status() + self.assertEqual(status["status"], "running") + self.assertEqual(status["detail"], "test detail") + self.assertEqual(status["pid"], os.getpid()) + + def test_read_status_returns_none_when_no_file_yet(self): + self.assertIsNone(ds.read_status()) + + def test_write_status_is_atomic_no_tmp_file_left_behind(self): + ds.write_status("running") + entries = os.listdir(ds.runtime_dir()) + self.assertEqual(entries, ["supervisor.status.json"]) + + +class IsRunningTest(DeploySupervisorTestBase): + def test_false_when_no_pid_file(self): + self.assertFalse(ds.is_running()) + + def test_true_when_pid_file_points_to_a_live_process(self): + with open(ds.pid_file_path(), "w", encoding="utf-8") as handle: + handle.write(str(os.getpid())) + self.assertTrue(ds.is_running()) + + def test_false_when_pid_file_points_to_a_dead_process(self): + with open(ds.pid_file_path(), "w", encoding="utf-8") as handle: + handle.write(str(2**30)) + self.assertFalse(ds.is_running()) + + def test_false_when_pid_file_is_corrupt(self): + with open(ds.pid_file_path(), "w", encoding="utf-8") as handle: + handle.write("not-a-number") + self.assertFalse(ds.is_running()) + + +class RunForegroundTest(DeploySupervisorTestBase): + def test_writes_pid_file_and_running_status_during_the_loop(self): + seen_running = [] + + def poll_fn(): + seen_running.append(ds.is_running()) + seen_running.append(ds.read_status()["status"]) + + ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1) + + self.assertEqual(seen_running, [True, "running"]) + + def test_calls_poll_fn_exactly_iterations_times(self): + calls = [] + ds.run_foreground(poll_fn=lambda: calls.append(1), poll_interval=0, iterations=3) + self.assertEqual(len(calls), 3) + + def test_writes_stopped_status_and_removes_pid_file_on_normal_completion(self): + ds.run_foreground(poll_fn=None, poll_interval=0, iterations=1) + self.assertEqual(ds.read_status()["status"], "stopped") + self.assertFalse(os.path.exists(ds.pid_file_path())) + + def test_sigterm_breaks_the_loop_and_still_writes_stopped_status(self): + calls = [] + + def poll_fn(): + calls.append(1) + if len(calls) == 2: + os.kill(os.getpid(), signal.SIGTERM) + + # iterations=None (would run forever) -- SIGTERM is the only exit. + ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=None) + + self.assertEqual(len(calls), 2) + self.assertEqual(ds.read_status()["status"], "stopped") + self.assertFalse(os.path.exists(ds.pid_file_path())) + + def test_poll_fn_exception_propagates_but_cleanup_still_runs(self): + def poll_fn(): + raise RuntimeError("boom") + + with self.assertRaises(RuntimeError): + ds.run_foreground(poll_fn=poll_fn, poll_interval=0, iterations=1) + + self.assertEqual(ds.read_status()["status"], "stopped") + self.assertFalse(os.path.exists(ds.pid_file_path())) + + +class RealSubprocessLifecycleTest(DeploySupervisorTestBase): + """Not mocked -- actually spawns `python3 -m supervisor.deploy_supervisor` + as a real independent OS process and stops it for real, per this Unit's + Done criterion ("骨架能独立启停").""" + + def test_start_detached_then_stop(self): + env = dict(os.environ) + # start_detached itself calls is_running()/Popen using the current + # process's env var (already patched by setUp); the child process + # inherits it via Popen's default env=None -> os.environ passthrough. + pid = ds.start_detached(poll_interval=0.1) + try: + deadline = time.time() + 5 + while time.time() < deadline and not ds.is_running(): + time.sleep(0.05) + self.assertTrue(ds.is_running()) + self.assertTrue(ds.is_alive(pid)) + + stopped = ds.stop(timeout=5) + self.assertTrue(stopped) + self.assertFalse(ds.is_running()) + finally: + if ds.is_alive(pid): + os.kill(pid, signal.SIGKILL) + + def test_start_detached_twice_raises_while_already_running(self): + pid = ds.start_detached(poll_interval=0.1) + try: + deadline = time.time() + 5 + while time.time() < deadline and not ds.is_running(): + time.sleep(0.05) + self.assertTrue(ds.is_running()) + with self.assertRaises(RuntimeError): + ds.start_detached(poll_interval=0.1) + finally: + ds.stop(timeout=5) + if ds.is_alive(pid): + os.kill(pid, signal.SIGKILL) + + def test_stop_returns_false_when_nothing_is_running(self): + self.assertFalse(ds.stop(timeout=1)) + + +if __name__ == "__main__": + unittest.main()