"""Real high-numbered pidfds with synthetic service observations; no service effects."""
import fcntl
import os
import socket
import resource
from pathlib import Path
import sys
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import c01_admission as admission


class AdmissionDescriptorTests(unittest.TestCase):
    def setUp(self):
        original = resource.getrlimit(resource.RLIMIT_NOFILE)
        self.assertTrue(original[1] == resource.RLIM_INFINITY or original[1] >= 4096)
        self.addCleanup(resource.setrlimit, resource.RLIMIT_NOFILE, original)
        if original[0] != resource.RLIM_INFINITY and original[0] < 4096:
            # Fixture setup must work under systemd's default soft limit too;
            # do not depend on the production helper to construct its regression.
            resource.setrlimit(resource.RLIMIT_NOFILE, (4096, original[1]))
        low = os.pidfd_open(os.getpid())
        try:
            self.pidfd = fcntl.fcntl(low, fcntl.F_DUPFD_CLOEXEC, 1024)
        finally:
            os.close(low)
        self.addCleanup(self.cleanup_fd)
        contract = admission.Contract.__new__(admission.Contract)
        contract.policy_path = Path('/fixture-only/admission.json')
        contract.raw = b'fixture-policy'
        contract.assets = {}; contract.paths = {}; contract.initial = None
        contract.processes = {'wg': (os.getpid(), self.pidfd)}
        contract.value = {'version': 1, 'services': {name: 'running' if name == 'wg' else 'uninstalled' for name in admission.UNITS}}
        blocks = []
        for name, unit in admission.UNITS.items():
            fields = {'Id': unit, 'LoadState': 'loaded' if name == 'wg' else 'not-found',
                      'ActiveState': 'active' if name == 'wg' else 'inactive',
                      'MainPID': str(os.getpid()) if name == 'wg' else '0',
                      'InvocationID': 'fixture-only', 'DropInPaths': '',
                      'FragmentPath': admission.SERVICE_ASSETS['wg']['wg_unit'] if name == 'wg' else ''}
            blocks.append('\n'.join(key + '=' + value for key, value in fields.items()))
        def command(name, args):
            if name == 'systemctl':
                return '\n\n'.join(blocks)
            if args == ['show', 'interfaces']:
                return 'wg0\n'
            self.assertEqual(args, ['show', 'wg0', 'public-key'])
            return 'fixture-public-key\n'
        contract.command = command
        self.contract = contract

    def cleanup_fd(self):
        if self.pidfd is not None:
            os.close(self.pidfd)

    def check(self):
        with patch.object(admission.c01_network_owner, 'protected_read', return_value=b'fixture-policy'), \
             patch.object(admission.c01_network, 'process_namespace', return_value=(1, 2)), \
             patch.object(Path, 'resolve', lambda path: path), \
             patch.object(Path, 'read_text', return_value='fixture-public-key\n'):
            self.contract.check()

    def test_live_pidfd_beyond_select_limit_is_accepted(self):
        self.assertGreaterEqual(self.pidfd, 1024)
        self.check()
        self.assertIsNotNone(self.contract.initial)

    def test_invalid_high_pidfd_still_fences_admission(self):
        os.close(self.pidfd); self.pidfd = None
        with self.assertRaisesRegex(ValueError, 'service process lifetime'):
            self.check()

    def test_high_control_channel_data_and_peer_loss_remain_observable(self):
        left, right = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
        try:
            fd = fcntl.fcntl(left.fileno(), fcntl.F_DUPFD_CLOEXEC, 1024)
            with socket.socket(fileno=fd) as high:
                self.assertFalse(admission.read_ready(high))
                right.sendall(b'fixture-pulse')
                self.assertTrue(admission.read_ready(high))
                self.assertEqual(high.recv(32), b'fixture-pulse')
                self.assertFalse(admission.read_ready(high))
                right.close()
                self.assertTrue(admission.read_ready(high))
                self.assertEqual(high.recv(32), b'')
        finally:
            left.close(); right.close()

    def test_default_soft_budget_is_raised_without_changing_hard_allowance(self):
        original = resource.getrlimit(resource.RLIMIT_NOFILE)
        self.assertTrue(original[1] == resource.RLIM_INFINITY or original[1] >= 4096)
        try:
            resource.setrlimit(resource.RLIMIT_NOFILE, (1024, original[1]))
            admission.prepare_descriptor_budget()
            self.assertEqual(resource.getrlimit(resource.RLIMIT_NOFILE), (4096, original[1]))
        finally:
            resource.setrlimit(resource.RLIMIT_NOFILE, original)

    def test_smaller_hard_budget_refuses_without_escalating_it(self):
        with patch.object(resource, 'getrlimit', return_value=(1024, 2048)), \
             patch.object(resource, 'setrlimit') as set_limit:
            with self.assertRaisesRegex(ValueError, 'descriptor budget'):
                admission.prepare_descriptor_budget()
            set_limit.assert_not_called()

    def test_larger_existing_soft_budget_is_preserved(self):
        with patch.object(resource, 'getrlimit', return_value=(65536, 65536)), \
             patch.object(resource, 'setrlimit') as set_limit:
            admission.prepare_descriptor_budget()
            set_limit.assert_not_called()


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