| """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') |
|
|
