"""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()