#!/usr/bin/env python3 """Synthetic transactional controls for the #58 fleet source-alias repair.""" import importlib.util import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location('unify', Path(__file__).with_name('unify-fleet-goal.py')) u = importlib.util.module_from_spec(spec) spec.loader.exec_module(u) class OwnershipTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='goal58-transaction-') self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) for p in (u.SHARED, *u.ALIASES): directory = self.root / p directory.mkdir(parents=True) (directory / 'index.ts').write_text('// fixture ' + p) launcher = self.root / u.LAUNCHER launcher.parent.mkdir(parents=True) before = (b'#!/bin/bash\nARGS=()\n' b'# Issue #57: Resume uses the operator-selected shared NG goal\n' b'if false; then :; fi\nfor f in "${PROMPT_FILES[@]}"; do :; done\n') after = b'#!/bin/bash\nARGS=()\nfor f in "${PROMPT_FILES[@]}"; do :; done\n' launcher.write_bytes(before) launcher.chmod(0o755) self.addCleanup(patch.stopall) patch.object(u, 'HOTFIX_SHA', u.sha(before)).start() patch.object(u, 'ORIGINAL_SHA', u.sha(after)).start() self.settings = self.root / 'fleet/roles/code/.pi/agent/settings.json' self.settings.parent.mkdir(parents=True) self.settings.write_text(json.dumps({'extensions': ['~/.mosaic/fleet/extensions/goal'], 'packages': []})) link = self.root / 'fleet/agents/joe/.pi/agent/settings.json' link.parent.mkdir(parents=True) link.symlink_to(self.settings) template = self.root / 'fleet/templates/pi-settings.json.template' template.parent.mkdir(parents=True) template.write_text('{"extensions":["~/.mosaic/fleet/extensions/goal"]}') self.env_file = self.root / 'fleet/agents/joe/launch.env' self.env_file.write_text('# synthetic content must not be read\n') self.states = [link.parent / 'goal-state.json', self.root / '.pi/state/goal/goal-state.fixture.json'] for state in self.states: state.parent.mkdir(parents=True, exist_ok=True) state.write_text('owned synthetic state sentinel') for p in u.DEPENDENCIES: target = self.root / p if target.suffix == '.ts': target.parent.mkdir(parents=True, exist_ok=True) target.write_text('// synthetic guard') else: target.mkdir(parents=True, exist_ok=True) (target / 'module.ts').write_text('// synthetic core') self.plan = u.prepare(self.root) self.raw = json.dumps(self.plan).encode() def test_apply_restore_and_state_preservation(self): backup = u.apply(self.plan, self.raw) u.verify_after(self.plan) self.assertEqual((backup / 'planned.json').read_bytes(), self.raw) for state in self.states: self.assertEqual(state.read_text(), 'owned synthetic state sentinel') u.recover(self.plan, backup) u.verify_before(self.plan) def test_partial_failure_reverses_exchange(self): original = u.exchange calls = 0 def exchange(a, b): nonlocal calls calls += 1 if calls == 2: raise OSError('injected second exchange failure') original(a, b) with patch.object(u, 'exchange', exchange): with self.assertRaisesRegex(OSError, 'injected'): u.apply(self.plan, self.raw) u.verify_before(self.plan) self.assertEqual(calls, 3) def test_receipt_recovers_interrupted_partial_deployment(self): backup = u.apply(self.plan, self.raw) # Simulate interruption after aliases changed, before launcher exchange. u.exchange(self.root / u.LAUNCHER, u.stage_path(backup, u.LAUNCHER)) u.recover(self.plan, backup) u.verify_before(self.plan) def test_settings_drift_refuses(self): self.settings.write_text('{}') with self.assertRaisesRegex(RuntimeError, 'inventory drift'): u.apply(self.plan, self.raw) self.assertFalse((self.root / u.ALIASES[0]).is_symlink()) def test_environment_contents_are_never_read(self): original = Path.read_bytes def read_bytes(path): if path.name == 'launch.env': self.fail('environment contents read') return original(path) with patch.object(Path, 'read_bytes', read_bytes): u.verify_before(self.plan) def test_environment_metadata_drift_refuses(self): self.env_file.write_text('# changed synthetic environment\n') with self.assertRaisesRegex(RuntimeError, 'inventory drift'): u.apply(self.plan, self.raw) def test_shared_source_drift_refuses(self): (self.root / u.SHARED / 'index.ts').write_text('drift') with self.assertRaisesRegex(RuntimeError, 'shared source drift'): u.apply(self.plan, self.raw) def test_dependency_drift_refuses(self): (self.root / u.DEPENDENCIES[0] / 'module.ts').write_text('drift') with self.assertRaisesRegex(RuntimeError, 'dependency drift'): u.apply(self.plan, self.raw) def test_target_drift_refuses(self): (self.root / u.ALIASES[0] / 'index.ts').write_text('drift') with self.assertRaisesRegex(RuntimeError, 'before drift'): u.apply(self.plan, self.raw) def test_mode_drift_refuses(self): (self.root / u.ALIASES[0] / 'index.ts').chmod(0o700) with self.assertRaisesRegex(RuntimeError, 'before drift'): u.verify_before(self.plan) def test_nested_symlink_refuses(self): (self.root / u.ALIASES[0] / 'link').symlink_to(self.settings) with self.assertRaisesRegex(RuntimeError, 'symlink/special'): u.prepare(self.root) def test_special_file_refuses(self): os.mkfifo(self.root / u.ALIASES[0] / 'fifo') with self.assertRaisesRegex(RuntimeError, 'symlink/special'): u.prepare(self.root) def test_repeat_apply_refuses(self): u.apply(self.plan, self.raw) with self.assertRaisesRegex(RuntimeError, 'non-canonical source'): u.apply(self.plan, self.raw) def test_rollback_validates_all_pairs_before_effect(self): backup = u.apply(self.plan, self.raw) (u.stage_path(backup, u.LAUNCHER)).write_text('backup drift') with self.assertRaisesRegex(RuntimeError, 'rollback backup drift'): u.recover(self.plan, backup) u.verify_after(self.plan) def test_rollback_refuses_alias_drift(self): backup = u.apply(self.plan, self.raw) target = self.root / u.ALIASES[1] target.unlink() target.symlink_to(self.root / 'unreviewed') with self.assertRaises(RuntimeError): u.recover(self.plan, backup) self.assertTrue(u.is_alias(self.root / u.ALIASES[0], self.root / u.SHARED)) def test_concurrent_deployment_lock_refuses(self): with u.deployment_lock(self.plan): with self.assertRaises(BlockingIOError): with u.deployment_lock(self.plan): self.fail('second writer acquired the lock') def test_unknown_launcher_refuses(self): (self.root / u.LAUNCHER).write_text('# not reviewed\n') with self.assertRaisesRegex(RuntimeError, 'reviewed Resume hotfix'): u.prepare(self.root) if __name__ == '__main__': unittest.main()