"""Activation guard tests; privileged success is qualified in the installed VM."""
import importlib.util
import json
import os
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).parents[1]))
import os_updater_activate as A


class ActivationGuards(unittest.TestCase):
    def test_owner_policy_allows_required_locks_but_not_os_package_roots(self):
        for root in A.OS_ROOTS:
            paths=A.owner_policy(root).decode().split('ReadWritePaths=')[1].split()
            self.assertEqual(set(paths),{'/var/lib/dpkg/lock-frontend','/var/lib/dpkg/lock',
                str(root/'transactions'),str(root/'executor.lock')})
            self.assertNotIn(str(root),paths)
            self.assertNotIn('/var/lib/dpkg',paths)
        self.assertEqual(A.OWNER_POLICY,A.owner_policy(A.OS_ROOTS[0]))

    def test_only_pinned_fixed_executor_storage_abi_is_accepted(self):
        for root in A.OS_ROOTS:
            raw=f"ROOT = pathlib.Path({str(root)!r})\n".encode()
            with patch.object(A.bootstrap,'read',return_value=raw):
                self.assertEqual(A.declared_root(Path('/pinned-package')),root)
        for raw in (b"ROOT = pathlib.Path('/home/archipelago/os')\n",
                    b"ROOT = pathlib.Path(os.environ['ROOT'])\n",
                    b"ROOT = pathlib.Path('/var/lib/archipelago-os-update')\nROOT = evil()\n"):
            with patch.object(A.bootstrap,'read',return_value=raw), self.assertRaises((ValueError,TypeError)):
                A.declared_root(Path('/pinned-package'))

    def test_unprotected_destination_is_rejected_without_creation(self):
        with tempfile.TemporaryDirectory() as directory:
            destination = Path(directory)/'os-update/plans'
            with self.assertRaisesRegex(ValueError,'custody'):
                A.destination_custody(destination)
            self.assertFalse(destination.parent.exists())

    def test_interrupted_activation_refuses_before_any_command(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root/'activation-interrupted').mkdir()
            (root/'activation-interrupted/INTENT').write_bytes(b'consumed')
            approval = dict(version=1, prepared_sha256='0'*64,package_root='/unused',
                            package_files={},admission_sha256='1'*64,os_inputs={},helper_sha256='2'*64,
                            owner_dropin_sha256=A.bootstrap.digest(A.OWNER_POLICY))
            with patch.object(A, 'BASE', root), patch.object(A.os,'geteuid',return_value=0), patch.object(
                    A,'same',return_value=json.dumps(approval).encode()), patch.object(A,'protected_dir'), patch.object(
                    A.bootstrap,'secure',side_effect=lambda p: os.open(p,os.O_RDONLY)), patch.object(A,'command') as command:
                with self.assertRaisesRegex(ValueError, 'existing activation intent'):
                    A.activate(root/'approval.json','3'*64)
                command.assert_not_called()
            self.assertEqual((root/'activation-interrupted/INTENT').read_bytes(),b'consumed')

    def test_nonroot_never_reads_approval(self):
        with patch.object(A.os, 'geteuid', return_value=1000), patch.object(A, 'same') as read:
            with self.assertRaisesRegex(ValueError, 'root administrator'):
                A.activate(Path('/not-read'), '0'*64)
            read.assert_not_called()

    def test_all_package_writers_must_be_masked_and_drained(self):
        for bad in ('ActiveState=active\nUnitFileState=masked\n',
                    'ActiveState=inactive\nUnitFileState=enabled\n'):
            with patch.object(A, 'command', return_value=bad):
                with self.assertRaisesRegex(ValueError, 'writer exclusion'):
                    A.current_exclusion()
        def state(*args):
            return ('MainPID=42\nActiveState=active\n' if args[2]=='archipelago.service'
                    else 'ActiveState=inactive\nUnitFileState=masked\n')
        with patch.object(A, 'command', side_effect=state):
            with self.assertRaisesRegex(ValueError, 'externally drained'):
                A.current_exclusion()

    def test_unresolved_os_history_not_reinitialized(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root/'transactions').mkdir()
            (root/'transactions/old-authority').mkdir()
            sentinel = root/'transactions/old-authority/status.json'
            sentinel.write_bytes(b'APPLY_INTENT')
            with patch.object(A, 'OS_ROOT', root), patch.object(A, 'OS_ROOTS', (root,)), patch.object(
                    A, 'D09', root/'d09'), patch.object(A, 'protected_dir'), patch.object(
                    A, 'Path', side_effect=lambda p: root/'binding.json' if p=='/etc/archipelago-d09/binding.json' else Path(p)):
                with self.assertRaisesRegex(ValueError, 'existing OS transaction'):
                    A.untouched_authority()
            self.assertEqual(sentinel.read_bytes(), b'APPLY_INTENT')

    def test_d09_history_not_reinitialized(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root/'ledger.json').write_bytes(b'uncertain')
            with patch.object(A, 'D09', root):
                with self.assertRaisesRegex(ValueError, 'existing D09 authority'):
                    A.untouched_authority()
            self.assertEqual((root/'ledger.json').read_bytes(), b'uncertain')

    def test_package_inventory_checks_files_and_directory_aliases(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root/'file').write_bytes(b'code')
            with patch.object(A, 'protected_dir'), patch.object(A.bootstrap, 'read', side_effect=lambda p: p.read_bytes()):
                self.assertEqual(A.package_inventory(root), {'file': A.bootstrap.digest(b'code')})
            (root/'alias').symlink_to(root, target_is_directory=True)
            def protected(path):
                if path.is_symlink():
                    raise ValueError('alias')
            with patch.object(A, 'protected_dir', side_effect=protected), patch.object(A.bootstrap, 'read', side_effect=lambda p: p.read_bytes()):
                with self.assertRaisesRegex(ValueError, 'alias'):
                    A.package_inventory(root)


if __name__ == '__main__':
    unittest.main()