#!/usr/bin/env python3
"""Exercise consent and real UPnP SOAP actions against native media workers."""
import argparse,json,os,select,socket,subprocess,tempfile,time,urllib.request
from pathlib import Path
from http.server import ThreadingHTTPServer,SimpleHTTPRequestHandler
from threading import Thread
parser=argparse.ArgumentParser();parser.add_argument('--binaries',required=True);parser.add_argument('--upnp',required=True);args=parser.parse_args()
binaries=Path(args.binaries).resolve()
with tempfile.TemporaryDirectory(prefix='framely-media-test-') as directory:
    root=Path(directory);events=socket.socket(socket.AF_UNIX,socket.SOCK_DGRAM);events.bind(str(root/'events.sock'));events.settimeout(5)
    token='0123456789abcdef0123456789abcdef';env=dict(os.environ,GST_PLUGIN_PATH=str(binaries),FRAMELY_CAST_DIR=directory,FRAMELY_CAST_ID=token)
    # Check the actual receive elements, including plugins absent on stock Frame.
    for element in ['framelyvideosink','framelyaudiosink','level','pulsesink','videoflip','souphttpsrc','qtdemux']:
        result=subprocess.run(['gst-inspect-1.0','--exists',element],env=env,capture_output=True)
        assert result.returncode==0,f'Required receive element unavailable: {element}'
    test_name=f'Framely test {os.getpid()}'
    airplay=subprocess.Popen([str(binaries/'uxplay'),'-n',test_name,'-nh','-m','-avdec','-hls','-vs','framelyvideosink','-as','framelyaudiosink'],env=env,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)
    try:
        time.sleep(2);assert airplay.poll() is None,airplay.stdout.read().decode()
        print('PASS: UxPlay starts with bundled decoders and consent sinks')
        # Use the system daemon only to browse. Publication must also work when
        # that daemon's disable-publishing policy is enabled (Frame's default).
        import shutil
        if shutil.which('avahi-browse'):
            for service_type in ['_airplay._tcp','_raop._tcp']:
                browser=subprocess.Popen(['avahi-browse','--resolve','--parsable',
                    '--no-db-lookup',service_type],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,text=True)
                from queue import Queue,Empty
                discovered=Queue()
                def browse_lines():
                    for line in browser.stdout:discovered.put(line)
                Thread(target=browse_lines,daemon=True).start()
                try:
                    deadline=time.monotonic()+15
                    while True:
                        try:line=discovered.get(timeout=max(.01,deadline-time.monotonic()))
                        except Empty:raise AssertionError(f'No resolved {service_type} advertisement')
                        if line.startswith('=;') and test_name in line.replace('\\032',' '):
                            assert 'features=' in line or 'cn=' in line,line
                            break
                        assert time.monotonic()<deadline,f'No resolved {service_type} advertisement'
                finally:
                    browser.terminate();browser.wait(timeout=3)
            print('PASS: AirPlay and RAOP multicast discovery, address resolution and TXT records')
        else:
            print('SKIP: multicast discovery test requires avahi-browse')
    finally:
        airplay.terminate();airplay.wait(timeout=3)
    p=subprocess.Popen(['gst-launch-1.0','-q','videotestsrc','is-live=true','!','video/x-raw,width=320,height=240,framerate=10/1','!','framelyvideosink'],env=env,stderr=subprocess.PIPE)
    try:
        event=json.loads(events.recv(4096));assert event['event']=='video' and event['width']==320,event
        frame=root/(token+'.frame');time.sleep(.2);assert not frame.exists(),'video visible before consent'
        (root/(token+'.accept')).touch();time.sleep(.3);assert frame.stat().st_size==16+320*240*4
    finally:
        p.terminate();p.wait(timeout=3)
        assert not p.stderr.read(), 'GStreamer consent sink emitted errors'
    print('PASS: video consent gate')
    # Exercise the H.264 decoder used by UxPlay desktop mirroring. Raw video
    # alone cannot detect a hardware decoder producing corrupted frames.
    encoded=root/'mirror.h264'
    subprocess.run(['ffmpeg','-hide_banner','-loglevel','error','-f','lavfi','-i','smptebars=size=320x240:rate=30','-t','1','-c:v','libx264','-preset','ultrafast','-pix_fmt','yuv420p','-f','h264',str(encoded)],check=True)
    subprocess.run(['gst-launch-1.0','-q','filesrc',f'location={encoded}','!','h264parse','!','avdec_h264','!','videoconvert','!','framelyvideosink'],env=env,check=True,timeout=10)
    reference=subprocess.check_output(['ffmpeg','-hide_banner','-loglevel','error','-i',str(encoded),'-frames:v','1','-pix_fmt','bgra','-f','rawvideo','pipe:1'])
    pixels=frame.read_bytes()[16:]
    assert len(pixels)==len(reference) and sum(abs(a-b) for a,b in zip(reference,pixels))/len(reference)<2,'AirPlay mirror decoded colors are corrupted'
    print('PASS: AirPlay H.264 mirroring decoder and received colors')
    # Capture only a temporary null sink, not the user's active audio output.
    import array
    sink_name=f'framely_cast_test_{os.getpid()}'
    module=subprocess.check_output(['pactl','load-module','module-null-sink',f'sink_name={sink_name}','rate=48000','channels=2'],text=True).strip()
    env['PULSE_SINK']=sink_name
    audio_token='abcdef0123456789abcdef0123456789ab'
    # Frame spatializes ordinary streams; this test-only node name keeps its
    # own temporary sink route intact without changing system settings.
    audio_env=dict(env,FRAMELY_CAST_ID=audio_token,PULSE_PROP='node.name=framely-test-filter-chain')
    p=subprocess.Popen(['gst-launch-1.0','-q','audiotestsrc','is-live=true','wave=sine','!','audio/x-raw,rate=48000,channels=2','!','framelyaudiosink'],env=audio_env,stderr=subprocess.PIPE)
    def peak():
        raw=subprocess.check_output(['ffmpeg','-hide_banner','-loglevel','error','-f','pulse','-i',sink_name+'.monitor','-t','0.2','-acodec','pcm_s16le','-f','s16le','pipe:1'])
        samples=array.array('h');samples.frombytes(raw);return max(map(abs,samples),default=0)
    try:
        event=json.loads(events.recv(4096))
        while event['id']!=audio_token:event=json.loads(events.recv(4096))
        assert event['event']=='audio',event
        time.sleep(.3);assert peak()<10,'audio audible before consent'
        (root/(audio_token+'.accept')).touch();time.sleep(.3);assert peak()>1000,'audio not audible after consent'
        (root/(audio_token+'.accept')).unlink();time.sleep(.3);assert peak()<10,'audio continues after pause'
        print('PASS: audio consent gate and pause')
    finally:
        p.terminate();p.wait(timeout=3);subprocess.run(['pactl','unload-module',module],check=True)
        assert not p.stderr.read(),'GStreamer audio sink emitted errors'

    # A complete renderer description, with the same XML shipped by Framely.
    xml=root/'upnp';xml.mkdir()
    import shutil
    for source in Path(args.upnp).glob('*.xml'):shutil.copyfile(source,xml/source.name)
    services=''.join(f'<service><serviceType>urn:schemas-upnp-org:service:{s}:1</serviceType><serviceId>urn:upnp-org:serviceId:{s}</serviceId><SCPDURL>/{s}.xml</SCPDURL><controlURL>/{s}/control</controlURL><eventSubURL>/{s}/event</eventSubURL></service>' for s in ['AVTransport','RenderingControl','ConnectionManager'])
    (xml/'device.xml').write_text(f'<root xmlns="urn:schemas-upnp-org:device-1-0"><specVersion><major>1</major><minor>0</minor></specVersion><device><deviceType>urn:schemas-upnp-org:device:MediaRenderer:1</deviceType><friendlyName>Framely test</friendlyName><manufacturer>Framely</manufacturer><modelName>Test</modelName><UDN>uuid:56ce7555-6baf-4360-bbd6-61a32b06b6aa</UDN><serviceList>{services}</serviceList></device></root>')
    subprocess.run(['ffmpeg','-hide_banner','-loglevel','error','-f','lavfi','-i','smptebars=size=320x240:rate=10','-t','8','-c:v','libx264','-pix_fmt','yuv420p','-movflags','+faststart',str(root/'video.mp4')],check=True)
    class Handler(SimpleHTTPRequestHandler):
        def __init__(self,*a,**k):super().__init__(*a,directory=directory,**k)
        def log_message(self,*a):pass
        def send_head(self):
            # Real media servers support HTTP Range; SimpleHTTPRequestHandler does not.
            file=Path(self.translate_path(self.path)).open('rb');size=os.fstat(file.fileno()).st_size
            value=self.headers.get('Range');self.remaining=size
            if value:
                import re
                match=re.fullmatch(r'bytes=(\d+)-(\d*)',value);assert match,value
                start=int(match[1]);end=min(int(match[2]) if match[2] else size-1,size-1)
                assert 0<=start<=end<size
                self.send_response(206);self.send_header('Content-Range',f'bytes {start}-{end}/{size}')
                file.seek(start);self.remaining=end-start+1
            else:self.send_response(200)
            self.send_header('Accept-Ranges','bytes');self.send_header('Content-Type',self.guess_type(str(file.name)))
            self.send_header('Content-Length',str(self.remaining));self.end_headers();return file
        def copyfile(self,source,target):
            while self.remaining:
                chunk=source.read(min(65536,self.remaining))
                if not chunk:break
                target.write(chunk);self.remaining-=len(chunk)
    http=ThreadingHTTPServer(('0.0.0.0',0),Handler);Thread(target=http.serve_forever,daemon=True).start()
    p=subprocess.Popen([str(binaries/'framely-receiver'),directory,str(xml),'1'],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,bufsize=1,env=env)
    from queue import Queue
    replies=Queue()
    def reader():
        for line in p.stdout:replies.put(line)
    Thread(target=reader,daemon=True).start()
    def read():
        return json.loads(replies.get(timeout=10))
    try:
        ready=read();assert ready['event']=='ready',ready
        host=ready['host'];port=ready['port']
        uri=f'http://{host}:{http.server_port}/video.mp4'
        from xml.sax.saxutils import escape
        def soap(action,fields,service='AVTransport'):
            content=''.join(f'<{k}>{escape(str(v))}</{k}>' for k,v in fields.items())
            urn=f'urn:schemas-upnp-org:service:{service}:1'
            body=f'<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"><s:Body><u:{action} xmlns:u="{urn}">{content}</u:{action}></s:Body></s:Envelope>'
            req=urllib.request.Request(f'http://{host}:{port}/{service}/control',data=body.encode(),headers={'Content-Type':'text/xml; charset=utf-8','SOAPAction':f'"{urn}#{action}"'})
            try:
                with urllib.request.build_opener(urllib.request.ProxyHandler({})).open(req,timeout=5) as r:
                    assert r.status==200
                    return r.read().decode()
            except urllib.error.HTTPError as e:raise AssertionError(e.read().decode())
        sequence=0
        def control(kind,id,**fields):
            global sequence
            sequence+=1
            started=time.monotonic()
            p.stdin.write(json.dumps(dict(kind=kind,id=id,request=sequence,**fields))+'\n');p.stdin.flush()
            while True:
                reply=read()
                if reply.get('request')==sequence:
                    assert 'error' not in reply,reply
                    assert time.monotonic()-started<2,'Playback control blocked the receive service'
                    return reply['result']
        soap('SetAVTransportURI',dict(InstanceID=0,CurrentURI=uri,CurrentURIMetaData=''))
        request=read()
        while request['event']=='ready':request=read()
        assert request['event']=='request',request
        id=request['id'];frame=root/(id+'.frame');time.sleep(.2);assert not frame.exists(),'DLNA media played before consent'
        p.stdin.write(json.dumps({'kind':'accept','id':id,'request':1})+'\n');p.stdin.flush()
        response=read();assert response['result'] is True,response
        event=json.loads(events.recv(4096))
        while event['id']!=id:event=json.loads(events.recv(4096))
        assert event['width']==320,event
        time.sleep(.2);assert frame.stat().st_size==16+320*240*4
        # Decode a known static color pattern through the actual receive pipeline.
        # A frame with the right size can still contain green/corrupted pixels.
        import fcntl
        reference=subprocess.check_output(['ffmpeg','-hide_banner','-loglevel','error','-i',str(root/'video.mp4'),'-frames:v','1','-pix_fmt','bgra','-f','rawvideo','pipe:1'])
        with frame.open('rb') as captured:
            fcntl.flock(captured,fcntl.LOCK_SH);captured.read(16);pixels=captured.read()
        assert sum(abs(a-b) for a,b in zip(reference,pixels))/len(reference)<2,'DLNA decoded colors are corrupted'
        status=control('status',id)
        assert 7.9<=status['duration']<=8.1,status
        control('pause',id,paused=True);assert control('status',id)['paused'] is True
        soap('SetVolume',dict(InstanceID=0,Channel='Master',DesiredVolume=37),'RenderingControl')
        assert '<CurrentVolume>37</CurrentVolume>' in soap('GetVolume',dict(InstanceID=0,Channel='Master'),'RenderingControl')
        assert abs(control('status',id)['volume']-.37)<.001
        soap('SetMute',dict(InstanceID=0,Channel='Master',DesiredMute=1),'RenderingControl')
        assert '<CurrentMute>1</CurrentMute>' in soap('GetMute',dict(InstanceID=0,Channel='Master'),'RenderingControl')
        control('seek',id,seconds=2);time.sleep(.15)
        assert control('status',id)['position']>=1.9
        assert '<TrackDuration>00:00:08</TrackDuration>' in soap('GetPositionInfo',dict(InstanceID=0))
        # A second request must leave the first session and frame available.
        soap('SetAVTransportURI',dict(InstanceID=0,CurrentURI=uri,CurrentURIMetaData=''))
        request=read()
        while request.get('event')!='request':request=read()
        second=request['id'];control('accept',second)
        deadline=time.monotonic()+5
        while not (root/(second+'.frame')).exists() and time.monotonic()<deadline:time.sleep(.05)
        assert (root/(second+'.frame')).exists() and frame.exists()
        assert control('status',id)['paused'] is True
        control('stop',second);control('stop',id)
        assert not (root/(id+'.accept')).exists()
        assert '<CurrentTransportState>NO_MEDIA_PRESENT</CurrentTransportState>' in soap('GetTransportInfo',dict(InstanceID=0))
        # Repeated requests/rejections exercise session reclamation beyond the cap.
        for _ in range(18):
            soap('SetAVTransportURI',dict(InstanceID=0,CurrentURI=uri,CurrentURIMetaData=''))
            request=read()
            while request.get('event')!='request':request=read()
            control('reject',request['id'])
        print('PASS: DLNA consent, decoding, pause, seek, volume, concurrent sessions and reclamation')
        # A separate renderer exercises GUPnP discovery and outgoing SOAP calls.
        target_dir=root/'target';target_dir.mkdir();target_xml=root/'target-upnp';shutil.copytree(xml,target_xml)
        target_udn='uuid:084442d0-ed68-427b-8dc8-ae4b310f6830'
        description=(target_xml/'device.xml').read_text().replace('uuid:56ce7555-6baf-4360-bbd6-61a32b06b6aa',target_udn)
        (target_xml/'device.xml').write_text(description)
        target=subprocess.Popen([str(binaries/'framely-receiver'),str(target_dir),str(target_xml),'1'],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,bufsize=1,env=env)
        target_replies=Queue()
        def target_reader():
            for line in target.stdout:target_replies.put(json.loads(line))
        Thread(target=target_reader,daemon=True).start()
        try:
            deadline=time.monotonic()+12
            while True:
                found=control('devices','')
                if any(d['id']==target_udn for d in found):break
                assert time.monotonic()<deadline,found
                time.sleep(.2)
            control('send',target_udn,uri=uri)
            while True:
                incoming=target_replies.get(timeout=10)
                if incoming.get('event')=='request':break
            assert incoming['uri']==uri,incoming
            assert not (target_dir/(incoming['id']+'.frame')).exists()
            control('send.stop',target_udn)
            while target_replies.get(timeout=10).get('event')!='ended':pass
            print('PASS: DLNA renderer discovery, outgoing SetAVTransportURI/Play and Stop')
        finally:
            target.terminate();target.wait(timeout=3)
            errors=target.stderr.read();assert 'CRITICAL' not in errors and 'WARNING' not in errors,errors

    finally:
        p.terminate();p.wait(timeout=3);http.shutdown();errors=p.stderr.read();assert 'CRITICAL' not in errors and 'WARNING' not in errors,errors