"""Source contracts only. Never creates accounts, namespaces, grants or units."""
import os
from pathlib import Path
import sys
import tempfile
import types
import unittest
from unittest.mock import patch, Mock

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import c01_network as network
import c01_supervisor as supervisor
import c01_access_kernel as kernel


class NetworkTests(unittest.TestCase):
    def test_kernel_type_refuses_regular_file_and_mount_namespace(self):
        for path in ('/dev/null', '/proc/self/ns/mnt'):
            fd = os.open(path, os.O_RDONLY)
            try:
                with self.assertRaises((ValueError, OSError)):
                    network.identity(fd)
            finally:
                os.close(fd)
        fd = os.open('/proc/self/ns/net', os.O_RDONLY)
        try:
            self.assertEqual(network.identity(fd), network.process_namespace(os.getpid()))
        finally:
            os.close(fd)

    def test_host_or_unappointed_private_namespace_refuses(self):
        with patch.object(network, 'appointed', return_value=(1, 2)):
            with patch.object(network, 'process_namespace', return_value=(1, 2)), self.assertRaisesRegex(ValueError, 'host'):
                network.require_private(1234, 'managed')
            with patch.object(network, 'process_namespace', side_effect=[(1, 3), (1, 4)]), self.assertRaisesRegex(ValueError, 'appointed'):
                network.require_private(1234, 'managed')
            with patch.object(network, 'process_namespace', side_effect=[(1, 3), (1, 2)]):
                self.assertEqual(network.require_private(1234, 'managed'), (1, 2))

    def test_supervisor_exact_group_nnp_and_namespace(self):
        policy = types.SimpleNamespace(value={'supervisor_cgroup':'/sys/fs/cgroup/system.slice/archipelago.service'})
        item = dict(pid=1234, start=7, uids=[0]*4, gids=[0]*4, nnp=1,
                    group='/sys/fs/cgroup/system.slice/archipelago.service/supervisor')
        task=Mock(start=7);task.record.return_value=item;task.finish.return_value=True
        task.namespace.return_value=(1,2);task.dependencies=[]
        with patch.object(kernel,'PinnedTask') as pin, patch.object(kernel,'network_appointment',return_value=(1,2)):
            pin.return_value.__enter__.return_value=task
            self.assertEqual(kernel.verify_supervisor(item, policy), (1, 2))
            pin.assert_called_once_with(1234);task.namespace.assert_called_once_with('net')
            for altered in (dict(item, nnp=0), dict(item, group=item['group']+'/other'), dict(item, uids=[1000]*4)):
                with self.assertRaises(ValueError):
                    kernel.verify_supervisor(altered, policy)
            task.finish.return_value=False
            with self.assertRaisesRegex(ValueError,'mandatory supervisor exited'):
                kernel.verify_supervisor(item,policy)


class AdmissionTests(unittest.TestCase):
    def test_persisted_flags_and_duck_typed_leases_cannot_admit(self):
        for owner in (None, types.SimpleNamespace(access_lease=Mock(), runtime_lease=Mock())):
            with self.assertRaises(supervisor.AdmissionUnavailable):
                supervisor.admit({'migration_complete':True,'old_writers_excluded':True}, None, None, owner)

    def test_partial_acquisition_remains_owned_on_runtime_failure(self):
        owner = types.SimpleNamespace(access_lease=None, runtime_lease=None, network_lease=None)
        value = dict(runtime_uid=1000, runtime_gid=1000, runtime_data_dir='/runtime',
            manager_backing='/private-manager/data', snapshot_backing='/private-snapshot/data',
            authority_backing='/private-authority/data')
        app = types.SimpleNamespace(pw_uid=61001, pw_gid=61001)
        authority = types.SimpleNamespace(pw_uid=61002, pw_gid=61002)
        runtime = dict(manager_uid=61001,manager_gid=61001,authority_uid=61002,authority_gid=61002,
            runtime_uid=1000,runtime_gid=1000,runtime_data_dir='/runtime',home='/home/runtime',
            private_parents=['/private-manager','/private-snapshot','/private-authority'])
        access = Mock()
        with patch.object(network, 'require_private'), patch.object(supervisor, 'root_json', return_value=(value,'a'*64)), \
                patch.object(supervisor.c01_access_client, 'AccessLease', return_value=access), \
                patch.object(supervisor.runtime_service, 'load_config', return_value=(runtime,'b'*64)), \
                patch.object(supervisor.c01_runtime_view, 'RuntimeLease', side_effect=ValueError('runtime unavailable')):
            with self.assertRaisesRegex(ValueError, 'runtime unavailable'):
                supervisor.acquire_admission_leases(owner, value, app, authority)
        self.assertIs(owner.access_lease, access)
        self.assertIsNone(owner.runtime_lease)
        access.close.assert_not_called() # retained until Supervisor.retire

    def test_shared_access_directory_has_no_supervisor_cleanup_owner(self):
        path = Path(__file__).resolve().parents[2] / 'image-recipe/configs/70-c01-supervisor.conf.in'
        text = path.read_text()
        values = [line.split('=',1)[1] for line in text.splitlines() if line.startswith('RuntimeDirectory=')]
        self.assertEqual(values, [''])
        self.assertIn('EnvironmentFile=\n', text)
        self.assertIn('ExecStartPre=\n', text)
        self.assertIn('NetworkNamespacePath=/run/netns/archipelago-managed', text)


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