91 lines
4.5 KiB
Python
91 lines
4.5 KiB
Python
"""Plan and owner error-path tests; real PID1 cases are a separate VM fixture."""
|
|
import importlib.util
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
|
from test_os_update import candidate
|
|
from os_update_plan import validate
|
|
import os_update_executor as executor
|
|
|
|
spec = importlib.util.spec_from_file_location('manager_lifecycle_fixture', pathlib.Path(__file__).resolve().parents[2] / 'tests/regression/d09-lifecycle.py')
|
|
fixture = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(fixture)
|
|
|
|
|
|
class ManagerTests(unittest.TestCase):
|
|
def test_existing_acceptance_is_readable_but_no_new_notice_enters_window(self):
|
|
f = fixture.Lifecycle()
|
|
f.setUp()
|
|
self.addCleanup(f.tearDown)
|
|
request = f.request()
|
|
accepted = f.submit(request)
|
|
before = f.o.store.raw
|
|
f.o.manager_window = SimpleNamespace(state='ARMED')
|
|
self.assertEqual(f.submit(request)['acceptance'], accepted['acceptance'])
|
|
self.assertEqual(f.submit(f.request('RECOVER', '6'*32))['reason'], 'DISPATCH_PENDING')
|
|
self.assertEqual(f.o.store.raw, before)
|
|
|
|
def test_failed_manager_observation_denies_handoff_without_listener_exception(self):
|
|
owner = fixture.owner.Owner.__new__(fixture.owner.Owner)
|
|
with patch.object(fixture.owner, 'service', side_effect=subprocess.CalledProcessError(1, ['systemctl'])):
|
|
with self.assertRaises(PermissionError):
|
|
owner.handoff_service()
|
|
owner.manager_blocked = True
|
|
with patch.object(fixture.owner, 'service') as service:
|
|
with self.assertRaises(PermissionError):
|
|
owner.handoff_service()
|
|
service.assert_not_called()
|
|
|
|
def test_unchanged_manager_byte_mismatch_refuses_before_effects(self):
|
|
obj = executor.Executor.__new__(executor.Executor)
|
|
obj.plan = candidate()
|
|
obj.plan['manager_reexec'] = dict(package_version='257.9-1', architecture='amd64',
|
|
expected_executable_sha256='a'*64)
|
|
with patch.object(executor, 'command', return_value='257.9-1\tamd64') as command, patch.object(executor, 'file_hash', return_value='b'*64):
|
|
with self.assertRaisesRegex(ValueError, 'executable differs'):
|
|
obj.preflight()
|
|
command.assert_called_once_with(['/usr/bin/dpkg-query', '-W', '-f=${Version}\t${Architecture}', 'systemd'])
|
|
|
|
def test_reexec_is_an_explicit_bounded_signed_plan_operation(self):
|
|
plan = candidate()
|
|
self.assertNotIn('manager_reexec', validate(plan))
|
|
policy = dict(version=1, operation='debian-systemd-daemon-reexec', package='systemd',
|
|
package_version='257.9-1', architecture='amd64',
|
|
expected_executable_sha256='a'*64, max_unavailable_seconds=120)
|
|
plan['manager_reexec'] = policy
|
|
validate(plan)
|
|
for field, value in [('max_unavailable_seconds', 121), ('max_unavailable_seconds', True),
|
|
('operation', 'ignore-manager-errors'), ('architecture', 'arm64'),
|
|
('package', 'arbitrary')]:
|
|
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
|
|
validate(dict(plan, manager_reexec=dict(policy, **{field:value})))
|
|
|
|
def test_bus_error_and_hold_error_do_not_kill_or_readmit_owner(self):
|
|
f = fixture.Lifecycle()
|
|
f.setUp()
|
|
self.addCleanup(f.tearDown)
|
|
f.o.incarnation = '9'*32
|
|
anchor = f.o.store.anchor
|
|
failure = subprocess.CalledProcessError(1, ['systemctl', 'show'])
|
|
with patch.object(f.o, 'dispatch', side_effect=failure) as dispatch, patch.object(fixture.owner, 'command', side_effect=failure):
|
|
f.o.run_iteration()
|
|
self.assertTrue(f.o.manager_blocked)
|
|
self.assertEqual(f.o.store.value['manager_fault']['state'], 'HELD')
|
|
self.assertIsNotNone(f.o.store.anchor)
|
|
self.assertNotEqual(f.o.store.anchor, anchor) # Durable uncertainty, not reset.
|
|
epoch = f.o.store.value['epoch']
|
|
f.o.manager_readmitted(epoch) # Old/idempotent record grants no clearance.
|
|
self.assertTrue(f.o.manager_blocked)
|
|
f.o.run_iteration()
|
|
dispatch.assert_called_once()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|