"""Disposable Git repositories only. Never imports or calls application services."""
import copy
import json
import os
import shutil
from pathlib import Path
import subprocess
import sys
import tempfile
import time
import unittest
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import controller as c
class QueueTests(unittest.TestCase):
def test_real_queue_is_ordered_and_missing_sources_are_explicit(self):
root = Path(__file__).resolve().parents[1]
manifest = json.loads((root / 'backlog.json').read_text(encoding='utf-8'))
c.validate_manifest(manifest, root)
self.assertEqual(list(range(5, 12)), [s['sequence'] for s in manifest['stages']])
self.assertEqual('PHASE-10', manifest['stages'][0]['id'])
self.assertFalse(manifest['executionEnabled'])
def test_production_branches_rejected(self):
m = fixture_manifest()
m['stages'][0]['branch'] = 'main'
with self.assertRaises(c.GateError):
c.validate_manifest(m)
def test_forward_dependency_rejected(self):
m = fixture_manifest()
m['stages'][0]['dependencies'] = ['TWO']
with self.assertRaises(c.GateError):
c.validate_manifest(m)
def fixture_manifest():
def stage(name, sequence, deps):
return dict(id=name, sequence=sequence, sourcePrompt='prompts/task.txt',
promptSha256='', sourceStatus='VERIFIED', dependencies=deps,
branch='codex/' + name.lower(), requiredOwnerApprovals=['entry'],
retryPolicy=dict(maxAttempts=2, maxSessions=4, backoffSeconds=0),
status='PENDING', evidence=[], reports=[], contextFiles=[],
checkpoints=[dict(id='work', instruction='Synthetic task only',
approvalsBefore=[], approvalsAfter=[], deferrableGates=[],
requiredArtifacts=['result.txt'], checks=[dict(id='self-check',
argv=[sys.executable, '-c', 'print("verified")'], cwd='.',
report=None)], evidenceRequirements=['synthetic-only'])])
return dict(schemaVersion=1, executionEnabled=True, integrationBranch='codex/integration',
minimumCodexVersion='0.163.0', remote='origin', baselineCommit='0'*40,
baselineBranch='codex/base', stages=[stage('ONE', 5, []), stage('TWO', 6, ['ONE'])])
class SyntheticRunner:
def __init__(self, result='complete', fail_checks=False):
self.result = result
self.fail_checks = fail_checks
self.calls = 0
def run_session(self, engine, stage, checkpoint, directory, prompt):
self.calls += 1
(engine.repo / 'result.txt').write_text(f'{stage["id"]} {self.calls}', encoding='utf-8')
if self.result == 'interrupt':
raise c.Interrupted('simulated CLI termination')
if self.result == 'fail':
raise c.GateError('synthetic failure')
return dict(outcome=self.result, summary='synthetic', blockers=[],
requestedApprovals=[], checkpoint=checkpoint['id'])
def run_check(self, engine, check, directory):
if self.fail_checks:
raise c.GateError('synthetic test exit 1')
return {'id': check['id'], 'exitCode': 0, 'report': None, 'passed': None, 'failed': 0}
class EngineTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory(prefix='jobjakt-orchestrator-test-')
self.root = Path(self.temp.name)
self.repo = self.root / 'worker'
self.repo.mkdir()
self.remote = self.root / 'remote.git'
self.git_at(self.root, 'init', '--bare', str(self.remote))
self.git_at(self.repo, 'init', '-b', 'codex/base')
self.git_at(self.repo, 'config', 'user.name', 'Synthetic Test')
self.git_at(self.repo, 'config', 'user.email', 'synthetic@invalid.example')
(self.repo / 'README').write_text('synthetic only')
self.git_at(self.repo, 'add', '.')
self.git_at(self.repo, 'commit', '-m', 'baseline')
self.git_at(self.repo, 'remote', 'add', 'origin', str(self.remote))
self.git_at(self.repo, 'push', 'origin', 'HEAD:refs/heads/codex/base')
self.control = self.root / 'control'
(self.control / 'prompts').mkdir(parents=True)
(self.control / 'prompts/task.txt').write_text('Complete synthetic task. No real backlog.')
self.manifest = fixture_manifest()
self.manifest['baselineCommit'] = self.git_at(self.repo, 'rev-parse', 'HEAD')
for s in self.manifest['stages']:
s['promptSha256'] = c.sha_file(self.control / s['sourcePrompt'])
self.config = dict(repository=str(self.repo), codexExecutable='codex',
powershellExecutable='pwsh', publicKeyCertificate='owner.cer',
ownerSid='OWNER', runnerSid='RUNNER', executionWindows=[],
maxSessionSeconds=15, maxCheckSeconds=15, maxRunSeconds=60,
maxLogBytes=1048576, workerCodeXHome=str(self.root / 'codex-home'))
self.runner = SyntheticRunner()
self.engine = c.Engine(self.control, self.manifest, self.config, runner=self.runner,
approval_verifier=lambda request: True, host_verifier=lambda: None)
self.engine.initialize()
def tearDown(self):
self.temp.cleanup()
@staticmethod
def git_at(directory, *args):
p = subprocess.run(['git', '-c', 'core.hooksPath=', '-C', str(directory), *args],
text=True, capture_output=True, encoding='utf-8')
if p.returncode:
raise AssertionError(p.stderr)
return p.stdout.strip()
def run_engine(self):
self.engine.run()
return self.engine.store.read()
def test_success_orders_stages_and_verifies_remote(self):
state = self.run_engine()
self.assertEqual(['COMPLETED', 'COMPLETED'], [s['status'] for s in state['stages'].values()])
self.assertEqual(2, self.runner.calls)
remote = self.git_at(self.repo, 'ls-remote', 'origin', 'refs/heads/codex/two').split()[0]
self.assertEqual(remote, state['integrationCommit'])
def test_missing_approval_does_not_run_model(self):
self.engine.approval_verifier = lambda request: False
state = self.run_engine()
self.assertEqual('AWAITING_APPROVAL', state['stages']['ONE']['status'])
self.assertEqual(0, self.runner.calls)
self.assertTrue(list((self.control / 'requests').glob('*.json')))
def test_dependency_cannot_be_bypassed(self):
state = self.engine.store.read()
state['stages']['ONE']['status'] = 'CONDITIONALLY_ACCEPTED'
self.engine.store.write(state)
self.engine.approval_verifier = lambda request: False
self.assertEqual(0, self.runner.calls)
self.assertNotEqual('COMPLETED', self.run_engine()['stages']['TWO']['status'])
def test_failed_stage_stops_and_exhausts_retry_budget(self):
self.runner.result = 'fail'
state = self.run_engine()
self.assertEqual('FAILED', state['stages']['ONE']['status'])
self.assertEqual(2, self.runner.calls)
self.engine.approval_verifier = lambda request: request['kind'] != 'retry-budget-reset'
self.engine.retry('ONE')
self.run_engine()
self.assertEqual(2, self.runner.calls)
def test_interruption_is_not_success_and_resume_has_budget(self):
self.runner.result = 'interrupt'
self.assertNotEqual('COMPLETED', self.run_engine()['stages']['ONE']['status'])
self.runner.result = 'complete'
state = self.run_engine()
self.assertEqual('COMPLETED', state['stages']['ONE']['status'])
def test_missing_prompt_blocks(self):
(self.control / 'prompts/task.txt').unlink()
self.assertEqual('BLOCKED', self.run_engine()['stages']['ONE']['status'])
self.assertEqual(0, self.runner.calls)
def test_changed_prompt_blocks(self):
(self.control / 'prompts/task.txt').write_text('changed')
self.assertEqual('BLOCKED', self.run_engine()['stages']['ONE']['status'])
def test_unavailable_remote_preserves_commit_and_retries_publication_only(self):
self.git_at(self.repo, 'remote', 'set-url', 'origin', str(self.root / 'absent.git'))
state = self.run_engine()
self.assertEqual(0, self.runner.calls) # baseline publication must be verified first
self.assertEqual('BLOCKED', state['stages']['ONE']['status'])
def test_push_failure_preserves_tested_commit(self):
original = self.engine.publish
self.engine.publish = lambda *args: (_ for _ in ()).throw(c.PublicationError('push unavailable'))
state = self.run_engine()
commit = state['stages']['ONE']['pendingPublication']
self.assertTrue(commit)
self.assertEqual(1, self.runner.calls)
self.engine.publish = original
state = self.run_engine()
self.assertEqual('COMPLETED', state['stages']['ONE']['status'])
self.assertEqual(2, self.runner.calls) # only stage TWO ran after retry
def test_push_verification_mismatch_stops(self):
self.engine.verify_remote = lambda branch, commit: (_ for _ in ()).throw(c.PublicationError('mismatch'))
self.assertEqual('BLOCKED', self.run_engine()['stages']['ONE']['status'])
self.assertEqual(0, self.runner.calls)
def test_conditional_requires_explicit_deferrable_gates(self):
self.runner.result = 'conditional'
state = self.run_engine()
self.assertEqual('CONDITIONALLY_ACCEPTED', state['stages']['ONE']['status'])
self.assertEqual(1, self.runner.calls)
def test_pause_persists_and_resume_clears_it(self):
self.engine.pause()
self.assertEqual(0, self.runner.calls)
self.run_engine()
self.assertEqual(0, self.runner.calls)
self.engine.resume()
self.run_engine()
self.assertEqual(2, self.runner.calls)
def test_duplicate_lock_rejected(self):
with c.RepoLock(self.repo):
with self.assertRaises(c.GateError):
with c.RepoLock(self.repo):
pass
def test_corrupt_state_recovers_last_good_without_success(self):
self.engine.pause()
self.engine.store.path.write_text('{corrupt')
state = self.engine.store.read()
self.assertTrue(state['recoveryRequired'])
self.assertEqual('PENDING', state['stages']['ONE']['status'])
def test_double_corruption_fails_closed(self):
self.engine.store.path.write_text('bad')
self.engine.store.backup.write_text('bad')
with self.assertRaises(c.GateError):
self.engine.store.read()
def test_accidental_main_checkout_rejected(self):
self.git_at(self.repo, 'switch', '-c', 'main')
state = self.run_engine()
self.assertEqual('BLOCKED', state['stages']['ONE']['status'])
self.assertEqual(0, self.runner.calls)
def test_failing_checks_never_commit_or_advance(self):
self.runner.fail_checks = True
self.assertEqual('FAILED', self.run_engine()['stages']['ONE']['status'])
self.assertEqual(self.manifest['baselineCommit'], self.git_at(self.repo, 'rev-parse', 'HEAD'))
def test_model_cannot_approve_itself(self):
self.engine.approval_verifier = lambda request: False
(self.repo / 'approval.json').write_text('{"approved":true}')
self.assertEqual('AWAITING_APPROVAL', self.run_engine()['stages']['ONE']['status'])
self.assertEqual(0, self.runner.calls)
def test_disabled_manifest_never_executes(self):
self.engine.manifest['executionEnabled'] = False
with self.assertRaises(c.GateError):
self.run_engine() # mutating policy without signed adoption is rejected
self.assertEqual(0, self.runner.calls)
def test_manifest_initially_disabled(self):
state = self.engine.store.read()
self.engine.manifest['executionEnabled'] = False
state['manifestSha256'] = c.digest(self.engine.manifest)
self.engine.store.write(state)
self.run_engine()
self.assertEqual(0, self.runner.calls)
def test_after_approval_binds_tested_commit_without_rerunning_worker(self):
self.manifest['stages'][0]['checkpoints'][0]['approvalsAfter'] = ['design-selection']
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
self.engine.store.write(state)
self.engine.approval_verifier = lambda r: r['kind'] != 'design-selection'
state = self.run_engine()
self.assertEqual('AWAITING_APPROVAL', state['stages']['ONE']['status'])
self.assertTrue(state['stages']['ONE']['checkpointCommit'])
self.assertEqual(1, self.runner.calls)
self.engine.approval_verifier = lambda r: {'decision': 'Use concept B, cyan buttons from D'}
self.assertEqual('COMPLETED', self.run_engine()['stages']['TWO']['status'])
self.assertEqual(2, self.runner.calls)
def test_tampered_evidence_cannot_be_approved(self):
self.manifest['stages'][0]['checkpoints'][0]['approvalsAfter'] = ['review']
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
self.engine.store.write(state)
self.engine.approval_verifier = lambda r: r['kind'] != 'review'
self.run_engine()
(self.repo / 'result.txt').write_text('tampered')
self.engine.approval_verifier = lambda r: True
self.assertEqual('BLOCKED', self.run_engine()['stages']['ONE']['status'])
def test_restart_running_marker_cannot_mean_completed(self):
state = self.engine.store.read()
state['stages']['ONE']['status'] = 'RUNNING'
self.engine.store.write(state)
c.atomic(self.control / 'active-process.json', {'pid': 999999})
with self.assertRaises(c.GateError):
self.run_engine()
self.assertEqual(0, self.runner.calls)
def test_same_repo_other_control_root_cannot_take_lock(self):
with c.RepoLock(self.repo):
other = c.Engine(self.root / 'other-control', self.manifest, self.config)
with self.assertRaises(c.GateError):
with c.RepoLock(other.repo):
pass
def test_execution_window_stops_before_worker(self):
self.engine.in_window = lambda: False
self.run_engine()
self.assertEqual(0, self.runner.calls)
def test_conditional_can_advance_only_with_explicit_signed_deferral(self):
for stage in self.manifest['stages']:
stage['checkpoints'][0]['deferrableGates'] = ['production-human-review']
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
self.engine.store.write(state)
self.runner.result = 'conditional'
self.engine.approval_verifier = lambda r: r['kind'] != 'conditional-deferral'
self.assertEqual('AWAITING_APPROVAL', self.run_engine()['stages']['ONE']['status'])
self.engine.approval_verifier = lambda r: True
self.assertEqual('COMPLETED', self.run_engine()['stages']['TWO']['status'])
def test_unconfigured_test_adapter_blocks_before_model(self):
self.manifest['stages'][0]['checkpoints'][0]['checks'][0]['configured'] = False
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
self.engine.store.write(state)
self.assertEqual('BLOCKED', self.run_engine()['stages']['ONE']['status'])
self.assertEqual(0, self.runner.calls)
def test_design_revision_reruns_current_checkpoint_and_keeps_owner_selection(self):
self.manifest['stages'][0]['checkpoints'][0]['approvalsAfter'] = ['homepage-design-selection']
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
self.engine.store.write(state)
self.engine.approval_verifier = lambda r: ({'decision': 'Revise A using buttons from D', 'disposition': 'Revise'}
if r['kind'] == 'homepage-design-selection' else True)
self.assertEqual('PENDING', self.run_engine()['stages']['ONE']['status'])
self.assertEqual(1, self.runner.calls)
self.engine.approval_verifier = lambda r: {'decision': 'Approve revised concept A', 'disposition': 'Approve'}
state = self.run_engine()
self.assertEqual('COMPLETED', state['stages']['TWO']['status'])
self.assertEqual(3, self.runner.calls)
self.assertTrue(any('buttons from D' in d['decision'] for d in state['ownerDecisions']))
def test_recovered_running_without_marker_requires_reconciliation(self):
state = self.engine.store.read()
state['stages']['ONE']['status'] = 'RUNNING'
self.engine.store.write(state)
with self.assertRaisesRegex(c.GateError, 'RUNNING'):
self.run_engine()
self.assertEqual(0, self.runner.calls)
def test_unsigned_recovery_cannot_clear_running_guard(self):
state = self.engine.store.read()
state['stages']['ONE']['status'] = 'RUNNING'
self.engine.store.write(state)
c.atomic(self.control / 'config.json', self.config)
c.atomic(self.control / 'backlog.json', self.manifest)
self.engine.approval_verifier = lambda r: False
with mock.patch.object(c, 'Engine', return_value=self.engine):
self.assertEqual(3, c.main(['recover','--control-root',str(self.control)]))
with self.assertRaises(c.GateError):
self.engine.resume()
self.run_engine()
self.assertEqual(0, self.runner.calls)
def test_cryptographically_approved_resume_e2e(self):
pwsh = shutil.which('pwsh')
self.assertIsNotNone(pwsh, 'PowerShell is a required orchestrator test prerequisite')
signer = Path(__file__).with_name('Synthetic-Signer.ps1')
verifier = Path(c.__file__).parent.parent / 'scripts/Approve-JobjaktBacklog.ps1'
shutil.copyfile(verifier,self.control / 'Approve-JobjaktBacklog.ps1')
subprocess.run([pwsh,'-NoProfile','-File',str(signer),'-Action','Setup','-Root',str(self.control)],
check=True,capture_output=True)
self.engine.config['powershellExecutable'] = pwsh
self.engine.approval_verifier = self.engine.verify_approval
for _ in range(8):
state = self.run_engine()
if all(ss['status'] == 'COMPLETED' for ss in state['stages'].values()):
break
pending = next(ss for ss in state['stages'].values() if ss['awaiting'])
request = Path(pending['awaiting'][0]['request'])
if state['stages']['ONE']['entryBaseline'] is None:
self.assertEqual(0,self.runner.calls)
subprocess.run([pwsh,'-NoProfile','-File',str(signer),'-Action','Sign','-Root',str(self.control),
'-RequestPath',str(request),'-ApprovalPath',str(self.control / 'approvals' / request.name)],
check=True,capture_output=True)
self.assertTrue(all(ss['status'] == 'COMPLETED' for ss in state['stages'].values()))
self.assertEqual(2,self.runner.calls)
class ReportTests(unittest.TestCase):
def test_report_membership_and_skip_enforcement(self):
with tempfile.TemporaryDirectory() as td:
path = Path(td) / 'report.xml'
path.write_text('')
spec = dict(format='junit', expectedNames=['A.B'])
self.assertEqual(1, c.parse_report(path, spec)['passed'])
spec['expectedNames'] = ['A.Missing']
with self.assertRaises(c.GateError):
c.parse_report(path, spec)
path.write_text('')
with self.assertRaises(c.GateError):
c.parse_report(path, dict(format='junit'))
def test_empty_report_is_not_a_pass(self):
with tempfile.TemporaryDirectory() as td:
path = Path(td) / 'report.xml'
path.write_text('')
with self.assertRaises(c.GateError):
c.parse_report(path, dict(format='junit'))
def test_playwright_retry_and_expected_failure_are_not_pass(self):
with tempfile.TemporaryDirectory() as td:
path = Path(td) / 'report.json'
for case in [dict(projectName='x',results=[dict(status='failed'),dict(status='passed')]),
dict(projectName='x',expectedStatus='failed',results=[dict(status='failed')])]:
c.atomic(path, dict(suites=[dict(specs=[dict(title='a',tests=[case])])]))
with self.assertRaises(c.GateError):
c.parse_report(path, dict(format='playwright'))
def test_path_escape_rejected(self):
with tempfile.TemporaryDirectory() as td:
with self.assertRaises(c.GateError):
c.under(td, '../outside')
def test_redacts_common_credentials(self):
result = c.redact('password=secret Bearer abc https://user:pass@example.invalid token=hello')
for secret in ['secret', 'abc', 'user:pass', 'hello']:
self.assertNotIn(secret, result)
def test_json_log_redaction_preserves_event_structure(self):
line = json.dumps({'type':'item.completed','item':{'text':'password=secret','authorization':'Bearer hidden'}})
value = json.loads(c.redact(line))
self.assertEqual('item.completed',value['type'])
self.assertNotIn('secret',json.dumps(value))
self.assertNotIn('hidden',json.dumps(value))
class ProtocolRunner(c.ProcessRunner):
def codex_command(self, engine, *args):
return [sys.executable, str(Path(__file__).with_name('fake_codex.py')), *args]
class SubprocessTests(unittest.TestCase):
tearDown = EngineTests.tearDown
git_at = staticmethod(EngineTests.git_at)
run_engine = EngineTests.run_engine
def setUp(self):
EngineTests.setUp(self)
self.engine.runner = ProtocolRunner()
self.engine.config['workerCodeXHome'] = str(self.root / 'auth-only')
def test_cli_protocol_real_process_and_complete_prompt(self):
self.assertEqual('COMPLETED', self.run_engine()['stages']['TWO']['status'])
observed = (self.repo / 'observed-prompt.txt').read_text(encoding='utf-8')
self.assertIn('Complete synthetic task. No real backlog.', observed)
self.assertIn('ACTIVE CHECKPOINT ONLY', observed)
self.assertTrue(list((self.control / 'sessions').rglob('session.json')))
def test_cli_nonzero_exit_never_accepts(self):
(self.repo / 'simulate-interruption').write_text('synthetic')
self.git_at(self.repo,'add','.')
self.git_at(self.repo,'commit','-m','synthetic interruption fixture')
commit = self.git_at(self.repo,'rev-parse','HEAD')
self.git_at(self.repo,'push','origin','HEAD:refs/heads/codex/base')
self.manifest['baselineCommit'] = commit
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
state['integrationCommit'] = commit
self.engine.store.write(state)
self.assertEqual('FAILED', self.run_engine()['stages']['ONE']['status'])
def test_real_check_report_must_match_expected_membership(self):
check = self.manifest['stages'][0]['checkpoints'][0]['checks'][0]
xml = ''
check['argv'] = [sys.executable,'-c',
f'from pathlib import Path; Path("suite.xml").write_text({xml!r})']
check['report'] = dict(format='junit',path='suite.xml',expectedNames=['.proof'])
state = self.engine.store.read()
state['manifestSha256'] = c.digest(self.manifest)
self.engine.store.write(state)
self.assertEqual('COMPLETED', self.run_engine()['stages']['ONE']['status'])
def test_prompt_stall_still_obeys_timeout(self):
started = time.monotonic()
with self.assertRaises(c.Interrupted):
self.engine.runner.supervise(self.engine,
[sys.executable,'-c','import time; time.sleep(8)'], self.control / 'stall',
1, stdin='x' * 4_000_000)
self.assertLess(time.monotonic() - started, 5)
def test_no_newline_output_obeys_log_budget(self):
self.engine.config['maxLogBytes'] = 1024
with self.assertRaisesRegex(c.GateError, 'log byte budget'):
self.engine.runner.supervise(self.engine,
[sys.executable,'-c','import sys,time; sys.stdout.write("x"*200000); sys.stdout.flush(); time.sleep(8)'],
self.control / 'huge-line', 3)
def test_marker_write_failure_does_not_leave_worker(self):
original = c.atomic
def fail_marker(path, value):
if Path(path).name == 'active-process.json' and value.get('pid'):
raise OSError('synthetic disk full')
return original(path,value)
with mock.patch.object(c,'atomic',side_effect=fail_marker):
with self.assertRaises(OSError):
self.engine.runner.supervise(self.engine,
[sys.executable,'-c','import time; time.sleep(8)'],self.control / 'disk-full',3)
child = self.engine.runner.child
if child and child.poll() is None:
child.kill()
child.wait()
self.fail('Worker survived marker-write failure')
@unittest.skipUnless(os.name == 'nt', 'Windows suspended launch only')
def test_worker_cannot_execute_before_job_assignment(self):
early = self.repo / 'early.txt'
original = c.WindowsJob.assign
def delayed_assign(job, process):
time.sleep(.3)
self.assertFalse(early.exists(), 'Worker ran before containment assignment')
return original(job, process)
with mock.patch.object(c.WindowsJob,'assign',delayed_assign):
self.engine.runner.supervise(self.engine,[sys.executable,'-c',
f'from pathlib import Path; Path({str(early)!r}).write_text("started")'],self.control / 'suspended',5)
self.assertTrue(early.exists())
@unittest.skipUnless(os.name == 'nt', 'Windows Job Objects only')
def test_killed_controller_terminates_worker_and_grandchild(self):
script = self.root / 'parent.py'
grandchild = self.root / 'grandchild.pid'
worker_code = (f'import subprocess,sys,time; from pathlib import Path; '
f'p=subprocess.Popen([sys.executable,"-c","import time; time.sleep(60)"]); '
f'Path({str(grandchild)!r}).write_text(str(p.pid)); time.sleep(60)')
script.write_text('import sys\nfrom pathlib import Path\n'
f'sys.path.insert(0,{str(Path(c.__file__).parent)!r})\nimport controller as c\n'
f'e=c.Engine(Path({str(self.control)!r}), {self.manifest!r}, {self.config!r})\n'
f'e.runner.supervise(e,[sys.executable,"-c",{worker_code!r}],Path({str(self.control / "orphan")!r}),60)\n',encoding='utf-8')
parent = subprocess.Popen([sys.executable,str(script)],stdout=subprocess.DEVNULL,stderr=subprocess.PIPE)
try:
until = time.monotonic() + 10
while not grandchild.exists() and parent.poll() is None and time.monotonic() < until:
time.sleep(.05)
self.assertTrue(grandchild.exists(), parent.stderr.read().decode() if parent.poll() is not None else 'child did not start')
worker_pid = c.read_json(self.control / 'active-process.json')['pid']
grandchild_pid = int(grandchild.read_text())
parent.kill()
parent.wait(timeout=5)
import ctypes
api = ctypes.WinDLL('kernel32',use_last_error=True)
api.OpenProcess.restype = ctypes.c_void_p
api.OpenProcess.argtypes = [ctypes.c_ulong,ctypes.c_int,ctypes.c_ulong]
api.WaitForSingleObject.argtypes = [ctypes.c_void_p,ctypes.c_ulong]
api.CloseHandle.argtypes = [ctypes.c_void_p]
for pid in (worker_pid,grandchild_pid):
handle = api.OpenProcess(0x100000,0,pid)
if handle:
self.assertEqual(0,api.WaitForSingleObject(handle,5000),f'contained process {pid} survived')
api.CloseHandle(handle)
finally:
if parent.poll() is None:
parent.kill()
parent.wait()
parent.stderr.close()
if __name__ == '__main__':
unittest.main(verbosity=2)