Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Keep media flowing while YouTube completes liveStarting/testStarting. | |
| No cloud calls or real ingest URLs. | |
| """ | |
| import os, pathlib, sys, tempfile, time, unittest | |
| from unittest.mock import Mock | |
| ROOT=pathlib.Path(sys.argv[1]).resolve() if len(sys.argv)>1 else pathlib.Path(__file__).parent | |
| sys.path.insert(0,str(ROOT)) | |
| from youtube_manager import ManagedYouTube | |
| class TransitionTests(unittest.TestCase): | |
| def setup_event(self,life,poll=True): | |
| tmp=tempfile.TemporaryDirectory();self.addCleanup(tmp.cleanup) | |
| m=ManagedYouTube('/bin/true',lambda u:None,lambda *a:None,lambda:['alice'], | |
| pathlib.Path(tmp.name)/'journal.json') | |
| key='alice:room1' | |
| event={'username':'alice','room_id':'room1','session_key':key,'marker':'OWNED-MARKER', | |
| 'bound':True,'broadcast_id':'OWNED-VIDEO','youtube_status':life,'phase':'starting', | |
| 'was_live':False,'closing':False,'created_at':time.time(), | |
| 'last_poll':0 if poll else time.time(),'note':''} | |
| m.journal.change(lambda d:d.update(channel={'id':'CHANNEL'},streams={'alice':{'id':'OWNED-STREAM'}},events={key:event})) | |
| source={'url':'https://example.invalid/source.flv','media':'flv','codec':'h264','label':'fixture'} | |
| m._observe=Mock(return_value={'status':'live','room_id':'room1','sources':[source]}) | |
| m.video_sources=lambda sources:list(sources) | |
| m.destinations['alice']='rtmps://example.invalid/test-only' | |
| m.remote=Mock(return_value={'id':'OWNED-VIDEO','status':{'lifeCycleStatus':life}, | |
| 'snippet':{'description':'OWNED-MARKER'},'contentDetails':{'boundStreamId':'OWNED-STREAM'}}) | |
| m.start_sender=Mock();m.stop_sender=Mock();m.finish_event=Mock();m.stream_health=Mock() | |
| m.api.call=Mock(side_effect=AssertionError('Transitional observation must not create or re-transition the event')) | |
| return m,key | |
| def check_transition(self,life,poll): | |
| m,key=self.setup_event(life,poll) | |
| m.drive_event(key) | |
| m.start_sender.assert_called_once_with(key) | |
| m.stop_sender.assert_not_called() | |
| m.finish_event.assert_not_called() | |
| m.api.call.assert_not_called() | |
| self.assertFalse(m.journal.read()['events'][key]['was_live']) | |
| self.assertEqual(m.journal.read()['events'][key]['broadcast_id'],'OWNED-VIDEO') | |
| def test_live_starting_fresh_poll_keeps_publisher(self):self.check_transition('liveStarting',True) | |
| def test_live_starting_cached_status_keeps_publisher(self):self.check_transition('liveStarting',False) | |
| def test_test_starting_fresh_poll_keeps_publisher(self):self.check_transition('testStarting',True) | |
| def test_test_starting_cached_status_keeps_publisher(self):self.check_transition('testStarting',False) | |
| def test_accepted_transition_is_not_cancelled_by_prestart_timeout(self): | |
| m,key=self.setup_event('liveStarting') | |
| m.journal.change(lambda d:d['events'][key].update(created_at=time.time()-3600)) | |
| m.drive_event(key) | |
| m.start_sender.assert_called_once_with(key) | |
| m.finish_event.assert_not_called() | |
| m.api.call.assert_not_called() | |
| def test_terminal_states_never_restart_publisher(self): | |
| for life in ('complete','revoked'): | |
| m,key=self.setup_event(life) | |
| m.drive_event(key) | |
| m.start_sender.assert_not_called() | |
| m.finish_event.assert_called_once_with(key) | |
| def test_created_state_is_still_gated(self): | |
| m,key=self.setup_event('created') | |
| m.drive_event(key) | |
| m.start_sender.assert_not_called() | |
| m.stop_sender.assert_called_once_with(key) | |
| if __name__=='__main__':unittest.main(argv=[sys.argv[0]],verbosity=2) | |