"""Exercise actual OS locks/processes, without starting inference or touching GPUs."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import os
import subprocess
import sys
import threading
import time
import pytest
from common import write_json
from controller.seed_service import identity
from test_release_integration import ROOT


CHILD='''
import sys,time
from pathlib import Path
sys.path.insert(0,sys.argv[1])
from project_paths import initialize
initialize()
from common import read_json
from controller.service_lease import service_lease
with service_lease(read_json(sys.argv[2]),sys.argv[3]):
    Path(sys.argv[4]).write_text('ready')
    deadline=time.monotonic()+8
    while not Path(sys.argv[5]).exists():
        if time.monotonic()>deadline:raise RuntimeError('test gate timeout')
        time.sleep(.02)
'''


@pytest.mark.parametrize('same_endpoint',[False,True])
def test_student_endpoint_leases_allow_independent_processes_only(tmp_path,config,same_endpoint):
    cfg=tmp_path/'config.json';write_json(cfg,config)
    release=tmp_path/'release';children=[];logs=[]
    def launch(i,endpoint):
        spec=tmp_path/f'lane-{i}.json'
        train_lock=str(tmp_path/'train.lock')
        write_json(spec,{'schema_version':2,'owner':identity(os.getpid()),'seed':i,
            'student_endpoint':endpoint,'training_lock':train_lock})
        env={**os.environ,'RSI_SUITE_LANE_SPEC':str(spec),'DECICO_RSI_API_BASE':endpoint,
            'DECICO_RSI_TRAIN_LOCK':train_lock,'PYTHONDONTWRITEBYTECODE':'1'}
        env.pop('RSI_SERVICE_LOCK_FD',None)
        log=(tmp_path/f'child-{i}.log').open('w');logs.append(log)
        child=subprocess.Popen([sys.executable,'-B','-c',CHILD,str(ROOT/'decico_rsi'),str(cfg),
            str(tmp_path/f'lease-{i}'),str(tmp_path/f'ready-{i}'),str(release)],
            cwd=ROOT,env=env,stdout=log,stderr=subprocess.STDOUT)
        children.append(child)
    def ready(i):
        deadline=time.monotonic()+4
        while not (tmp_path/f'ready-{i}').exists():
            assert children[i].poll() is None,(tmp_path/f'child-{i}.log').read_text()
            assert time.monotonic()<deadline,'service lanes were serialized unexpectedly'
            time.sleep(.02)
    try:
        launch(0,'http://localhost:7501/v1');ready(0)
        launch(1,'http://localhost:7501/v1' if same_endpoint else 'http://localhost:7502/v1')
        if same_endpoint:
            time.sleep(.2);assert not (tmp_path/'ready-1').exists()
        else:ready(1)
        release.write_text('release')
        for child in children:assert child.wait(timeout=5)==0
        assert (tmp_path/'ready-1').exists()
    finally:
        release.write_text('release')
        for child in children:
            if child.poll() is None:child.kill();child.wait()
        for log in logs:log.close()


def test_teacher_shared_reads_and_exclusive_training_never_overlap(monkeypatch,tmp_path):
    from modeling.resource_access import teacher_access
    from modeling.training_resource import training_resource
    monkeypatch.setenv('DECICO_RSI_TRAIN_LOCK',str(tmp_path/'resource.lock'))
    monkeypatch.delenv('DECICO_RSI_TRAIN_RESOURCE_COMMAND',raising=False)
    barrier=threading.Barrier(3);release=threading.Event();training=threading.Event()
    def reader():
        with teacher_access():barrier.wait(timeout=5);assert release.wait(5)
    def trainer():
        with training_resource(tmp_path/'training'):training.set()
    with ThreadPoolExecutor(max_workers=3) as pool:
        reads=[pool.submit(reader) for _ in range(2)]
        barrier.wait(timeout=5)
        write=pool.submit(trainer)
        assert not training.wait(.15)
        release.set()
        for future in reads:future.result(timeout=5)
        write.result(timeout=5)
    assert training.is_set()


def test_stale_suite_owner_cannot_bypass_standalone_queue(monkeypatch,tmp_path):
    from controller.service_lease import resource_scope
    spec=tmp_path/'lane.json'
    write_json(spec,{'schema_version':2,'owner':{'pid':12345},'seed':1,
        'student_endpoint':'http://localhost:5057/v1','training_lock':str(tmp_path/'lock')})
    monkeypatch.setenv('RSI_SUITE_LANE_SPEC',str(spec))
    monkeypatch.setenv('DECICO_RSI_API_BASE','http://localhost:5057/v1')
    monkeypatch.setenv('DECICO_RSI_TRAIN_LOCK',str(tmp_path/'lock'))
    monkeypatch.setattr('controller.seed_service.alive',lambda *a:False)
    with pytest.raises(ValueError,match='stale'):resource_scope()


def test_student_url_aliases_cannot_bypass_endpoint_exclusivity():
    from controller.service_lease import endpoint_key
    assert endpoint_key('http://localhost:5057/v1/')==endpoint_key('http://127.0.0.1:5057/v1')
    assert endpoint_key('https://EXAMPLE.COM:443/v1/')==endpoint_key('https://example.com/v1')
    with pytest.raises(ValueError):endpoint_key('http://user:secret@localhost:5057/v1')