501 lines
27 KiB
Python
501 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Disposable production-path fault checks. No host manager, privilege or network.
|
|
|
|
Uses the actual admission/store/worker implementations, substituting only OS
|
|
principal checks for the unprivileged fixture and system-manager observations.
|
|
"""
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import socket
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
SCRIPTS = pathlib.Path(__file__).resolve().parents[2] / 'scripts'
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import d09_store as store
|
|
import d09_wire as wire
|
|
import d09_worker as worker
|
|
owner = worker.owner
|
|
|
|
|
|
def fixture_secure(path, directory=False):
|
|
"""Same leaf guarantees with the disposable user's UID, no root escalation."""
|
|
path = pathlib.Path(path)
|
|
st = path.lstat()
|
|
if st.st_uid != os.getuid() or st.st_mode & 0o022 or stat.S_ISLNK(st.st_mode):
|
|
raise PermissionError('fixture object')
|
|
if not directory and (not stat.S_ISREG(st.st_mode) or st.st_nlink != 1):
|
|
raise PermissionError('fixture file')
|
|
return os.open(path, os.O_RDONLY | os.O_NOFOLLOW | (os.O_DIRECTORY if directory else 0))
|
|
|
|
|
|
class Lifecycle(unittest.TestCase):
|
|
def setUp(self):
|
|
# Caller chooses TMPDIR outside this test to keep every byte exclusive.
|
|
self.temp = tempfile.TemporaryDirectory(prefix='d09-functional-')
|
|
self.root = pathlib.Path(self.temp.name)
|
|
(self.root / 'observations').mkdir(mode=0o700)
|
|
(self.root / 'payloads').mkdir(mode=0o700)
|
|
self.patches = [patch.object(store, 'secure', fixture_secure),
|
|
patch.object(owner, 'secure', fixture_secure),
|
|
patch.object(worker, 'secure', fixture_secure),
|
|
patch.object(owner, 'ROOT', self.root),
|
|
patch.object(owner, 'RUN', self.root),
|
|
patch.object(owner, 'PERMIT', self.root / 'permit'),
|
|
patch.object(owner, 'command', side_effect=AssertionError('host command forbidden')),
|
|
patch.object(owner, 'service', return_value={'InvocationID': '7' * 32,
|
|
'MainPID': '0', 'ActiveState': 'active', 'Job': '0 /'})]
|
|
for p in self.patches:
|
|
p.start()
|
|
self.binding = dict(schema_version=1, target_binding_digest='a' * 64, composition_digest='b' * 64,
|
|
transaction_id='1' * 32, admission_epoch_id='2' * 32, requester_uid=os.getuid(),
|
|
requester_gid=os.getgid(), grants={route: 'c' * 64 for route in
|
|
('UPDATER', 'AUTOMATIC', 'MANUAL', 'GUARD', 'OBSERVER')})
|
|
self.binding['prerequisite'] = dict(grant_digest='0' * 64, backend_sha256='e' * 64,
|
|
effective_config_digest='f' * 64)
|
|
self.binding['release_identity'] = ['B', [['archipelago', 'B', 'e' * 64, 1]]]
|
|
value = store.initial(self.binding)
|
|
# These transaction tests begin after the separately tested grant-A
|
|
# activation. No production grant/continuity bypass is installed.
|
|
value['prerequisite'] = dict(state='STARTED', grant_digest='0' * 64,
|
|
observation={'invocation_id': '5' * 32})
|
|
value['grants'].append('0' * 64)
|
|
store.atomic(self.root / 'ledger.json', wire.canonical(value))
|
|
self.o = owner.Owner.__new__(owner.Owner)
|
|
import threading
|
|
self.o.mutex = threading.RLock()
|
|
self.o.binding = self.binding
|
|
self.o.boot = '3' * 32
|
|
self.o.store = store.Store(self.root, self.binding, admitted=True)
|
|
self.o.authorize = lambda req, peer: None
|
|
# Admission fixtures below focus on journal/wire semantics. Actual
|
|
# object preflight is covered independently with real disposable files.
|
|
self.o.preflight = lambda: None
|
|
self.peer = (os.getpid(), os.getuid(), os.getgid())
|
|
|
|
def tearDown(self):
|
|
os.close(self.o.store.fd)
|
|
os.close(self.o.store.head_fd)
|
|
for p in reversed(self.patches):
|
|
p.stop()
|
|
self.temp.cleanup()
|
|
|
|
def request(self, operation='APPLY', request_id='4' * 32):
|
|
return dict(schema_version=1, operation=operation, route='UPDATER' if operation == 'APPLY' else 'MANUAL',
|
|
reason='APPLY_REQUEST' if operation == 'APPLY' else 'MANUAL_REQUEST',
|
|
transaction_id='1' * 32, request_id=request_id, requester_invocation_id='5' * 32,
|
|
target_binding_digest='a' * 64, composition_digest='b' * 64, grant_digest='c' * 64,
|
|
expected_sequence=self.o.store.value['sequence'], observer_receipt_digest=None,
|
|
selected_release_digest=wire.release_digest(self.binding['release_identity']) if operation == 'APPLY' else None,
|
|
staging_generation='a' * 32 if operation == 'APPLY' else None)
|
|
|
|
def submit(self, request):
|
|
response = self.o.submit(request, self.peer)
|
|
wire.bind_submit(request, response)
|
|
return response
|
|
|
|
def test_actual_a_retry_returns_immutable_acceptance_with_b_head(self):
|
|
a = self.request()
|
|
acceptance = self.submit(a)['acceptance']
|
|
b = self.request('RECOVER', '6' * 32)
|
|
self.submit(b)
|
|
before = (self.root / 'ledger.json').read_bytes()
|
|
response = self.submit(a)
|
|
self.assertEqual(response['acceptance'], acceptance)
|
|
self.assertEqual(response['receipt']['request_digest'], wire.digest(wire.canonical(b)))
|
|
self.assertEqual(response['receipt']['operation'], 'RECOVER')
|
|
self.assertEqual(before, (self.root / 'ledger.json').read_bytes())
|
|
|
|
def test_duplicate_after_later_journal_head_does_not_consume(self):
|
|
a = self.request()
|
|
self.submit(a)
|
|
r = dict(self.o.store.value['receipt'], phase='DISPATCH_INTENT')
|
|
self.o.store.commit(self.o.store.value, r)
|
|
before = copy.deepcopy(self.o.store.value)
|
|
self.assertEqual(self.submit(a)['receipt']['sequence'], 2)
|
|
self.assertEqual(before, self.o.store.value)
|
|
|
|
def test_changed_body_and_new_stale_key_refuse(self):
|
|
a = self.request()
|
|
self.submit(a)
|
|
changed = dict(a, expected_sequence=1)
|
|
self.assertEqual(self.submit(changed)['reason'], 'IDEMPOTENCY_MISMATCH')
|
|
self.assertEqual(self.submit(dict(a, request_id='6' * 32))['reason'], 'STALE_SEQUENCE')
|
|
|
|
@patch('os.fsync') # Saturation checks admission, not power-loss durability.
|
|
def test_inbox_saturation_keeps_accepted_retries(self, _fsync):
|
|
a = self.request()
|
|
self.submit(a)
|
|
for n in range(1, 64):
|
|
self.assertEqual(self.submit(self.request('RECOVER', '%032x' % n))['status'], 'ACCEPTED')
|
|
before = (self.root / 'ledger.json').read_bytes()
|
|
self.assertEqual(self.submit(self.request('RECOVER', 'f' * 32))['reason'], 'INBOX_FULL')
|
|
self.assertEqual(self.submit(a)['status'], 'ACCEPTED')
|
|
self.assertEqual(before, (self.root / 'ledger.json').read_bytes())
|
|
|
|
def test_owner_restart_never_seeds_anchor(self):
|
|
self.submit(self.request())
|
|
fresh = store.Store(self.root, self.binding)
|
|
with self.assertRaises(store.ContinuityError):
|
|
fresh.check()
|
|
os.close(fresh.fd)
|
|
|
|
def test_zero_sequence_restart_also_requires_bootstrap(self):
|
|
fresh = store.Store(self.root, self.binding)
|
|
with self.assertRaises(store.ContinuityError):
|
|
fresh.check()
|
|
os.close(fresh.fd)
|
|
|
|
def test_coherent_rollback_is_not_cached_success(self):
|
|
a = self.request()
|
|
self.submit(a)
|
|
old = (self.root / 'ledger.json').read_bytes()
|
|
self.submit(self.request('RECOVER', '6' * 32))
|
|
store.atomic(self.root / 'ledger.json', old)
|
|
response = self.submit(a)
|
|
self.assertEqual(response['reason'], 'CONTINUITY_REQUIRED')
|
|
self.assertEqual(response['reservation'], 'UNKNOWN')
|
|
self.assertIsNone(response['receipt'])
|
|
|
|
def test_uncommitted_record_tail_is_preserved_and_not_reexecuted(self):
|
|
a = self.request()
|
|
original = store.atomic
|
|
def crash(path, raw, mode=0o600):
|
|
if pathlib.Path(path).name == 'ledger.json':
|
|
raise OSError('fault after record fsync, before head publication')
|
|
return original(path, raw, mode)
|
|
with patch.object(store, 'atomic', crash):
|
|
with self.assertRaises(OSError):
|
|
self.submit(a)
|
|
self.assertTrue((self.root / 'receipt-0000000000000001.json').exists())
|
|
self.assertEqual(self.submit(a)['reservation'], 'UNKNOWN')
|
|
|
|
def test_missing_store_does_not_initialize(self):
|
|
(self.root / 'ledger.json').unlink()
|
|
with self.assertRaises(FileNotFoundError):
|
|
store.Store(self.root, self.binding)
|
|
self.assertFalse((self.root / 'ledger.json').exists())
|
|
|
|
def test_inspect_challenge_and_current_head(self):
|
|
self.submit(self.request())
|
|
req = dict(schema_version=1, transaction_id=None, target_binding_digest='a' * 64, challenge='d' * 64)
|
|
response = self.o.inspect(req, self.peer)
|
|
wire.bind_inspect(req, response)
|
|
with self.assertRaises(ValueError):
|
|
wire.bind_inspect(dict(req, challenge='e' * 64), response)
|
|
|
|
def test_strict_wire_lexical_and_nested_types(self):
|
|
req = self.request()
|
|
raw = wire.canonical(req)
|
|
self.assertEqual(wire.decode('submit_request', raw), req)
|
|
for bad in (raw + b'\n', b' ' + raw, raw.replace(b'"schema_version":1', b'"schema_version":true'),
|
|
raw.replace(b'"schema_version":1', b'"schema_version":1.0'),
|
|
raw.replace(b'"expected_sequence":0', b'"expected_sequence":-0'),
|
|
raw.replace(b'{', b'{"operation":"APPLY",', 1)):
|
|
with self.subTest(bad=bad), self.assertRaises(ValueError):
|
|
wire.decode('submit_request', bad)
|
|
|
|
def test_mutated_response_cannot_cross_bind(self):
|
|
req = self.request()
|
|
response = self.submit(req)
|
|
for field in ('target_binding_digest', 'composition_digest', 'transaction_id'):
|
|
bad = copy.deepcopy(response)
|
|
bad['receipt'][field] = ('f' * len(bad['receipt'][field]))
|
|
bad['receipt_digest'] = wire.digest(wire.canonical(bad['receipt']))
|
|
with self.assertRaises(ValueError):
|
|
wire.bind_submit(req, bad)
|
|
|
|
def worker_fixture(self):
|
|
self.submit(self.request())
|
|
dest = self.root / 'payloads/web-ui'
|
|
dest.mkdir(mode=0o755)
|
|
(dest / 'index.html').write_text('old')
|
|
(dest / 'index.html').chmod(0o644)
|
|
staged = dest.with_name('web-ui.d09-' + '1' * 32 + '.new')
|
|
staged.mkdir(mode=0o755)
|
|
(staged / 'index.html').write_text('new')
|
|
(staged / 'index.html').chmod(0o644)
|
|
before, after = owner.object_digest(dest), owner.object_digest(staged)
|
|
self.binding['objects'] = {'frontend': {'old': before, 'new': after, 'restore': before}}
|
|
w = worker.Worker.__new__(worker.Worker)
|
|
w.transaction = '1' * 32
|
|
w.binding = self.binding
|
|
w.value = self.o.store.value
|
|
def transition(phase, intent=None, consume=None, **fields):
|
|
v = copy.deepcopy(self.o.store.value)
|
|
v['intent'] = intent
|
|
r = dict(v['receipt'], phase=phase, **fields)
|
|
self.o.store.commit(v, r)
|
|
w.value = self.o.store.value
|
|
w.transition = transition
|
|
w.refresh = lambda: None # live-owner adapter above publishes w.value
|
|
return w, dest, staged
|
|
|
|
def test_checked_frontend_move_and_reopen_after_first_rename(self):
|
|
w, dest, staged = self.worker_fixture()
|
|
original = os.rename
|
|
def crash(src, dst, **kwargs):
|
|
if src == staged.name:
|
|
raise OSError('fault between payload renames')
|
|
return original(src, dst, **kwargs)
|
|
with patch.dict(owner.DESTINATIONS, frontend=dest), patch.object(os, 'rename', crash):
|
|
with self.assertRaises(OSError):
|
|
w.disposition('frontend', 'APPLY')
|
|
self.assertFalse(dest.exists())
|
|
self.assertTrue(staged.exists())
|
|
with patch.dict(owner.DESTINATIONS, frontend=dest):
|
|
w.disposition('frontend', 'APPLY')
|
|
self.assertEqual((dest / 'index.html').read_text(), 'new')
|
|
self.assertEqual((dest.with_name('web-ui.d09-' + '1' * 32 + '.old') / 'index.html').read_text(), 'old')
|
|
|
|
def test_foreign_edit_is_preserved(self):
|
|
w, dest, staged = self.worker_fixture()
|
|
(dest / 'index.html').write_text('operator custom edit')
|
|
with patch.dict(owner.DESTINATIONS, frontend=dest), self.assertRaises(ValueError):
|
|
w.disposition('frontend', 'APPLY')
|
|
self.assertEqual((dest / 'index.html').read_text(), 'operator custom edit')
|
|
self.assertTrue(staged.exists())
|
|
|
|
def test_missing_backend_snapshot_is_not_success(self):
|
|
w, dest, staged = self.worker_fixture()
|
|
(staged / 'index.html').unlink()
|
|
with patch.dict(owner.DESTINATIONS, frontend=dest), self.assertRaises(ValueError):
|
|
w.disposition('frontend', 'APPLY')
|
|
self.assertEqual((dest / 'index.html').read_text(), 'old')
|
|
|
|
def test_complete_requires_independent_observation(self):
|
|
self.submit(self.request())
|
|
r = copy.deepcopy(self.o.store.value['receipt'])
|
|
r.update(phase='COMPLETE', outcome='COMPLETE')
|
|
with self.assertRaises(ValueError):
|
|
self.o.store.commit(self.o.store.value, r)
|
|
|
|
def test_restored_notice_never_repeats_activation(self):
|
|
w, _, _ = self.worker_fixture()
|
|
r = dict(self.o.store.value['receipt'], phase='AWAITING_OBSERVATION', config_action='RESTORE',
|
|
observed_invocation_id='7' * 32)
|
|
self.o.store.commit(self.o.store.value, r)
|
|
w.value = self.o.store.value
|
|
w.execute('fixture-notice', dict(self.request('RECOVER'), reason='MANUAL_REQUEST'))
|
|
self.assertEqual(w.value['receipt']['phase'], 'AWAITING_OBSERVATION')
|
|
w.execute('fixture-notice', dict(self.request('RECOVER'), reason='AUTOMATIC_FAILURE'))
|
|
self.assertEqual(w.value['receipt']['phase'], 'CONFLICT')
|
|
self.assertEqual(w.value['receipt']['outcome'], 'RECOVERY_REQUIRED')
|
|
|
|
def test_revoked_retry_keeps_acceptance_and_rejects_execution(self):
|
|
req = self.request()
|
|
acceptance = self.submit(req)['acceptance']
|
|
store.atomic(self.root / 'revocation.json', wire.canonical(dict(target_binding_digest='a' * 64,
|
|
transaction_id='1' * 32, grant_digest='c' * 64)))
|
|
response = self.submit(req)
|
|
self.assertEqual(response['acceptance'], acceptance)
|
|
self.assertEqual(response['outcome'], 'RECOVERY_REQUIRED')
|
|
self.assertEqual(response['reason'], 'AUTHORITY_REVOKED')
|
|
before = (self.root / 'ledger.json').read_bytes()
|
|
self.submit(req)
|
|
self.assertEqual(before, (self.root / 'ledger.json').read_bytes())
|
|
self.assertEqual(self.submit(self.request('RECOVER', '6' * 32))['reason'], 'AUTHORITY_REVOKED')
|
|
|
|
def test_readmission_changes_epoch_without_rewriting_acceptance(self):
|
|
req = self.request()
|
|
acceptance = self.submit(req)['acceptance']
|
|
v = self.o.store.value
|
|
index = {k: self.binding[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')}
|
|
history = owner.consumption_history(v)
|
|
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(history)),
|
|
preserved_target_index_digest=wire.digest(wire.canonical(index)),
|
|
new_epoch_id='8' * 32, new_grant_digest='d' * 64, execution_grant_digest='e' * 64,
|
|
external_history_digest='f' * 64, external_history_sequence=1,
|
|
allowed_action='RESUME_SAME_TRANSACTION')
|
|
store.atomic(self.root / 'readmission.json', wire.canonical(record))
|
|
self.o.store.anchor = None
|
|
old_store = self.o.store
|
|
self.assertTrue(self.o.readmit()['readmitted'])
|
|
response = self.submit(req)
|
|
self.assertEqual(response['acceptance'], acceptance)
|
|
self.assertEqual(response['receipt']['admission_epoch_id'], '8' * 32)
|
|
before = (self.root / 'ledger.json').read_bytes()
|
|
self.assertTrue(self.o.readmit()['readmitted'])
|
|
self.assertEqual(before, (self.root / 'ledger.json').read_bytes())
|
|
self.assertEqual(self.submit(self.request('RECOVER', '6' * 32))['reason'], 'GRANT_INVALID')
|
|
self.assertEqual(self.submit(dict(self.request('RECOVER', '6' * 32), grant_digest='d' * 64))['status'], 'ACCEPTED')
|
|
|
|
def test_authenticated_independent_observation_can_complete(self):
|
|
self.submit(self.request())
|
|
r = dict(self.o.store.value['receipt'], phase='AWAITING_OBSERVATION',
|
|
config_action='APPLY', config_pre_digest='d' * 64, config_post_digest='e' * 64,
|
|
config_current_digest='e' * 64, effective_config_observation_digest='f' * 64,
|
|
observed_invocation_id='7' * 32,
|
|
executor=dict(pid=os.getpid(), start_ticks=1, boot_id='3' * 32, incarnation_id='9' * 32))
|
|
self.o.store.commit(self.o.store.value, r)
|
|
directory = self.root / 'observations'
|
|
directory.mkdir(mode=0o700, exist_ok=True)
|
|
base = dict(transaction_id='1' * 32, observed_invocation_id='7' * 32)
|
|
namespace = dict(base, default_terminal_created=True, default_terminal_workspace='/home/archipelago/Work',
|
|
namespace_writable=['/home/archipelago/Work', '/home/archipelago/.local/state/archipelago/sessions'])
|
|
preservation = dict(base, preserved={name: {'before': 'a' * 64, 'after': 'a' * 64}
|
|
for name in ('work', 'history', 'custom_fragments')})
|
|
def publish(obj):
|
|
raw = wire.canonical(obj)
|
|
hashed = wire.digest(raw)
|
|
store.atomic(directory / (hashed + '.json'), raw)
|
|
return hashed
|
|
observation = {k: r[k] for k in ('transaction_id', 'target_binding_digest', 'composition_digest',
|
|
'observed_invocation_id', 'effective_config_observation_digest', 'config_current_digest')}
|
|
observation.update(namespace_observation_digest=publish(namespace), preservation_observation_digest=publish(preservation))
|
|
observed = publish(observation)
|
|
req = dict(self.request('ACK_OBSERVATION', '6' * 32), route='OBSERVER', reason='OBSERVATION_READY',
|
|
requester_invocation_id=None, observer_receipt_digest=observed)
|
|
self.assertEqual(self.submit(req)['status'], 'ACCEPTED')
|
|
self.assertEqual(self.o.observation(observed), observation)
|
|
r = dict(self.o.store.value['receipt'], phase='COMPLETE', outcome='COMPLETE',
|
|
namespace_observation_digest=observation['namespace_observation_digest'],
|
|
preservation_observation_digest=observation['preservation_observation_digest'])
|
|
self.o.store.commit(self.o.store.value, r)
|
|
request = dict(schema_version=1, transaction_id=None, target_binding_digest='a' * 64, challenge='d' * 64)
|
|
self.assertEqual(self.o.inspect(request, self.peer)['receipt']['outcome'], 'COMPLETE')
|
|
with patch.object(owner, 'service', return_value={'InvocationID': '8' * 32}):
|
|
self.assertEqual(self.o.inspect(request, self.peer)['status'], 'UNKNOWN')
|
|
|
|
@patch('os.fsync')
|
|
def test_worker_snapshot_remains_bounded_at_full_inbox(self, _fsync):
|
|
self.submit(self.request())
|
|
for n in range(1, 64):
|
|
self.submit(self.request('RECOVER', '%032x' % n))
|
|
result = self.o.worker_snapshot(self.o.store.value)
|
|
self.assertLess(len(wire.canonical(result)), wire.MAX)
|
|
self.assertEqual(len(result['value']['inbox']), 1)
|
|
|
|
def test_no_public_restart_readmit_or_arbitrary_exec(self):
|
|
for op in ('activate', 'readmit', 'execute', 'prepare'):
|
|
with self.subTest(operation=op), self.assertRaises(ValueError):
|
|
wire.decode('submit_request', wire.canonical(dict(self.request(), operation=op)))
|
|
|
|
def test_guard_unknown_is_nonfatal_and_never_downgrades(self):
|
|
import subprocess
|
|
text = (SCRIPTS / 'ota-crash-guard.sh').read_text()
|
|
client = self.root / 'client'
|
|
client.write_text('#!/bin/sh\nexit 3\n')
|
|
client.chmod(0o700)
|
|
text = text.replace('D09_CLIENT=/usr/libexec/archipelago-d09ctl', 'D09_CLIENT=' + str(client))
|
|
# Disable logging and make every legacy mutator fail loudly. No real
|
|
# command can escape the pre-marker unknown-state branch in this fixture.
|
|
text = text.replace('set -u', 'set -u\nsystemd-cat() { :; }\ncp() { exit 98; }\nmv() { exit 98; }\nrm() { exit 98; }')
|
|
result = subprocess.run(['bash'], input=text, capture_output=True, text=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_initial_foreign_preflight_refuses_without_stopping(self):
|
|
import shutil
|
|
w, dest, staged = self.worker_fixture()
|
|
# Fresh empty target authority, with one real admitted fixture object.
|
|
# worker_fixture already accepted one request; use a distinct Owner to
|
|
# test preflight directly and assert the existing head is untouched.
|
|
restore = dest.with_name(dest.name + '.d09-' + '1' * 32 + '.restore')
|
|
shutil.copytree(dest, restore)
|
|
self.binding['helper_digest'] = wire.digest(b'helper')
|
|
read = owner.read
|
|
before = (self.root / 'ledger.json').read_bytes()
|
|
with patch.dict(owner.DESTINATIONS, {'frontend': dest}, clear=True), patch.object(owner, 'read',
|
|
side_effect=lambda p: b'helper' if str(p) == '/usr/libexec/archipelago-d09-prepare-workspace' else read(p)):
|
|
owner.Owner.preflight(self.o)
|
|
(dest / 'index.html').write_text('operator edit')
|
|
with self.assertRaises(ValueError):
|
|
owner.Owner.preflight(self.o)
|
|
self.assertEqual(before, (self.root / 'ledger.json').read_bytes())
|
|
|
|
@patch('os.fsync') # Ordered lifecycle functional check; fsync faults above use real fsync.
|
|
def test_full_apply_and_manual_recovery_own_both_payloads(self, _fsync):
|
|
import shutil
|
|
payloads = self.root / 'payloads'
|
|
paths = {name: payloads / name for name in ('frontend', 'backend', 'config')}
|
|
paths['frontend'].mkdir(mode=0o755)
|
|
(paths['frontend'] / 'index.html').write_text('old frontend')
|
|
(paths['frontend'] / 'index.html').chmod(0o644)
|
|
paths['backend'].write_text('old backend')
|
|
paths['backend'].chmod(0o755)
|
|
objects = {}
|
|
for name, dest in paths.items():
|
|
old = owner.object_digest(dest)
|
|
new = dest.with_name(dest.name + '.d09-' + '1' * 32 + '.new')
|
|
restore = dest.with_name(dest.name + '.d09-' + '1' * 32 + '.restore')
|
|
if name == 'frontend':
|
|
shutil.copytree(dest, restore)
|
|
new.mkdir(mode=0o755)
|
|
(new / 'index.html').write_text('new frontend')
|
|
(new / 'index.html').chmod(0o644)
|
|
elif name == 'backend':
|
|
shutil.copy2(dest, restore)
|
|
new.write_text('new backend')
|
|
new.chmod(0o755)
|
|
else:
|
|
new.write_text('[Service]\nReadWritePaths=/home/archipelago/Work\n')
|
|
new.chmod(0o644)
|
|
objects[name] = dict(old=old, new=owner.object_digest(new), restore=owner.object_digest(restore))
|
|
self.binding.update(objects=objects, helper_digest=wire.digest(b'helper'),
|
|
effective_config={action: wire.digest(b'effective') for action in ('APPLY', 'RESTORE')})
|
|
request = self.request()
|
|
self.submit(request)
|
|
w = worker.Worker.__new__(worker.Worker)
|
|
w.transaction = '1' * 32
|
|
w.binding = self.binding
|
|
w.value = self.o.store.value
|
|
def transition(phase, intent=None, consume=None, **fields):
|
|
v = copy.deepcopy(self.o.store.value)
|
|
v['intent'] = intent
|
|
if consume:
|
|
v['inbox'].remove(consume)
|
|
v['consumed'].append(consume)
|
|
r = dict(v['receipt'], phase=phase, **fields)
|
|
r['executor'] = dict(pid=os.getpid(), start_ticks=1, boot_id='3' * 32, incarnation_id='9' * 32)
|
|
self.o.store.commit(v, r)
|
|
w.value = self.o.store.value
|
|
w.transition = transition
|
|
w.observed_start = lambda action: '7' * 32 if action == 'APPLY' else '8' * 32
|
|
w.refresh = lambda: None
|
|
state = dict(MainPID='42', InvocationID='5' * 32, ActiveState='active', Job='0 /')
|
|
events = []
|
|
def command(*args):
|
|
events.append(args)
|
|
if args[1] == 'stop':
|
|
state.update(MainPID='0', InvocationID='', ActiveState='inactive')
|
|
elif args[1] == 'start':
|
|
state.update(MainPID='43', InvocationID='7' * 32, ActiveState='active')
|
|
elif args[1] == 'show':
|
|
return 'no' if 'NeedDaemonReload' in args else 'effective'
|
|
return ''
|
|
with patch.dict(owner.DESTINATIONS, paths, clear=True), patch.object(owner, 'service', side_effect=lambda: dict(state)), \
|
|
patch.object(owner, 'command', side_effect=command), patch.object(worker, 'read', return_value=b'helper'):
|
|
key = w.value['inbox'][0]
|
|
w.execute(key, request)
|
|
self.assertEqual(w.value['receipt']['phase'], 'AWAITING_OBSERVATION')
|
|
self.assertEqual((paths['frontend'] / 'index.html').read_text(), 'new frontend')
|
|
self.assertEqual(paths['backend'].read_text(), 'new backend')
|
|
self.assertTrue(paths['config'].exists())
|
|
recovery = self.request('RECOVER', '6' * 32)
|
|
self.submit(recovery)
|
|
w.value = self.o.store.value
|
|
w.execute(w.value['inbox'][0], recovery)
|
|
self.assertEqual((paths['frontend'] / 'index.html').read_text(), 'old frontend')
|
|
self.assertEqual(paths['backend'].read_text(), 'old backend')
|
|
self.assertFalse(paths['config'].exists())
|
|
self.assertEqual(w.value['receipt']['backend_restoration'], 'SUCCEEDED')
|
|
self.assertEqual(w.value['receipt']['frontend_restoration'], 'SUCCEEDED')
|
|
self.assertEqual(w.value['receipt']['outcome'], 'PENDING')
|
|
self.assertEqual(len([event for event in events if event[1] == 'start']), 2)
|
|
self.assertEqual(len(w.value['consumed']), 2)
|
|
self.assertTrue(paths['backend'].with_name('backend.d09-' + '1' * 32 + '.old').exists())
|
|
self.assertTrue(paths['backend'].with_name('backend.d09-' + '1' * 32 + '.failed').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|