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