178 lines
7.6 KiB
Python
178 lines
7.6 KiB
Python
#!/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()
|