65 lines
9.4 KiB
PHP
65 lines
9.4 KiB
PHP
<?php
|
|
require_once __DIR__ . '/../web/server/worker.php';
|
|
$originalCwd=getcwd();$tmp=sys_get_temp_dir().'/exoplasim-worker-'.bin2hex(random_bytes(5));mkdir($tmp,0700,true);
|
|
$storage=['root'=>$tmp,'database'=>'data/lab.sqlite3','jobs'=>'jobs','runtime'=>'runtime','catalog'=>'catalog/schema.json'];
|
|
foreach(['data','jobs','runtime','catalog'] as $dir) mkdir($tmp.'/'.$dir,0750,true);
|
|
$schema=['fields'=>[],'model'=>[],'run'=>[],'model_defaults'=>[],'run_defaults'=>[]];file_put_contents($tmp.'/catalog/schema.json',json_encode($schema));
|
|
$science=$tmp.'/science';mkdir($science,0750,true);
|
|
$script=<<<'PY'
|
|
import argparse,json,os,pathlib
|
|
p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--job');a=p.parse_args();j=json.load(open(a.job));print('ENV_MARKER='+os.environ.get('EXOPLASIM_TEST_INHERIT','missing'),flush=True);d=('e' if j['kind']=='planet' else 'f')*32;root=pathlib.Path(j['directory'])/'artifacts'/'datasets'/d;root.mkdir(parents=True,exist_ok=True);json.dump({'id':d,'fields':[]},open(root/'index.json','w'));event={'cycle':1,'cycles':1,'dataset':d};event.update({'planet':True,'spec':j['spec']} if j['kind']=='planet' else {});print(json.dumps(event),flush=True)
|
|
PY;
|
|
file_put_contents($science.'/model.py',$script);file_put_contents($science.'/planet.py',$script);
|
|
$config=['workspace'=>'local','storage'=>$storage,'python'=>'/usr/bin/python3','science_scripts'=>$science,'project_root'=>dirname(__DIR__),'poll_seconds'=>1,'job_timeout_seconds'=>60,'minimum_free_bytes'=>0,'queue_limit'=>10];
|
|
file_put_contents($tmp.'/config.json',json_encode($config));putenv('EXOPLASIM_CONFIG='.$tmp.'/config.json');
|
|
function expect($condition,$message){if(!$condition)throw new RuntimeException($message);}
|
|
try {
|
|
$worker=new LabWorker();
|
|
expect($worker->acquire(),'First worker must acquire flock.');
|
|
$second=new LabWorker();expect(!$second->acquire(),'Second worker must not acquire flock.');
|
|
$db=LabDb::connection();$now=gmdate('c');
|
|
$db->exec("INSERT INTO jobs(id,owner,name,spec,status,total_cycles,created_at,updated_at) VALUES('".str_repeat('a',32)."','local','bad','{\"invalid\":true}','queued',1,'$now','$now')");
|
|
$installation=fopen(lab_config()['runtime_lock'],'c');flock($installation,LOCK_EX);
|
|
expect($worker->claim()===null,'Installation lock must prevent a worker claim.');
|
|
expect($db->query("SELECT status FROM jobs WHERE id='".str_repeat('a',32)."'")->fetchColumn()==='queued','Installation must leave the queued job intact.');
|
|
fclose($installation);
|
|
$claimed=$worker->claim();expect($claimed['id']===str_repeat('a',32),'Worker must claim queued job after installation lock is released.');
|
|
$worker->recover();$status=$db->query("SELECT status FROM jobs WHERE id='".str_repeat('a',32)."'")->fetchColumn();expect($status==='interrupted','Recovery must interrupt running job.');
|
|
$db->exec("UPDATE jobs SET status='running' WHERE id='".str_repeat('a',32)."'");
|
|
$reflection=new ReflectionClass(LabWorker::class);$method=$reflection->getMethod('run');$method->setAccessible(true);
|
|
$method->invoke($worker,['id'=>str_repeat('a',32),'owner'=>'local','name'=>'bad','kind'=>'model','spec'=>['invalid'=>true],'total_cycles'=>1]);
|
|
$row=$db->query("SELECT status,error FROM jobs WHERE id='".str_repeat('a',32)."'")->fetch();expect($row['status']==='failed'&&str_contains($row['error'],'validation'),'Invalid stored spec must fail before task launch.');
|
|
putenv('EXOPLASIM_TEST_INHERIT=inherited-value');chdir($tmp);putenv('EXOPLASIM_CONFIG=config.json');$id=str_repeat('b',32);$spec=json_encode(['model'=>[],'run'=>[],'parameters'=>[]]);$s=$db->prepare("INSERT INTO jobs(id,owner,name,kind,spec,status,total_cycles,created_at,updated_at) VALUES(?,?,?,'model',?,'running',1,?,?)");$s->execute([$id,'local','test',$spec,$now,$now]);
|
|
$method->invoke($worker,['id'=>$id,'owner'=>'local','name'=>'test','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);
|
|
$complete=$db->prepare('SELECT status,progress,completed_cycles FROM jobs WHERE id=?');$complete->execute([$id]);$result=$complete->fetch();$complete->closeCursor();expect($result['status']==='completed'&&(float)$result['progress']===100.0&&(int)$result['completed_cycles']===1,'Worker must drain final event and complete job.');
|
|
$published=json_decode(file_get_contents($tmp.'/jobs/'.$id.'/artifacts/index.json'),true);expect(($published['latest']??null)===str_repeat('f',32),'PHP worker must publish the job results index.');
|
|
expect(str_contains(file_get_contents($tmp.'/jobs/'.$id.'/worker.log'),'ENV_MARKER=inherited-value'),'Worker must preserve inherited process environment.');
|
|
$planetId=str_repeat('e',32);$planetSpec=json_encode(['model'=>[],'run'=>[],'parameters'=>[],'surface'=>[]]);$p=$db->prepare("INSERT INTO jobs(id,owner,name,kind,spec,status,total_cycles,created_at,updated_at) VALUES(?,?,?,'planet',?,'running',1,?,?)");$p->execute([$planetId,'local','generated',$planetSpec,$now,$now]);
|
|
$method->invoke($worker,['id'=>$planetId,'owner'=>'local','name'=>'generated','kind'=>'planet','spec'=>json_decode($planetSpec,true),'total_cycles'=>1]);
|
|
$savedQuery=$db->query("SELECT experiment_id,status FROM jobs WHERE id='$planetId'");$saved=$savedQuery->fetch();$savedQuery->closeCursor();expect($saved['status']==='completed'&&preg_match('/^[a-f0-9]{32}$/',$saved['experiment_id']??''),'Planet job must save an experiment and link it to the job.');
|
|
$planetIndex=json_decode(file_get_contents($tmp.'/jobs/'.$planetId.'/artifacts/index.json'),true);expect(($planetIndex['latest']??null)===str_repeat('e',32),'PHP worker must publish planet result index.');expect(count($planetIndex['datasets'])===1,'Planet event must publish its result index once.');
|
|
$noOutputId=str_repeat('9',32);$s->execute([$noOutputId,'local','silent',$spec,$now,$now]);
|
|
file_put_contents($science.'/model.py',implode(PHP_EOL,["import argparse","p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--job');p.parse_args()"]));
|
|
$method->invoke($worker,['id'=>$noOutputId,'owner'=>'local','name'=>'silent','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);$silent=$db->query("SELECT status,error FROM jobs WHERE id='$noOutputId'")->fetch();expect($silent['status']==='failed'&&str_contains($silent['error'],'without publishing'),'Zero exit without results must fail.');
|
|
$badKindId=str_repeat('8',32);$s->execute([$badKindId,'local','kind',$spec,$now,$now]);$method->invoke($worker,['id'=>$badKindId,'owner'=>'local','name'=>'kind','kind'=>'other','spec'=>json_decode($spec,true),'total_cycles'=>1]);expect($db->query("SELECT status FROM jobs WHERE id='$badKindId'")->fetchColumn()==='failed','Unsupported job kind must fail.');
|
|
$badJsonId=str_repeat('7',32);$db->exec("INSERT INTO jobs(id,owner,name,spec,status,total_cycles,created_at,updated_at) VALUES('$badJsonId','local','json','{','queued',1,'$now','$now')");$badJson=$worker->claim();expect(($badJson['spec_error']??'')!=='','Malformed stored JSON must not terminate claim.');$method->invoke($worker,$badJson);expect($db->query("SELECT status FROM jobs WHERE id='$badJsonId'")->fetchColumn()==='failed','Malformed stored JSON must fail its job.');
|
|
file_put_contents($tmp.'/large.log',str_repeat('x',1048576));$tailMethod=(new ReflectionClass(LabWorker::class))->getMethod('tail');$tailMethod->setAccessible(true);$tail=$tailMethod->invoke($worker,$tmp.'/large.log');expect(strlen($tail)===8192&&$tail===str_repeat('x',8192),'Log tail must read only its bounded suffix.');
|
|
$sleepScript=<<<'PY'
|
|
import argparse,time
|
|
p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--job');p.parse_args();time.sleep(20)
|
|
PY;
|
|
file_put_contents($science.'/model.py',$sleepScript);
|
|
$timeoutId=str_repeat('c',32);$s->execute([$timeoutId,'local','timeout',$spec,$now,$now]);
|
|
$property=(new ReflectionClass(LabWorker::class))->getProperty('config');$workerConfig=$property->getValue($worker);$workerConfig['job_timeout_seconds']=1;$property->setValue($worker,$workerConfig);
|
|
$begin=microtime(true);$method->invoke($worker,['id'=>$timeoutId,'owner'=>'local','name'=>'timeout','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);$timedQuery=$db->query("SELECT status,error FROM jobs WHERE id='$timeoutId'");$timed=$timedQuery->fetch();$timedQuery->closeCursor();expect($timed['status']==='failed'&&str_contains($timed['error'],'time limit')&µtime(true)-$begin<8,'Worker must enforce timeout and stop the process group.');
|
|
file_put_contents($science.'/model.py',$sleepScript);$cancelId=str_repeat('d',32);$s->execute([$cancelId,'local','cancel',$spec,$now,$now]);
|
|
$pid=pcntl_fork();if($pid===0){usleep(500000);$cancelDb=new PDO('sqlite:'.$tmp.'/data/lab.sqlite3');$cancelDb->exec("UPDATE jobs SET cancel_requested=1 WHERE id='$cancelId'");exit(0);}
|
|
$begin=microtime(true);$method->invoke($worker,['id'=>$cancelId,'owner'=>'local','name'=>'cancel','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);pcntl_waitpid($pid,$childStatus);$cancelled=$db->query("SELECT status FROM jobs WHERE id='$cancelId'")->fetchColumn();expect($cancelled==='cancelled'&µtime(true)-$begin<8,'Worker must honor cancellation and stop the process group.');
|
|
echo "PHP worker tests passed\n";
|
|
} finally {
|
|
putenv('EXOPLASIM_CONFIG');chdir($originalCwd);
|
|
$it=new RecursiveIteratorIterator(new RecursiveDirectoryIterator($tmp,FilesystemIterator::SKIP_DOTS),RecursiveIteratorIterator::CHILD_FIRST);
|
|
foreach($it as $file){$file->isDir()?rmdir($file->getPathname()):unlink($file->getPathname());}
|
|
rmdir($tmp);
|
|
}
|