wt
7a737eeb49
Changes: Record completed termination on the Popen instance after all cleanup steps succeed; avoid repeated signals to the reaped group. Add deterministic repeated-EPERM and failed-descendant-kill retry regressions without suppressing permission errors. Validation: Four focused cleanup tests passed; make check passed 72 tests and plan/snapshot validation. git diff --check passed. Natural prior EPERM did not recur; injected EPERM reproduced the pre-fix failure. No application or parent-task acceptance is claimed. Task: S00-014 Agent: auto-ci-cd-14343249da62 Role: ci-cd Action: claim Status: review Task title: Fix repeated runner cleanup after a submitted parent is reaped
401 lines
23 KiB
Python
401 lines
23 KiB
Python
"""Isolated blocker orchestration tests. No real model/provider/product task is run."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
from unittest.mock import Mock, patch
|
|
|
|
from test_tasks import WorkflowTests, ROOT, MODULE
|
|
|
|
RUN_SPEC = importlib.util.spec_from_file_location('luma_runner', ROOT / 'scripts/runner.py')
|
|
RUN_MODULE = importlib.util.module_from_spec(RUN_SPEC)
|
|
RUN_SPEC.loader.exec_module(RUN_MODULE)
|
|
|
|
|
|
class BlockerTests(WorkflowTests):
|
|
# Inherit temporary fixture helpers, but do not re-run the inherited test suite.
|
|
def issue(self, **changes):
|
|
result = {'title': 'Fix concrete cause', 'reason': 'A required local condition is unavailable',
|
|
'role': 'front', 'reviewer_role': 'qa', 'scope': ['src/fix.*'],
|
|
'outcome': 'Make the single missing condition verifiably available',
|
|
'steps': ['Reproduce', 'Fix local cause', 'Check'],
|
|
'acceptance': ['Original failure resolved'], 'checks': ['Positive and negative fixture'],
|
|
'requirements': ['REQ-001']}
|
|
result.update(changes)
|
|
return result
|
|
|
|
def finish(self, ident, owner, reviewer):
|
|
self.tracker.mutate('claim', ident, owner, role='front')
|
|
self.tracker.mutate('submit', ident, owner,
|
|
evidence=self.evidence(ident + '-' + owner + '.md'),
|
|
handoff=self.evidence(ident + '.md', 'state/handoffs'))
|
|
self.tracker.mutate('approve', ident, reviewer, role='qa',
|
|
evidence=self.evidence(ident + '-review.md'))
|
|
|
|
def test_escalation_blocks_parent_creates_ready_child_and_retires(self):
|
|
self.claim()
|
|
result = self.tracker.escalate('S00-001', 'owner', 'front', self.issue())
|
|
child = result['resolution_task']
|
|
self.assertTrue(result['agent_must_exit'])
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked')
|
|
self.assertIn(child, self.tracker.task('S00-001')['depends_on'])
|
|
self.assertNotIn('S00-001', self.tracker.task(child)['depends_on'])
|
|
self.assertEqual(self.tracker.ready()[0]['id'], child)
|
|
with self.assertRaisesRegex(ValueError, 'fresh session'):
|
|
self.claim('S00-003', 'owner')
|
|
self.tracker.validate() # Child card is in the same journal, no render race.
|
|
self.tracker.render()
|
|
|
|
def test_reviewed_resolution_requeues_parent_for_new_agent(self):
|
|
self.claim()
|
|
child = self.tracker.escalate('S00-001', 'owner', 'front', self.issue())['resolution_task']
|
|
self.finish(child, 'resolver', 'resolver-reviewer')
|
|
row = self.tracker.state()['tasks']['S00-001']
|
|
self.assertEqual((row['status'], row['owner']), ('todo', None))
|
|
self.assertEqual(self.tracker.ready()[0]['id'], 'S00-001')
|
|
with self.assertRaises(ValueError):
|
|
self.claim('S00-001', 'owner')
|
|
self.claim('S00-001', 'fresh-owner')
|
|
self.assertIn('Previous handoff', self.tracker.packet('S00-001', 'fresh-owner'))
|
|
|
|
def test_submit_only_does_not_unblock(self):
|
|
self.claim()
|
|
child = self.tracker.escalate('S00-001', 'owner', 'front', self.issue())['resolution_task']
|
|
self.tracker.mutate('claim', child, 'resolver', role='front')
|
|
self.tracker.mutate('submit', child, 'resolver', evidence=self.evidence('resolve.md'), handoff=self.evidence('resolve.md', 'state/handoffs'))
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked')
|
|
|
|
def test_reviewer_can_escalate_then_new_reviewer_continues(self):
|
|
self.claim()
|
|
self.submit()
|
|
child = self.tracker.escalate('S00-001', 'reviewer', 'qa', self.issue())['resolution_task']
|
|
self.finish(child, 'resolver', 'resolve-review')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'review')
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('approve', 'S00-001', 'reviewer', role='qa', evidence=self.evidence('review.md'))
|
|
self.tracker.mutate('approve', 'S00-001', 'fresh-reviewer', role='qa', evidence=self.evidence('fresh-review.md'))
|
|
|
|
def test_cycle_or_unbounded_scope_is_rejected_without_partial_state(self):
|
|
self.claim()
|
|
before = self.tracker.state()
|
|
for changes in [{'depends_on': ['S00-001']}, {'depends_on': ['S00-002']}, {'scope': ['**']}, {'scope': ['../outside/**']}, {'scope': ['specs/tz_clear.md']}]:
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.escalate('S00-001', 'owner', 'front', self.issue(**changes))
|
|
self.assertEqual(self.tracker.state(), before)
|
|
self.assertEqual(len(self.tracker.definitions()['tasks']), 3)
|
|
|
|
def test_external_blocker_waits_until_release_and_verification(self):
|
|
self.claim()
|
|
child = self.tracker.escalate('S00-001', 'owner', 'front', self.issue(requires_user_input=True))['resolution_task']
|
|
self.assertEqual(self.tracker.state()['tasks'][child]['status'], 'blocked')
|
|
self.assertNotIn(child, [r['id'] for r in self.tracker.ready()])
|
|
self.tracker.render()
|
|
self.tracker.validate()
|
|
MODULE.task_extensions.release_external(self.tracker, child, 'coordinator', 'orchestrator', 'Required test service restored')
|
|
self.assertEqual(self.tracker.state()['tasks'][child]['status'], 'todo')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked')
|
|
|
|
def test_duplicate_escalation_cannot_spawn_second_child(self):
|
|
self.claim()
|
|
self.tracker.escalate('S00-001', 'owner', 'front', self.issue())
|
|
before = self.tracker.state()
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.escalate('S00-001', 'owner', 'front', self.issue())
|
|
self.assertEqual(before, self.tracker.state())
|
|
self.assertEqual(len(self.tracker.definitions()['tasks']), 4)
|
|
|
|
def test_invalid_context_or_string_boolean_rejected(self):
|
|
self.claim()
|
|
for change in [{'requires_user_input': 'false'}, {'ui_sections': [999]}, {'spec_sections': 'all'}]:
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.escalate('S00-001', 'owner', 'front', self.issue(**change))
|
|
self.assertEqual(len(self.tracker.definitions()['tasks']), 3)
|
|
|
|
def test_nested_blocker_limit_waits_for_external_resolution(self):
|
|
self.claim()
|
|
current, agent = 'S00-001', 'owner'
|
|
for depth in range(1, 5):
|
|
result = self.tracker.escalate(current, agent, 'front', self.issue())
|
|
current = result['resolution_task']
|
|
if depth < 4:
|
|
agent = 'resolver-' + str(depth)
|
|
self.claim(current, agent)
|
|
self.assertTrue(result['requires_user_input'])
|
|
self.assertEqual(self.tracker.task(current)['blocker_depth'], 4)
|
|
|
|
def test_plan_state_journal_recovers_after_interrupted_artifact_write(self):
|
|
self.claim()
|
|
real = MODULE.atomic_write
|
|
calls = []
|
|
def fail_after_journal(path, text):
|
|
calls.append(str(path))
|
|
if len(calls) == 2:
|
|
raise OSError('Simulated interruption')
|
|
return real(path, text)
|
|
with patch.object(MODULE, 'atomic_write', side_effect=fail_after_journal):
|
|
with self.assertRaises(OSError):
|
|
self.tracker.escalate('S00-001', 'owner', 'front', self.issue())
|
|
self.assertTrue(self.tracker.journal_path.exists())
|
|
recovered = self.tracker.state()
|
|
self.assertFalse(self.tracker.journal_path.exists())
|
|
self.assertEqual(recovered['tasks']['S00-001']['status'], 'blocked')
|
|
child = recovered['tasks']['S00-001']['blocked_by'][0]
|
|
self.assertIn(child, self.tracker.index())
|
|
self.tracker.render()
|
|
self.tracker.validate()
|
|
|
|
|
|
class RunnerTests(WorkflowTests):
|
|
def setUp(self):
|
|
super().setUp()
|
|
for rel in ['AGENTS.md', 'README.md', 'master_prompt.md', 'roles/common.md',
|
|
'docs/WEB_PROFILE.md', 'docs/ARCHITECTURE.md', 'docs/CONTRACTS.md',
|
|
'docs/WORKFLOW.md', 'docs/AUTOMATION.md', 'docs/GIT_WORKFLOW.md']:
|
|
path = self.root / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text('# Fresh one-task instructions\n', encoding='utf-8')
|
|
fake = self.root / 'fake_agent.py'
|
|
fake.write_text('''import importlib.util,json,os,pathlib,sys
|
|
spec=importlib.util.spec_from_file_location('tool',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
|
|
r=pathlib.Path(os.environ['LUMA_PROJECT_ROOT']);t=m.Tracker(r)
|
|
id=os.environ['LUMA_TASK_ID'];agent=os.environ['LUMA_AGENT_ID'];role=os.environ['LUMA_ROLE'];action=os.environ['LUMA_ACTION']
|
|
prompt=sys.stdin.read();assert 'fresh session' in prompt
|
|
log=r/'state/test-launches.json';entries=json.loads(log.read_text()) if log.exists() else [];entries.append({'task':id,'agent':agent,'action':action});log.write_text(json.dumps(entries))
|
|
if id=='S00-001' and action=='claim' and not (r/'state/first-blocker').exists():
|
|
(r/'state/first-blocker').write_text('recorded')
|
|
issue={'title':'Fix local missing input','reason':'fixture blocker','role':'front','reviewer_role':'qa','scope':['src/fix.*'],'outcome':'Restore required local input','steps':['Restore local input'],'acceptance':['Input available'],'checks':['Original check succeeds'],'requirements':['REQ-001']}
|
|
t.escalate(id,agent,role,issue);sys.exit(0)
|
|
e=r/'state/evidence';h=r/'state/handoffs';e.mkdir(exist_ok=True);h.mkdir(exist_ok=True)
|
|
evidence=e/(id+'-'+agent+'.md');evidence.write_text('# Evidence\\n\\nActual isolated fixture check succeeded; expected negative outcome checked.\\n')
|
|
if action=='review':t.mutate('approve',id,agent,role=role,evidence=str(evidence))
|
|
else:
|
|
handoff=h/(id+'.md');handoff.write_text('# Handoff\\n\\nFixture implementation completed; next step independent review of this task.\\n')
|
|
t.mutate('submit',id,agent,evidence=str(evidence),handoff=str(handoff))
|
|
''', encoding='utf-8')
|
|
self.config = {'git_commits': False, 'argv': [sys.executable, str(fake), str(ROOT / 'scripts/tasks.py')], 'max_sessions': 20, 'timeout_seconds': 5}
|
|
|
|
def test_fresh_process_blocker_resolver_review_resume_chain(self):
|
|
runner = RUN_MODULE.Runner(self.root, self.config)
|
|
result = runner.run()
|
|
self.assertEqual(result['stop'], 'complete')
|
|
launches = json.loads((self.root / 'state/test-launches.json').read_text())
|
|
self.assertEqual(len(launches), 9)
|
|
self.assertEqual(len({x['agent'] for x in launches}), 9)
|
|
original = [x for x in launches if x['task'] == 'S00-001' and x['action'] == 'claim']
|
|
self.assertEqual(len(original), 2)
|
|
self.assertNotEqual(original[0]['agent'], original[1]['agent'])
|
|
self.assertIn(original[0]['agent'], self.tracker.state()['retired_agents'])
|
|
self.assertFalse(runner.lock_path.exists())
|
|
|
|
def test_dry_run_does_not_claim_or_write(self):
|
|
runner = RUN_MODULE.Runner(self.root, {})
|
|
before = self.tracker.state()
|
|
result = runner.run(dry_run=True)
|
|
self.assertTrue(result['dry_run'])
|
|
self.assertEqual(self.tracker.state(), before)
|
|
self.assertFalse((self.root / 'state/runs').exists())
|
|
|
|
def test_runner_lock_prevents_second_coordinator(self):
|
|
runner = RUN_MODULE.Runner(self.root, self.config)
|
|
with runner.lock():
|
|
with self.assertRaises(ValueError):
|
|
runner.run()
|
|
|
|
def test_missing_executable_creates_external_resolution_and_stops(self):
|
|
runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': ['missing-tool-8b5e6d']})
|
|
result = runner.run()
|
|
self.assertEqual(result['stop'], 'needs_input')
|
|
self.assertEqual(result['launched'], 1)
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked')
|
|
self.assertTrue(result['external_blockers'])
|
|
|
|
def test_incomplete_agent_exit_creates_remediation(self):
|
|
runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': [sys.executable, '-c', 'pass'], 'max_sessions': 1})
|
|
result = runner.run()
|
|
self.assertEqual(result['stop'], 'session_limit')
|
|
state = self.tracker.state()
|
|
self.assertEqual(state['tasks']['S00-001']['status'], 'blocked')
|
|
self.assertTrue(state['retired_agents'])
|
|
|
|
def test_delayed_reader_receives_complete_large_utf8_prompt_and_eof(self):
|
|
fake = self.root / 'fake_agent.py'
|
|
fake.write_text(fake.read_text().replace(
|
|
"prompt=sys.stdin.read();assert 'fresh session' in prompt",
|
|
"import time;time.sleep(.5)\n"
|
|
"payload=sys.stdin.buffer.read()\n"
|
|
"assert payload == pathlib.Path(os.environ['LUMA_PROMPT_FILE']).read_bytes()\n"
|
|
"prompt=payload.decode('utf-8');assert 'fresh session' in prompt\n"
|
|
"print('Complete UTF-8 prompt and EOF:',len(payload),flush=True)"))
|
|
(self.root / 'state/first-blocker').write_text('skip fixture blocker')
|
|
prompt = 'This is a fresh session.\n' + '\U0001f642\u0416\n' * 20000
|
|
runner = RUN_MODULE.Runner(self.root, dict(self.config, max_sessions=1,
|
|
timeout_seconds=3))
|
|
with patch.object(runner, 'prompt', return_value=prompt):
|
|
result = runner.run()
|
|
self.assertEqual(result['launched'], 1)
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'review')
|
|
record_path = next((self.root / 'state/runs').glob('*/run.json'))
|
|
record = json.loads(record_path.read_text())
|
|
self.assertEqual(record['result'], {'status': 'review'})
|
|
self.assertEqual(record['scope_violations'], [])
|
|
self.assertIn('Complete UTF-8 prompt and EOF: ' + str(len(prompt.encode('utf-8'))),
|
|
record_path.with_name('output.log').read_text())
|
|
|
|
def test_scope_violation_quarantines_and_requires_input(self):
|
|
runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': [sys.executable, '-c', "from pathlib import Path;Path('bad-edit.txt').write_text('wrong scope')"]})
|
|
result = runner.run()
|
|
self.assertEqual(result['stop'], 'needs_input')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked')
|
|
record = json.loads(next((self.root / 'state/runs').glob('*/run.json')).read_text())
|
|
self.assertEqual(record['scope_violations'], ['bad-edit.txt'])
|
|
|
|
def test_reporter_process_is_stopped_after_escalate_without_waiting_timeout(self):
|
|
import time
|
|
fake = self.root / 'fake_agent.py'
|
|
fake.write_text(fake.read_text().replace('t.escalate(id,agent,role,issue);sys.exit(0)', 't.escalate(id,agent,role,issue);import time;time.sleep(10);sys.exit(0)'))
|
|
config = dict(self.config, max_sessions=1, timeout_seconds=5)
|
|
started = time.monotonic()
|
|
result = RUN_MODULE.Runner(self.root, config).run()
|
|
self.assertLess(time.monotonic() - started, 3)
|
|
self.assertEqual(result['launched'], 1)
|
|
record = json.loads(next((self.root / 'state/runs').glob('*/run.json')).read_text())
|
|
self.assertTrue(record['stopped_after_transition'])
|
|
self.assertEqual(len(self.tracker.definitions()['tasks']), 4)
|
|
self.assertEqual(record['result']['blocked'], True)
|
|
|
|
def test_invalid_stage_is_not_silent_idle(self):
|
|
with self.assertRaisesRegex(ValueError, 'Unknown stage'):
|
|
RUN_MODULE.Runner(self.root, {}).run(stage='S99', dry_run=True)
|
|
|
|
def test_timeout_terminates_before_queuing_resolution(self):
|
|
runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': [sys.executable, '-c', 'import time;time.sleep(2)'], 'timeout_seconds': .05, 'max_sessions': 1})
|
|
result = runner.run()
|
|
self.assertEqual(result['stop'], 'session_limit')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked')
|
|
|
|
def _check_descendant_cleanup(self, transition):
|
|
import time
|
|
ready = self.root / 'state/descendant-ready'
|
|
late_write = self.root / 'src/late-descendant-write.txt'
|
|
child_code = (
|
|
"import pathlib,signal,time;signal.signal(signal.SIGTERM,signal.SIG_IGN);"
|
|
f"pathlib.Path({str(ready)!r}).write_text('ready');"
|
|
"time.sleep(1.5);"
|
|
f"pathlib.Path({str(late_write)!r}).write_text('unexpected')"
|
|
)
|
|
startup = (
|
|
"import subprocess,time\n"
|
|
f"subprocess.Popen([sys.executable,'-c',{child_code!r}])\n"
|
|
f"while not pathlib.Path({str(ready)!r}).exists():time.sleep(.01)\n"
|
|
)
|
|
fake = self.root / 'fake_agent.py'
|
|
source = fake.read_text().replace(
|
|
"prompt=sys.stdin.read();assert 'fresh session' in prompt",
|
|
startup + ("prompt=sys.stdin.read();assert 'fresh session' in prompt"
|
|
if transition else "time.sleep(10)"))
|
|
fake.write_text(source + '\nimport time;time.sleep(10)\n')
|
|
(self.root / 'state/first-blocker').write_text('skip fixture blocker')
|
|
runner = RUN_MODULE.Runner(self.root, dict(self.config, max_sessions=1,
|
|
timeout_seconds=2 if transition else .5))
|
|
result = runner.run()
|
|
self.assertEqual(result['launched'], 1)
|
|
self.assertTrue(ready.is_file(), 'Descendant must start before cleanup')
|
|
record = json.loads(next((self.root / 'state/runs').glob('*/run.json')).read_text())
|
|
self.assertEqual(record['stopped_after_transition'], transition)
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'],
|
|
'review' if transition else 'blocked')
|
|
time.sleep(1.6)
|
|
self.assertFalse(late_write.exists(), 'Descendant survived process-group cleanup')
|
|
self.assertEqual(record['scope_violations'], [])
|
|
|
|
@unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup')
|
|
def test_timeout_stops_source_writing_descendant(self):
|
|
self._check_descendant_cleanup(transition=False)
|
|
|
|
@unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup')
|
|
def test_submit_transition_stops_source_writing_descendant(self):
|
|
self._check_descendant_cleanup(transition=True)
|
|
|
|
@unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup')
|
|
def test_completed_cleanup_does_not_signal_reaped_group_again(self):
|
|
process = Mock(pid=12345, returncode=-15)
|
|
process.poll.return_value = -15
|
|
with patch.object(RUN_MODULE.os, 'killpg') as killpg:
|
|
RUN_MODULE.stop_process_tree(process)
|
|
self.assertEqual(killpg.call_args_list,
|
|
[unittest.mock.call(12345, RUN_MODULE.signal.SIGTERM),
|
|
unittest.mock.call(12345, RUN_MODULE.signal.SIGKILL)])
|
|
killpg.side_effect = PermissionError('Reaped process group inaccessible')
|
|
RUN_MODULE.stop_process_tree(process)
|
|
self.assertEqual(killpg.call_count, 2)
|
|
|
|
@unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup')
|
|
def test_failed_descendant_kill_is_not_marked_complete(self):
|
|
process = Mock(pid=12345, returncode=-15)
|
|
process.poll.return_value = -15
|
|
with patch.object(RUN_MODULE.os, 'killpg',
|
|
side_effect=[None, PermissionError('Live descendant inaccessible')]) as killpg:
|
|
with self.assertRaises(PermissionError):
|
|
RUN_MODULE.stop_process_tree(process)
|
|
killpg.side_effect = None
|
|
RUN_MODULE.stop_process_tree(process)
|
|
self.assertEqual(killpg.call_count, 4)
|
|
|
|
def test_scope_matcher_checks_modified_created_deleted_files(self):
|
|
self.assertEqual(RUN_MODULE.scope_violations({'a': 'old', 'gone': 'old'}, {'a': 'new', 'created': 'new'}, ['a']), ['created', 'gone'])
|
|
|
|
|
|
class FoundationRunnerTests(unittest.TestCase):
|
|
def test_ui_acceptance_prompt_keeps_exact_context_and_explicit_upstream_reads(self):
|
|
runner = RUN_MODULE.Runner(ROOT, {})
|
|
task = runner.tracker.task('S21-006')
|
|
for action, role in [('claim', task['role']), ('review', task['reviewer_role'])]:
|
|
with self.subTest(action=action):
|
|
prompt = runner.prompt({'id': task['id'], 'role': role, 'action': action}, 'auto-preview')
|
|
self.assertLessEqual(len(prompt), 80000)
|
|
self.assertIn(runner.tracker.packet(task['id'], 'auto-preview'), prompt)
|
|
self.assertIn('Required task input reads (not inlined)', prompt)
|
|
self.assertIn('Before implementation or review, read these task inputs', prompt)
|
|
self.assertIn('a missing required artifact is a blocker', prompt)
|
|
for raw in task['inputs']:
|
|
self.assertIn(raw, prompt)
|
|
if ' (upstream artifact' in raw:
|
|
self.assertIn('- ' + raw, prompt)
|
|
self.assertNotIn('# File: ' + raw.split(' (')[0] + '\n', prompt)
|
|
# Every UI Kit section stays verbatim, including tables and
|
|
# unnumbered details beyond the short requirement registry.
|
|
spec = (ROOT / 'specs/uikit_clear.md').read_text(encoding='utf-8')
|
|
for section in task['ui_sections']:
|
|
self.assertIn(runner.tracker.section(spec, section), prompt)
|
|
self.assertIn('This is a fresh session.', prompt)
|
|
self.assertIn('After submit/approve/reject EXIT.', prompt)
|
|
|
|
def test_all_current_fresh_agent_prompts_fit_configured_budget(self):
|
|
runner = RUN_MODULE.Runner(ROOT, {})
|
|
lengths = []
|
|
for task in runner.tracker.definitions()['tasks']:
|
|
for action, role in [('claim', task['role']), ('review', task['reviewer_role'])]:
|
|
with self.subTest(task=task['id'], action=action):
|
|
lengths.append(len(runner.prompt({'id': task['id'], 'role': role, 'action': action}, 'auto-preview')))
|
|
self.assertLessEqual(max(lengths), 80000)
|
|
|
|
def test_unconfigured_runner_does_not_claim_product_work(self):
|
|
before = (ROOT / 'state/tasks.json').read_bytes()
|
|
result = subprocess.run([sys.executable, str(ROOT / 'scripts/runner.py'), '--config', 'missing-' + uuid.uuid4().hex + '.json'], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn('configuration', result.stderr)
|
|
self.assertEqual(before, (ROOT / 'state/tasks.json').read_bytes())
|
|
|
|
|
|
# Imported helper class is not a second copy of its original tests.
|
|
# Remove inherited test methods from these subclasses' discovery via explicit suite.
|
|
def load_tests(loader, tests, pattern):
|
|
import unittest
|
|
suite = unittest.TestSuite()
|
|
for cls in [BlockerTests, RunnerTests, FoundationRunnerTests]:
|
|
suite.addTests(cls(name) for name in cls.__dict__ if name.startswith('test_'))
|
|
return suite
|