#!/usr/bin/env python3 """Deterministic decoder-record/reconnect tests; no network or transcoding.""" import json import os import pathlib import shutil import subprocess import sys import tempfile ROOT=pathlib.Path(sys.argv[1] if len(sys.argv)>1 else pathlib.Path(__file__).parent).resolve() sys.path.insert(0,str(ROOT)) from flv_bridge import FLVBridge # noqa: E402 from stream_monitor import StreamMonitor, video_dimensions # noqa: E402 passed=failed=0 def check(name,ok,detail=''): global passed,failed if ok: passed+=1;print('PASS ',name) else: failed+=1;print('FAIL ',name,detail) def tag(kind,ts,body): size=len(body) header=bytes([kind])+size.to_bytes(3,'big')+bytes([ (ts>>16)&255,(ts>>8)&255,ts&255,(ts>>24)&255,0,0,0]) return header+body+(11+size).to_bytes(4,'big') def flv(*records): return b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+b''.join(records) def parse(blob): assert blob[:3]==b'FLV' offset=int.from_bytes(blob[5:9],'big')+4;out=[] while offset+15<=len(blob): header=blob[offset:offset+11];size=int.from_bytes(header[1:4],'big') total=11+size+4 assert offset+total<=len(blob) body=blob[offset+11:offset+11+size] assert int.from_bytes(blob[offset+11+size:offset+total],'big')==11+size ts=(header[7]<<24)|(header[4]<<16)|(header[5]<<8)|header[6] out.append((header[0]&31,ts,body));offset+=total assert offset==len(blob) return out def chunked(bridge,data): cuts=(1,2,7,3,19,5,31,11);out=[];pos=0;i=0 while pos0 and not second.startswith(b'FLV') and all(x[0]!=18 for x in joined)) check('source reattach is recorded without claiming RTMP closure', any(e[0]=='RECONNECT' and e[1]['outbound_rtmp_reused'] is True for e in reconnect_events)) check('reattach drops only identical startup decoder record',sum(1 for x in joined if x[0]==9 and x[2][1:2]==b'\x00')==1) check('reattach shifts DTS once while preserving valid equal timestamps', br.last_out_ts>last and all(a[1]<=z[1] for a,z in zip(joined,joined[1:]))) bc=FLVBridge();part1=bc.feed(flv(tag(9,0,seq('avc1',AVC_A)),tag(9,20,coded('avc1',VIDEO_A)))) bc.reconnect();part2=bc.feed(flv(tag(9,0,seq('avc1',AVC_B)),tag(9,20,coded('avc1',VIDEO_A)))) changed=parse(part1+part2) check('reattach never suppresses a changed decoder record', [x[2][5:] for x in changed if x[0]==9 and x[2][1:2]==b'\x00']==[AVC_A,AVC_B]) check('changed reattach remains on one FLV byte stream',not part2.startswith(b'FLV') and bc.format_changes==1) # A sequence header stamped 0 must not consume the splice bias: the first real # frame of a refreshed TikTok FLV can carry a far-away room DTS, and letting a # ts=0 header set the bias makes that frame leap millions of ms into the future # (the output timeline collapses and YouTube discards the leg as noData). bh0=FLVBridge(user='alice',event_logger=lambda *a:None) bh0.feed(flv(tag(9,1000,seq('avc1',AVC_A)),tag(9,1000,coded('avc1',VIDEO_A,frame=1)), tag(9,1040,coded('avc1',VIDEO_H,frame=2)))) bh0.freeze_span_bytes(2000);bh0.reconnect() out_h0=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+bh0.feed(flv( tag(9,0,seq('avc1',AVC_B)),tag(9,500000,coded('avc1',VIDEO_A,frame=1)), tag(9,500040,coded('avc1',VIDEO_H,frame=2))))) check('ts=0 changed video header does not consume splice bias', [t[1] for t in out_h0]==[3080,3120,3160],str([t[1] for t in out_h0])) ba0=FLVBridge(user='alice',event_logger=lambda *a:None) ba0.feed(flv(tag(9,1000,seq('avc1',AVC_A)),tag(9,1000,coded('avc1',VIDEO_A,frame=1)), tag(9,1040,coded('avc1',VIDEO_H,frame=2)))) ba0.freeze_span_bytes(2000);ba0.reconnect() out_a0=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+ba0.feed(flv( tag(8,0,b'\xaf\x00\x12\x10'),tag(9,0,seq('avc1',AVC_A)), tag(9,500000,coded('avc1',VIDEO_A,frame=1))))) check('ts=0 audio header does not consume splice bias', [t[1] for t in out_a0]==[3080,3120],str([t[1] for t in out_a0])) # The byte bridge can represent a FourCC transition without raising or closing. FFmpeg 7 creates # a new late AVStream for such a transition, so production source selection still pins one codec; # same-codec decoder-record replacement above is the fully verified path. switch_events=[] bs=FLVBridge(detect_format_change=True,user='alice', event_logger=lambda user,event,details:switch_events.append(event)) switch=bs.feed(flv(tag(9,0,seq('avc1',AVC_A)),tag(9,1,coded('avc1',VIDEO_A)), tag(9,40,seq('hvc1',HEVC_A,legacy=False)),tag(9,41,coded('hvc1',VIDEO_H,legacy=False)))) switch_tags=parse(switch) check('AVC to HEVC transition holds dependents for a keyframe without raising',len(switch_tags)==3) check('codec transition is separately observable',bs.format_changes==1 and bs.codec_changes==1 and bs.video_codec=='hvc1') check('codec transition writes CODEC_CHANGE without stopping the bridge','CODEC_CHANGE' in switch_events) try: FLVBridge().feed(flv(tag(9,0,b'\x90hvc1'))) truncated=False except ValueError: truncated=True check('truncated Enhanced SequenceStart fails in a controlled way',truncated) # Metric transitions are inferred from tag bytes/timestamps, never by decoding the picture. metric_events=[];records=[tag(9,0,seq('avc1',AVC_A))] for ts in range(0,10000,40): frame=1 if ts in (0,2000,4000,5000,6000,7000,8000,9000) else 2 payload=(b'x'*100) if ts<5000 else (b'y'*1000) records.append(tag(9,ts,coded('avc1',payload,frame=frame))) FLVBridge(user='alice',event_logger=lambda user,event,details:metric_events.append(event)).feed(flv(*records)) check('byte-level monitor detects a sudden bitrate change','BITRATE_CHANGE' in metric_events) check('byte-level monitor detects a keyframe-interval change','KEYFRAME_CHANGE' in metric_events) check('AVC and HEVC SPS dimensions are parsed without decoding', video_dimensions('avc1',AVC_A)==(320,240) and video_dimensions('avc1',AVC_B)==(640,360) and video_dimensions('hvc1',HEVC_A)==(320,240) and video_dimensions('hvc1',HEVC_B)==(640,360)) with tempfile.TemporaryDirectory() as mon_dir: mon_path=pathlib.Path(mon_dir)/'logs'/'stream_events.json';mon=StreamMonitor(mon_path) mon.log('alice','CONNECTION_ERROR',{'scope':'youtube_rtmp','url':'secret-url', 'stream_key':'secret-key','reason':'failed https://secret-host/live/SECRET'}) mon.note_rtmp_open('alice');mon.note_rtmp_open('alice',{'mode':'copy'}) saved=json.loads(mon_path.read_text()) check('stream event file is valid JSON with the required schema', len(saved)==2 and all(set(('timestamp','broadcaster','event','details'))<=set(x) for x in saved)) check('stream monitor never persists URLs, keys or tokens', all(x not in mon_path.read_text() for x in ('secret-url','secret-key','secret-host','SECRET'))) check('stream event file is owner-only',mon_path.stat().st_mode&0o777==0o600) check('second outbound publisher open is recorded as RECONNECT',saved[-1]['event']=='RECONNECT') # FFmpeg 7 is the real outbound middle layer. If available, prove its FLV demux/mux pair writes # replacement AVC and HEVC configuration records instead of fixing extradata at startup. ff=os.environ.get('FFMPEG_PATH') or shutil.which('ffmpeg') if ff and pathlib.Path(ff).is_file(): with tempfile.TemporaryDirectory() as td: td=pathlib.Path(td) fixtures=( ('avc',flv(tag(9,0,seq('avc1',AVC_A)),tag(9,1,coded('avc1',VIDEO_A,frame=1)), tag(9,40,seq('avc1',AVC_B)),tag(9,41,coded('avc1',VIDEO_A,frame=1))),[AVC_A,AVC_B]), ('hevc',flv(tag(9,0,seq('hvc1',HEVC_A,legacy=False)),tag(9,1,coded('hvc1',VIDEO_H,legacy=False,frame=1)), tag(9,40,seq('hvc1',HEVC_B,legacy=False)),tag(9,41,coded('hvc1',VIDEO_H,legacy=False,frame=1))),[HEVC_A,HEVC_B]), ) for name,data,expected in fixtures: src=td/(name+'.flv');dst=td/(name+'-out.flv');src.write_bytes(data) run=subprocess.run([ff,'-y','-hide_banner','-loglevel','error','-i',str(src), '-map','0:v:0','-c:v','copy','-flvflags','no_duration_filesize','-f','flv',str(dst)], stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=20) bodies=[x[2] for x in parse(dst.read_bytes()) if x[0]==9] if run.returncode==0 and dst.exists() else [] configs=[] for body in bodies: if body[0]&0x80 and (body[0]&15)==0:configs.append(body[5:]) elif len(body)>5 and (body[0]&15)==7 and body[1]==0:configs.append(body[5:]) check('FFmpeg copy re-emits both %s decoder records'%name,run.returncode==0 and configs==expected, run.stderr.decode(errors='replace')[-300:]) units=[] for body in bodies: if name=='avc' and len(body)>5 and not(body[0]&0x80) and body[1]==1: units.append(body[5:]) elif name=='hevc' and len(body)>5 and body[0]&0x80: if (body[0]&15)==1 and len(body)>=8:units.append(body[8:]) elif (body[0]&15)==3:units.append(body[5:]) payload=VIDEO_A if name=='avc' else VIDEO_H check('FFmpeg copy preserves %s encoded access units byte-for-byte'%name,units==[payload,payload]) else: print('SKIP FFmpeg decoder-record remux checks (ffmpeg unavailable)') # --- Keyframe gate: splices/config changes resume only on a clean IDR, so the # viewer sees the held picture instead of gray/colour-static error concealment --- gev=[];bg=FLVBridge(user='alice',event_logger=lambda user,event,details:gev.append((event,details))) bg.feed(flv(tag(9,0,seq('avc1',AVC_A)),tag(9,0,coded('avc1',VIDEO_A,frame=1)), tag(9,40,coded('avc1',VIDEO_H,frame=2)))) bg.reconnect() seg2=flv(tag(9,100,seq('avc1',AVC_A)),tag(9,100,coded('avc1',VIDEO_H,frame=2)), tag(8,120,b'\xaf\x01\x11\x22'), tag(9,140,coded('avc1',VIDEO_H,frame=2)),tag(9,180,coded('avc1',VIDEO_A,frame=1)), tag(9,220,coded('avc1',VIDEO_H,frame=2))) out_g=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+bg.feed(seg2)) check('post-reconnect inter frames are held until a keyframe', [t[2] for t in out_g if t[0]==9]==[coded('avc1',VIDEO_A,frame=1),coded('avc1',VIDEO_H,frame=2)], str(out_g)) check('audio flows while the video gate is closed',any(t[0]==8 for t in out_g),str(out_g)) check('keyframe resync is recorded with the dropped count', ('KEYFRAME_RESYNC',{'dropped_tags':2,'capped':False}) in gev,str(gev)) gev2=[];bg2=FLVBridge(user='alice',event_logger=lambda u,e,d:gev2.append((e,d))) bg2.feed(flv(tag(9,0,seq('avc1',AVC_A)),tag(9,0,coded('avc1',VIDEO_A,frame=1)))) out_c=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+bg2.feed( tag(9,40,seq('avc1',AVC_B))+tag(9,40,coded('avc1',VIDEO_H,frame=2))+ tag(9,80,coded('avc1',VIDEO_A,frame=1)))) check('config change holds dependents until the next keyframe', [t[2] for t in out_c if t[0]==9]==[seq('avc1',AVC_B),coded('avc1',VIDEO_A,frame=1)],str(out_c)) # --- Gap filler: frozen keyframes keep the timeline (and YouTube leg) alive --- b2=FLVBridge(user='alice',event_logger=lambda *a:None) check('no filler before any keyframe was emitted',b2.freeze_bytes() is None) src=flv(tag(9,1000,seq('avc1',AVC_A)),tag(9,1000,coded('avc1',VIDEO_A,frame=1)), tag(9,1040,coded('avc1',VIDEO_H,frame=2)),tag(8,1040,b'\xaf\x01\x11\x22')) out2=b2.feed(src) f1=b2.freeze_bytes();f2=b2.freeze_bytes() ftags=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+f1+f2) check('filler repeats the last keyframe body untouched', [t[2] for t in ftags]==[coded('avc1',VIDEO_A,frame=1)]*2, str(ftags)) check('filler timestamps advance monotonically by the step', [t[1] for t in ftags]==[3040,5040], str([t[1] for t in ftags])) # Real source resumes after filling on the same connection: output must continue # from the filler-advanced clock, never backwards into a corrupted timeline. resumed=b2.feed(tag(9,1080,coded('avc1',VIDEO_H,frame=2))) rtags=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+resumed) check('resumed source is spliced after the filler clock', rtags and rtags[0][1]>=5040, str(rtags)) check('filler does not rewrite the resumed frame body', rtags and rtags[0][2]==coded('avc1',VIDEO_H,frame=2)) # --- A/V gap filler: silent AAC spans keep the audio leg alive too --- HE_ASC=bytes.fromhex('131056e598');SIL_STEREO=b'\xaf\x01'+bytes.fromhex('211004608c1c') b3=FLVBridge(user='alice',event_logger=lambda *a:None) b3.feed(flv(tag(9,1000,seq('avc1',AVC_A)),tag(8,1000,b'\xaf\x00'+HE_ASC), tag(9,1000,coded('avc1',VIDEO_A,frame=1)),tag(8,1042,b'\xaf\x01\x54\x54'), tag(9,1080,coded('avc1',VIDEO_H,frame=2)))) check('bridge learns the HE-AAC source config',b3.fill_audio_ready() and b3.audio_he) span1=b3.freeze_span_bytes(2000);span2=b3.freeze_span_bytes(2000) stags=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+span1+span2) aud=[t for t in stags if t[0]==8];vid=[t for t in stags if t[0]==9] check('HE fill first re-declares plain LC at the output rate', aud and aud[0][2]==b'\xaf\x00\x11\x90',str(aud[:1])) frames=[t for t in aud if t[2]==SIL_STEREO] check('silent AAC frames fill both spans at 1024-sample cadence', 184<=len(frames)<=190,str(len(frames))) diffs=[b-a for a,b in zip([t[1] for t in frames],[t[1] for t in frames[1:]])] check('silent frame timestamps step ~21.3ms (48kHz output rate)', diffs and all(d in (21,22) for d in diffs),str(diffs[:6])) check('each fill span still ends on the frozen keyframe', [t[1] for t in vid]==[3080,5080],str([t[1] for t in vid])) back=b3.feed(tag(8,1100,b'\xaf\x01\x54\x54')) btags=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+back) check('first real frame after an LC fill is preceded by the source HE header', len(btags)==2 and btags[0][2]==b'\xaf\x00'+HE_ASC and btags[1][2]==b'\xaf\x01\x54\x54', str(btags)) check('restored header keeps the spliced timeline',btags and btags[0][1]==btags[1][1]>=5080,str(btags)) # Pure-LC source: silent frames ride under the original header, no switching. b4=FLVBridge(user='alice',event_logger=lambda *a:None) b4.feed(flv(tag(9,1000,seq('avc1',AVC_A)),tag(8,1000,b'\xaf\x00\x12\x10'), tag(9,1000,coded('avc1',VIDEO_A,frame=1)),tag(8,1023,b'\xaf\x01\x54\x54'), tag(9,1040,coded('avc1',VIDEO_H,frame=2)))) span=b4.freeze_span_bytes(2000) ltags=parse(b'FLV\x01\x05\x00\x00\x00\x09\x00\x00\x00\x00'+span) laud=[t for t in ltags if t[0]==8] check('LC fill emits no replacement header', laud and all(t[2]==SIL_STEREO for t in laud),str(laud[:2])) ldiffs=[b-a for a,b in zip([t[1] for t in laud],[t[1] for t in laud[1:]])] check('LC silent cadence is ~23.2ms (44.1kHz)',ldiffs and all(d in (23,24) for d in ldiffs),str(ldiffs[:6])) print('\n%d passed, %d failed'%(passed,failed)) sys.exit(1 if failed else 0)