#!/usr/bin/env python3
"""Real C01 Children FD sweep -> credential drop -> native FD14 remap.

No hardware and no storage keys. The disposable native probe checks the channel
and absence of authority FD13; full native protocol checks run separately.
"""
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
import time

assert os.geteuid()==0 and os.environ.get('ARCHY_RADIO_FIXTURE')=='1'
repo=Path(__file__).resolve().parents[2];sys.path.insert(0,str(repo/'scripts'))
import c01_children
import c01_radio
import c01_radio_owner
import c01_supervisor

with tempfile.TemporaryDirectory() as temporary:
    root=Path(temporary);root.chmod(0o755)
    binary=root/'probe'
    subprocess.run(['/usr/bin/cc',str(repo/'scripts/tests/radio_handoff_fixture.c'),'-o',str(binary)],check=True)
    executable=os.open(binary,os.O_RDONLY|os.O_CLOEXEC)
    address=str(root/'radio.sock')
    peer=os.fork()
    if peer==0:
        listener=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET)
        listener.setsockopt(socket.SOL_SOCKET,socket.SO_PASSCRED,1)
        listener.bind(address);listener.listen(1)
        channel,_=listener.accept();channel.settimeout(5)
        import struct
        data,ancillary,_,_=channel.recvmsg(16,socket.CMSG_SPACE(12))
        credentials=[struct.unpack('3i',value) for level,kind,value in ancillary
                     if (level,kind)==(socket.SOL_SOCKET,socket.SCM_CREDENTIALS)]
        assert data==b'FD14' and len(credentials)==1 and credentials[0][1:]==(53001,53001)
        channel.send(b'OK');os._exit(0)
    end=time.monotonic()+5
    while not Path(address).exists():
        assert time.monotonic()<end;time.sleep(.01)
    channel=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET);channel.connect(address)
    children=c01_children.Children()
    parent=os.getpid()
    def body(control):
        c01_supervisor.close_except({control.fileno(),executable,channel.fileno()})
        c01_children.drop_credentials(53001,53001,parent)
        c01_supervisor.native(executable,'c01-managed-v7',{14:channel.fileno()},{'PATH':'/nonexistent'})
    try:
        assert c01_radio.application_fds((executable,),None)==(executable,), 'v6 ABI retention changed'
        children.spawn('application',body,c01_radio.application_fds((executable,),channel))
        channel.close()
        while children.children:
            assert not children.poll()
            assert time.monotonic()<end;time.sleep(.01)
        assert children.results['application']==0,children.results
        assert os.waitstatus_to_exitcode(os.waitpid(peer,0)[1])==0
        children.close()
        print('PASS actual Children pre-callback sweep retains only appointed radio capability; credential drop/native mapping supplies FD14, not FD13',flush=True)
    finally:
        os.close(executable)