278 lines
16 KiB
Python
278 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Review fixes against production APIs; every OS command is fixture-scoped."""
|
|
import copy
|
|
import os
|
|
import pathlib
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import importlib.util
|
|
|
|
spec = importlib.util.spec_from_file_location('corrections', pathlib.Path(__file__).with_name('d09-review-corrections.py'))
|
|
corrections = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(corrections)
|
|
owner, store, wire, client = corrections.owner, corrections.store, corrections.wire, corrections.client
|
|
worker = corrections.fixture.worker
|
|
|
|
|
|
class ReviewFixes(corrections.Corrections):
|
|
def record(self, action):
|
|
v = self.o.store.value
|
|
index = {k: self.binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')}
|
|
return dict(index, schema_version=1, predecessor_epoch_id=v['epoch'], head_sequence=v['sequence'],
|
|
observed_boot_id=owner.boot_id(), preserved_head_digest=wire.digest(wire.canonical(v['receipt'])) if v['receipt'] else wire.digest(self.o.store.raw),
|
|
preserved_journal_digest=wire.digest(wire.canonical({'receipts': v['journal'], 'intents': v['intents']})),
|
|
preserved_consumed_history_digest=wire.digest(wire.canonical(owner.consumption_history(v))),
|
|
preserved_target_index_digest=wire.digest(wire.canonical(index)), new_epoch_id='a' * 32,
|
|
new_grant_digest='8' * 64, execution_grant_digest='9' * 64,
|
|
external_history_digest='f' * 64, external_history_sequence=1, allowed_action=action,
|
|
disposition_digest=self.o.drained_disposition(v))
|
|
|
|
def reactivate(self, record):
|
|
store.atomic(self.root / 'activation.json', wire.canonical(record))
|
|
old = self.o.store
|
|
result = self.o.readmit(activation=True)
|
|
return result
|
|
|
|
def test_selected_release_mismatch_and_missing_refuse_before_transport(self):
|
|
before = self.o.store.raw
|
|
other = copy.deepcopy(self.binding['release_identity'])
|
|
other[0] = 'C'
|
|
for selected in (None, other, ['B', []]):
|
|
with self.subTest(selected=selected), self.assertRaises((ValueError, PermissionError)):
|
|
client.convenience(self.binding, 'prepare', None, selected)
|
|
self.assertEqual(self.sent, [])
|
|
self.assertEqual(before, self.o.store.raw)
|
|
self.assertFalse(client.CACHE.exists())
|
|
request = dict(self.case.request(), selected_release_digest=wire.release_digest(other))
|
|
self.assertEqual(self.case.submit(request)['reason'], 'COMPOSITION_MISMATCH')
|
|
self.assertEqual(before, self.o.store.raw)
|
|
|
|
def test_identity_domain_duplicate_size_and_order_rejected(self):
|
|
item = ['archipelago', 'B', 'e' * 64, 1]
|
|
self.assertEqual(wire.release_digest(['B', [item]]), wire.digest(wire.canonical(['B', [item]])))
|
|
for components in ([item, item], [item[:-1] + [True]], [item[:-1] + [0]],
|
|
[['z', 'B', 'e' * 64, 1], item]):
|
|
with self.assertRaises(ValueError):
|
|
wire.release_digest(['B', components])
|
|
|
|
def test_sequence_zero_external_reactivation_preserves_consumption_and_replay(self):
|
|
self.prerequisite_fixture()
|
|
self.o.activate_prerequisite()
|
|
consumed = copy.deepcopy(self.o.store.value['prerequisite'])
|
|
self.o.hold()
|
|
self.o.incarnation = 'b' * 32
|
|
self.o.store.anchor = None
|
|
record = self.record('REACTIVATE_PREREQUISITE')
|
|
self.assertTrue(self.reactivate(record)['readmitted'])
|
|
self.o.observe_prerequisite()
|
|
self.assertEqual(self.o.store.value['prerequisite_history'], [consumed])
|
|
self.assertIn(consumed['grant_digest'], self.o.store.value['grants'])
|
|
self.assertEqual(self.o.store.value['sequence'], 0)
|
|
self.assertEqual(self.o.store.value['prerequisite']['state'], 'STARTED')
|
|
before = self.o.store.raw
|
|
self.reactivate(record)
|
|
self.assertEqual(self.o.store.raw, before)
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 2)
|
|
admitted = dict(self.binding, grants={k: record['new_grant_digest'] for k in self.binding['grants']})
|
|
self.assertEqual(client.convenience(admitted, 'prepare', None, self.binding['release_identity'], 'a' * 32)['status'], 'ACCEPTED')
|
|
with self.assertRaisesRegex(PermissionError, 'another staging generation'):
|
|
client.convenience(admitted, 'prepare', None, self.binding['release_identity'], 'b' * 32)
|
|
|
|
def test_sequence_zero_before_consumption_and_lost_start_ack(self):
|
|
self.prerequisite_fixture()
|
|
self.o.store.anchor = None
|
|
record = self.record('REACTIVATE_PREREQUISITE')
|
|
original = owner.command.side_effect
|
|
def lost(*args):
|
|
result = original(*args)
|
|
if args[1] == 'start':
|
|
raise TimeoutError('lost external activation acknowledgement')
|
|
return result
|
|
with patch.object(owner, 'command', side_effect=lost), self.assertRaises(TimeoutError):
|
|
self.reactivate(record)
|
|
self.assertEqual(self.o.store.value['prerequisite']['state'], 'START_INTENT')
|
|
self.assertEqual(self.o.store.value['prerequisite_history'], [None])
|
|
self.reactivate(record)
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 1)
|
|
self.o.observe_prerequisite()
|
|
self.assertFalse(owner.PERMIT.exists())
|
|
|
|
def test_zero_reconciliation_rejects_stale_history_boot_grant_and_unknown_job(self):
|
|
self.prerequisite_fixture()
|
|
record = self.record('REACTIVATE_PREREQUISITE')
|
|
before = self.o.store.raw
|
|
for field, value in [('preserved_head_digest', '0' * 64), ('preserved_consumed_history_digest', '0' * 64),
|
|
('observed_boot_id', '1' * 32), ('execution_grant_digest', '0' * 64),
|
|
('disposition_digest', '0' * 64)]:
|
|
with self.subTest(field=field), self.assertRaises(ValueError):
|
|
self.reactivate(dict(record, **{field: value}))
|
|
self.assertEqual(self.o.store.raw, before)
|
|
self.state['Job'] = '77 /org/freedesktop/systemd1/job/77'
|
|
with self.assertRaises(ValueError):
|
|
self.reactivate(record)
|
|
self.assertIsNone(self.o.store.value['prerequisite'])
|
|
self.state['Job'] = '0 /'
|
|
store.atomic(self.root / 'revocation.json', wire.canonical({'grant_digest': '0' * 64}))
|
|
with self.assertRaisesRegex(ValueError, 'revocation'):
|
|
self.reactivate(record)
|
|
self.assertIsNone(self.o.store.value['prerequisite'])
|
|
|
|
def test_prerequisite_live_process_blocks_even_if_manager_claims_drained(self):
|
|
self.prerequisite_fixture()
|
|
self.o.activate_prerequisite()
|
|
self.o.hold()
|
|
previous = self.o.store.value['prerequisite']['observation']['process']
|
|
with patch.object(owner, 'boot_id', return_value=previous['boot_id']), patch.object(owner, 'process', return_value=previous):
|
|
with self.assertRaisesRegex(ValueError, 'prerequisite process still exists'):
|
|
self.o.drained_disposition(self.o.store.value)
|
|
|
|
def retained(self, action, phase, loss=True):
|
|
paths = self.prerequisite_fixture()
|
|
self.o.activate_prerequisite()
|
|
self.case.submit(self.case.request())
|
|
v = copy.deepcopy(self.o.store.value)
|
|
v['consumed'].extend(v['inbox'])
|
|
v['inbox'] = []
|
|
for name, path in paths.items():
|
|
self.binding['objects'][name]['new' if action == 'APPLY' else 'restore'] = owner.object_digest(path)
|
|
self.binding['effective_config'] = {a: wire.digest(b'prereq-effective') for a in ('APPLY', 'RESTORE')}
|
|
self.binding['artifacts'] = self.binding['restore_artifacts'] = {'backend': wire.digest(b'prerequisite backend'), 'frontend': 'd' * 64}
|
|
r = dict(v['receipt'], phase=phase, outcome='COMPLETE' if phase == 'COMPLETE' else 'PENDING',
|
|
config_action=action, config_pre_digest='d' * 64, config_post_digest='e' * 64,
|
|
config_current_digest=owner.object_digest(paths['config']), effective_config_observation_digest=wire.digest(b'prereq-effective'),
|
|
observed_invocation_id='7' * 32, namespace_observation_digest='a' * 64,
|
|
preservation_observation_digest='b' * 64,
|
|
executor=dict(pid=123, start_ticks=1, boot_id='3' * 32, incarnation_id='4' * 32))
|
|
self.o.store.commit(v, r)
|
|
if loss:
|
|
self.o.hold()
|
|
self.o.store.anchor = None
|
|
return paths
|
|
|
|
def successor_record(self, binding):
|
|
v = self.o.store.value
|
|
index = {k: self.binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')}
|
|
record = dict(index, schema_version=1, predecessor_epoch_id=v['epoch'], head_sequence=v['sequence'],
|
|
observed_boot_id=owner.boot_id(), preserved_head_digest=wire.digest(wire.canonical(v['receipt'])),
|
|
preserved_journal_digest=wire.digest(wire.canonical({'receipts': v['journal'], 'intents': v['intents']})),
|
|
preserved_consumed_history_digest=wire.digest(wire.canonical(owner.consumption_history(v))),
|
|
preserved_target_index_digest=wire.digest(wire.canonical(index)), terminal_disposition='APPLY',
|
|
external_history_digest='f' * 64, external_history_sequence=1,
|
|
successor_binding_digest=wire.digest(wire.canonical(binding)))
|
|
store.atomic(self.root / 'successor-binding.json', wire.canonical(binding))
|
|
store.atomic(self.root / 'successor.json', wire.canonical(record))
|
|
return record
|
|
|
|
def test_successor_rejects_nonterminal_stale_replay_and_wrong_rollback(self):
|
|
self.retained('APPLY', 'COMPLETE', loss=False)
|
|
candidate = copy.deepcopy(self.binding)
|
|
candidate.update(transaction_id='a' * 32, admission_epoch_id='b' * 32,
|
|
grants={k: ('%064x' % (i + 1)) for i, k in enumerate(self.binding['grants'])})
|
|
candidate['prerequisite']['grant_digest'] = '6' * 64
|
|
for obj in candidate['objects'].values():
|
|
obj['old'] = obj['restore'] = obj['new']
|
|
cases = []
|
|
bad = copy.deepcopy(candidate)
|
|
bad['objects']['backend']['restore'] = '0' * 64
|
|
cases.append(bad)
|
|
bad = copy.deepcopy(candidate)
|
|
bad['grants']['UPDATER'] = self.binding['grants']['UPDATER']
|
|
cases.append(bad)
|
|
cases.append(dict(candidate, transaction_id=self.binding['transaction_id']))
|
|
before = self.o.store.raw
|
|
for bad in cases:
|
|
self.successor_record(bad)
|
|
with self.assertRaises(ValueError):
|
|
self.o.admit_successor()
|
|
self.assertEqual(before, self.o.store.raw)
|
|
record = self.successor_record(candidate)
|
|
store.atomic(self.root / 'successor.json', wire.canonical(dict(record, preserved_consumed_history_digest='0' * 64)))
|
|
with self.assertRaisesRegex(ValueError, 'history binding'):
|
|
self.o.admit_successor()
|
|
self.o.store.commit(self.o.store.value, dict(self.o.store.value['receipt'], phase='AWAITING_OBSERVATION', outcome='PENDING'))
|
|
self.successor_record(candidate)
|
|
with self.assertRaisesRegex(ValueError, 'terminal'):
|
|
self.o.admit_successor()
|
|
|
|
def test_successor_atomic_publication_failure_retains_predecessor(self):
|
|
self.retained('APPLY', 'COMPLETE', loss=False)
|
|
before = self.o.store.raw
|
|
candidate = dict(self.binding, transaction_id='a' * 32, admission_epoch_id='b' * 32)
|
|
with patch.object(store, 'atomic', side_effect=OSError('head publication fault')), self.assertRaises(OSError):
|
|
self.o.store.commit_successor(candidate, 'f' * 64, self.o.store.value['prerequisite'])
|
|
self.assertEqual((self.root / 'ledger.json').read_bytes(), before)
|
|
self.assertEqual(self.o.store.binding['transaction_id'], self.binding['transaction_id'])
|
|
self.assertIsNone(self.o.store.anchor)
|
|
|
|
def test_effective_config_ignores_run_status_but_not_command(self):
|
|
prefix = 'ExecStart={ path=/usr/local/bin/archipelago ; argv[]=/usr/local/bin/archipelago ; ignore_errors=no'
|
|
before = prefix + ' ; start_time=[n/a] ; stop_time=[n/a] ; pid=0 ; code=(null) ; status=0/0 }'
|
|
after = prefix + ' ; start_time=[Sat 2026-10-31] ; stop_time=[Sat 2026-10-31] ; pid=42 ; code=exited ; status=0/0 }'
|
|
with patch.object(owner, 'command', return_value=before):
|
|
expected = owner.effective_configuration()
|
|
with patch.object(owner, 'command', return_value=after):
|
|
self.assertEqual(expected, owner.effective_configuration())
|
|
with patch.object(owner, 'command', return_value=after.replace('/usr/local/bin/archipelago', '/foreign')):
|
|
self.assertNotEqual(expected, owner.effective_configuration())
|
|
|
|
def test_retained_applied_complete_lost_start_ack_observes_without_restart(self):
|
|
self.retained('APPLY', 'COMPLETE')
|
|
record = self.record('REACTIVATE_RETAINED')
|
|
self.reactivate(record)
|
|
self.assertEqual(self.o.store.value['receipt']['phase'], 'REACTIVATION_READY')
|
|
self.assertIsNone(self.o.store.value['receipt']['namespace_observation_digest'])
|
|
w = worker.Worker.__new__(worker.Worker)
|
|
w.binding, w.value = self.binding, self.o.store.value
|
|
w.transaction = self.binding['transaction_id']
|
|
def transition(phase, intent=None, **fields):
|
|
v = copy.deepcopy(self.o.store.value)
|
|
v['intent'] = intent
|
|
if phase == 'AWAITING_OBSERVATION':
|
|
v['activation'] = None
|
|
self.o.store.commit(v, dict(v['receipt'], phase=phase, **fields))
|
|
w.value = self.o.store.value
|
|
w.transition = transition
|
|
w.refresh = lambda: None
|
|
original = owner.command.side_effect
|
|
def command(*args):
|
|
if args[1] == 'start':
|
|
self.assertEqual(w.value['receipt']['phase'], 'START_INTENT')
|
|
self.state.update(ActiveState='active', MainPID='4242', InvocationID='6' * 32)
|
|
self.commands.append(args)
|
|
raise TimeoutError('lost manager activation acknowledgement')
|
|
return original(*args)
|
|
with patch.object(owner, 'command', side_effect=command):
|
|
with self.assertRaises(TimeoutError):
|
|
w.reactivate()
|
|
self.assertEqual(w.value['receipt']['phase'], 'START_INTENT')
|
|
w.reactivate()
|
|
self.assertEqual(w.value['receipt']['phase'], 'AWAITING_OBSERVATION')
|
|
self.assertNotEqual(w.value['receipt']['outcome'], 'COMPLETE')
|
|
before = self.o.store.raw
|
|
self.reactivate(record)
|
|
self.assertEqual(before, self.o.store.raw)
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 2)
|
|
|
|
def test_restored_awaiting_and_complete_reactivate_but_old_request_does_not(self):
|
|
self.retained('RESTORE', 'AWAITING_OBSERVATION')
|
|
request = next(iter(self.o.store.value['requests'].values()))['request']
|
|
self.reactivate(self.record('REACTIVATE_RETAINED'))
|
|
before = self.o.store.raw
|
|
self.assertEqual(self.case.submit(request)['status'], 'ACCEPTED')
|
|
self.assertEqual(before, self.o.store.raw)
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 1)
|
|
|
|
def test_retained_foreign_object_refuses_before_start(self):
|
|
paths = self.retained('RESTORE', 'COMPLETE')
|
|
self.reactivate(self.record('REACTIVATE_RETAINED'))
|
|
paths['backend'].write_bytes(b'foreign')
|
|
w = worker.Worker.__new__(worker.Worker)
|
|
w.binding, w.value = self.binding, self.o.store.value
|
|
with self.assertRaisesRegex(ValueError, 'retained generation'):
|
|
w.reactivate()
|
|
self.assertEqual(sum(c[1] == 'start' for c in self.commands), 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|