"""Conditional admission refusals; no service mutations or host test commands."""
import copy
import hashlib
import os
from pathlib import Path
import socket
import sys
import time
from types import SimpleNamespace
import unittest
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
import c01_admission as admission


class AdmissionTests(unittest.TestCase):
    def policy(self):
        return {'version':1,'platform_sha256':'a'*64,'roles':sorted(admission.ROLES),'required_routes':['management'],
                'assets':{k:'b'*64 for k in admission.ASSETS},
                'services':{k:'uninstalled' for k in admission.UNITS},
                'optional_unavailable':sorted(admission.UNITS)}

    def test_required_contract_cannot_be_replaced_by_ready_flags(self):
        value=self.policy();launch={'platform_profile_sha256':'a'*64}
        admission.validate(value,launch)
        for change in ({'ready':True},{'roles':['application']},{'assets':{}},
                       {'platform_sha256':'c'*64},{'optional_unavailable':[]}):
            with self.subTest(change=change),self.assertRaises(ValueError):
                admission.validate({**copy.deepcopy(value),**change},launch)

    def test_running_service_needs_its_executable_and_unit_pins(self):
        value=self.policy();value['services']['fips']='running'
        value['optional_unavailable'].remove('fips')
        with self.assertRaisesRegex(ValueError,'asset set'):
            admission.validate(value,{'platform_profile_sha256':'a'*64})
        value['assets'].update({name:'d'*64 for name in admission.SERVICE_ASSETS['fips']})
        with self.assertRaisesRegex(ValueError,'service route'):
            admission.validate(value,{'platform_profile_sha256':'a'*64})
        value['required_routes'].append('fips-dns')
        admission.validate(value,{'platform_profile_sha256':'a'*64})

    def test_asset_digest_failure_does_not_leak_descriptor(self):
        path=Path('/usr/bin/true')
        before=len(os.listdir('/proc/self/fd'))
        fd,metadata=admission.open_asset(path,hashlib.sha256(path.read_bytes()).hexdigest())
        self.assertEqual(metadata[:2],(path.stat().st_dev,path.stat().st_ino));os.close(fd)
        with self.assertRaisesRegex(ValueError,'digest'):
            admission.open_asset(path,'0'*64)
        self.assertEqual(len(os.listdir('/proc/self/fd')),before)

    def test_installed_fips_identity_can_differ_from_nostr_without_rotation(self):
        # Public-only installed Framework vector: the FIPS seed-derived key is
        # independent of its Nostr identity. This npub is the retained file.
        public=bytes.fromhex('d157c1dd326da3ed03882afa9718937879fe59ccd8e48cc7059b413c071a5cf3')
        npub=b'npub169turhfjdk376qug9tafwxyn0pulukwvmrjge3c9ndqncpc6tnesrn4wa6\n'
        for encoded in (public,public.hex().encode(),npub,npub.upper()):
            admission.require_public_identity(encoded,public)
        foreign=bytes.fromhex('07cb23083180dc86c542b31f95a61670d1d7cb965d53168c5356967515533405')
        with self.assertRaisesRegex(ValueError,'differs'):
            admission.require_public_identity(npub,foreign)
        with self.assertRaises(ValueError):
            admission.require_public_identity(npub[:-3]+b'aa\n',public)
        policy=dict(self.policy(),version=2,fips_public_key=public.hex(),service_argv_sha256={name:None for name in admission.UNITS})
        admission.validate(policy,{'platform_profile_sha256':'a'*64})
        for value in ('0'*64,public.hex().upper(),'bad',None):
            with self.assertRaises(ValueError):
                admission.validate(dict(policy,fips_public_key=value),{'platform_profile_sha256':'a'*64})

    def test_checker_pid_credentials_and_deadline_are_required(self):
        a,b=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET)
        a.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
        hold=admission.Hold.__new__(admission.Hold)
        hold.record=SimpleNamespace(channel=a,pid=os.getpid()+1);hold.deadline=time.monotonic()+3
        try:
            b.send(b'ADMISSION_HELD')
            with self.assertRaisesRegex(ValueError,'credentials'):hold.check()
            hold.deadline=0
            with self.assertRaisesRegex(ValueError,'lifetime'):hold.check()
        finally:a.close();b.close()


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