#!/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)