Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Concurrency isolation for multi-broadcaster relay. | |
| No live TikTok. No FFmpeg. Proves A stall/restart does not drop B, | |
| duplicate occupy is rejected, env key is exclusive, and capacity is bounded. | |
| """ | |
| import os, sys, pathlib, threading, tempfile | |
| os.environ.setdefault('PASSWORD', 'test-pass-not-for-prod') | |
| os.environ['MAX_CONCURRENT_RELAYS'] = '2' | |
| os.environ['YOUTUBE_MAX_EVENTS'] = '2' | |
| ROOT = pathlib.Path(__file__).resolve().parent | |
| sys.path.insert(0, str(ROOT)) | |
| from ingest import hub, max_concurrent_relays | |
| passed = failed = 0 | |
| def check(name, cond, detail=''): | |
| global passed, failed | |
| if cond: | |
| passed += 1 | |
| print('PASS %s' % name) | |
| else: | |
| failed += 1 | |
| print('FAIL %s %s' % (name, detail)) | |
| class Flag: | |
| def __init__(self, on=True): | |
| self.on = on | |
| def alive(self): | |
| return self.on | |
| def is_alive(self): | |
| return self.on | |
| def reset(): | |
| hub.slots.clear() | |
| hub.restarts.clear() | |
| reset() | |
| check('max_concurrent_relays honors env', max_concurrent_relays() == 2, str(max_concurrent_relays())) | |
| a_snd, a_th = Flag(), Flag() | |
| b_snd, b_th = Flag(), Flag() | |
| check('occupy A', hub.occupy('alice', 'oauth', sender=a_snd, thread=a_th, dest='oauth')) | |
| check('occupy B independent', hub.occupy('bob', 'oauth', sender=b_snd, thread=b_th, dest='oauth')) | |
| check('duplicate A blocked', not hub.occupy('alice', 'oauth', sender=Flag(), thread=Flag(), dest='oauth')) | |
| check('two ffmpeg counted', hub.active_count() == 2, str(hub.active_count())) | |
| check('at capacity for third', hub.at_capacity(except_user='carol')) | |
| check('not at capacity for restarting A', not hub.at_capacity(except_user='alice')) | |
| # A stalls: ffmpeg + supervisor die. B must keep running. | |
| a_snd.on = False | |
| a_th.on = False | |
| check('A no longer busy after stall', not hub.busy('alice')) | |
| check('B still busy after A stall', hub.busy('bob')) | |
| check('active_count drops only A', hub.active_count() == 1, str(hub.active_count())) | |
| check('A can occupy again', hub.occupy('alice', 'oauth', sender=Flag(), thread=Flag(), dest='oauth')) | |
| check('A restart counted', hub.restart_count('alice') == 1, str(hub.restart_count('alice'))) | |
| check('B still busy after A reconnect', hub.busy('bob')) | |
| check('B restart still 0', hub.restart_count('bob') == 0) | |
| reset() | |
| check('env occupy', hub.occupy('alice', 'direct', sender=Flag(), thread=Flag(), dest='env')) | |
| check('env key busy', hub.env_key_busy()) | |
| check('other user env blocked by helper', hub.env_key_busy(except_user='bob')) | |
| check('same user env not blocked', not hub.env_key_busy(except_user='alice')) | |
| check('oauth dest still allowed', hub.occupy('bob', 'oauth', sender=Flag(), thread=Flag(), dest='oauth')) | |
| reset() | |
| from youtube_manager import ManagedYouTube | |
| logs = [] | |
| yt = ManagedYouTube('ffmpeg', lambda u, force=False: {'status': 'offline', 'room_id': '', 'sources': []}, | |
| lambda u, m: logs.append((u, m)), lambda: ['alice', 'bob'], | |
| pathlib.Path(tempfile.mkdtemp()) / 'yt.json') | |
| src = (ROOT / 'youtube_manager.py').read_text() | |
| tick = src.split('def tick')[1].split('def no_source_wait')[0] | |
| check('tick does not HTTP-probe on discover', '_select_source' not in tick) | |
| check('pushing offline does not exit supervisor', not yt.sender_exit_on_observe('alice', 'offline', True)) | |
| check('live observe does not exit supervisor', not yt.sender_exit_on_observe('alice', 'live', False)) | |
| check('first idle offline stays', not yt.sender_exit_on_observe('bob', 'offline', False)) | |
| check('second idle offline confirmed', yt.sender_exit_on_observe('bob', 'offline', False)) | |
| loop = src.split('def sender_loop')[1].split('def mark_stream_stable')[0] | |
| check('idle confirmed offline does not return from sender_loop', "self.log(user,'[OFFLINE] انتهى بث تيك توك" not in loop and 'continue' in loop) | |
| print('\n%d passed, %d failed' % (passed, failed)) | |
| sys.exit(1 if failed else 0) | |