"""Appointment refusal and owned-child lifetime tests; no host commands."""
import copy
from pathlib import Path
import sys
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
import c01_uplink as uplink


class UplinkTests(unittest.TestCase):
    def test_host_routing_keeps_ndp_and_kernel_source_selection_without_loopback_mapping(self):
        value={'executable':'/usr/bin/pasta.avx2','sha256':'a'*64,
               'wan':['enp1s0','192.0.2.1','192.0.2.2'],'dns':['1.1.1.1'],
               'routing':'host-dual-stack','dns_host':'100.100.100.100'}
        argv=uplink.arguments(value,'/owned-netns')
        for forbidden in ('-4','--no-icmp','--no-ndp','--outbound','--outbound-if4','--outbound-if6','--map-host-loopback'):
            self.assertNotIn(forbidden,argv)
        self.assertIn('--no-map-gw',argv)
        for flag in ('-t','-u','-T','-U'):
            self.assertEqual(argv[argv.index(flag)+1],'none')
        self.assertEqual(argv[argv.index('--dns-host')+1],'100.100.100.100')
        for change in ({'routing':'arbitrary'},{'dns_host':'127.0.0.1'},
                       {'dns_host':'192.168.1.1'},{'dns':['8.8.8.8']},
                       {'argv':['--map-host-loopback','1.1.1.1']}):
            with self.subTest(change=change),self.assertRaises(ValueError):
                uplink.validate({**value,**change})
        with patch.object(uplink.Path,'read_text',return_value='nameserver 100.100.100.100\n'):
            uplink.check_resolver(value)
        with patch.object(uplink.Path,'read_text',return_value='nameserver 127.0.0.1\n'):
            with self.assertRaisesRegex(ValueError,'no longer configured'):uplink.check_resolver(value)

    def test_appointment_refuses_argv_host_dns_and_unbound_interface(self):
        value={'executable':'/usr/bin/pasta','sha256':'a'*64,
               'wan':['enp1s0','192.0.2.1','192.0.2.2'],'dns':['1.1.1.1']}
        self.assertEqual(uplink.validate(value),value)
        for change in ({'executable':'/bin/sh'}, {'argv':['--map-host-loopback','1.2.3.4']},
                       {'dns':['127.0.0.1']},{'dns':['192.168.1.1']},{'dns':[]},
                       {'wan':['x;command','192.0.2.1','192.0.2.2']},
                       {'wan':['eth0','127.0.0.1','192.0.2.2']}):
            with self.subTest(change=change),self.assertRaises((ValueError,TypeError)):
                uplink.validate({**copy.deepcopy(value),**change})

    def test_pidfd_loss_is_terminal_even_if_poll_has_not_reaped(self):
        owner=uplink.Uplink.__new__(uplink.Uplink)
        owner.child=Mock();owner.child.poll.return_value=None;owner.pidfd=10
        with patch.object(uplink.select,'select',return_value=([10],[],[])):
            with self.assertRaisesRegex(ValueError,'egress exited'):owner.alive()

    def test_second_realm_failure_reaps_first_owned_uplink(self):
        first=Mock()
        with patch.object(uplink,'Uplink',side_effect=[first,ValueError('managed startup failed')]) as factory:
            with self.assertRaisesRegex(ValueError,'managed startup failed'):
                uplink.Uplinks('policy','appointment')
        self.assertEqual([call.args[2] for call in factory.call_args_list],['runtime','managed'])
        first.close.assert_called_once()

    def test_either_realm_lifetime_loss_fails_combined_owner(self):
        runtime,managed=Mock(),Mock()
        with patch.object(uplink,'Uplink',side_effect=[runtime,managed]):
            owner=uplink.Uplinks('policy','appointment')
        managed.check.side_effect=ValueError('managed egress lost')
        with self.assertRaisesRegex(ValueError,'managed egress lost'):owner.check()
        owner.close()
        runtime.close.assert_called_once();managed.close.assert_called_once()

    def test_close_reaps_before_namespace_unpin(self):
        owner=uplink.Uplink.__new__(uplink.Uplink)
        owner.child=Mock();owner.child.poll.return_value=None
        owner.pidfd=10;owner.exe=11;owner.mounted=True;owner.target=Path('/owned-ns');owner.created=[]
        order=[];owner.child.wait.side_effect=lambda **kw:order.append('reaped')
        with patch.object(uplink.os,'close'),patch.object(uplink.LIBC,'umount2',side_effect=lambda *args:order.append('unpin') or 0):
            owner.close()
        self.assertEqual(order,['reaped','unpin'])
        owner.child.terminate.assert_called_once()

    def test_live_pid_with_different_executable_is_not_the_appointed_producer(self):
        owner=uplink.Uplink.__new__(uplink.Uplink)
        owner.child=Mock(pid=321);owner.child.poll.return_value=None
        owner.pidfd=10;owner.exe_identity=(1,2,3,4,5)
        with patch.object(uplink.select,'select',return_value=([],[],[])),patch.object(uplink.os,'stat',return_value=Mock(st_dev=1,st_ino=9)):
            with self.assertRaisesRegex(ValueError,'executable changed'):owner.alive()


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