179 lines
9.9 KiB
Python
179 lines
9.9 KiB
Python
"""Distinct capture/cleanup failures; simulated CLI is not real-model evidence."""
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import runpy
|
|
import sys
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from codex_context import capture_tool_context, selected_event
|
|
|
|
|
|
class ManagedCoverageTests(unittest.TestCase):
|
|
def test_verifies_extra_counts_and_preserves_unknown_fields(self):
|
|
review = runpy.run_path(str(Path(__file__).with_name('review-model.py')))
|
|
project = review['scope_with_verified_counts']
|
|
expected = {'state': 'partial', 'issues': ['unreturned rows remain unknown']}
|
|
original = {**expected, 'module_ref_count': 0, 'pinvoke_import_count': 2}
|
|
scope, checks = project(original, expected, {'module_ref_count': 0, 'pinvoke_import_count': 2})
|
|
self.assertEqual(scope, expected)
|
|
self.assertTrue(all(checks.values()))
|
|
self.assertEqual(original['pinvoke_import_count'], 2)
|
|
for count in [3, '2', True]:
|
|
scope, checks = project({**expected, 'pinvoke_import_count': count}, expected, {'pinvoke_import_count': 2})
|
|
self.assertFalse(checks['pinvoke_import_count'])
|
|
_, checks = project({**expected, 'module_ref_count': False}, {**expected, 'module_ref_count': 0}, {'module_ref_count': 0})
|
|
self.assertFalse(checks['module_ref_count'])
|
|
scope, _ = project({**expected, 'unsupported_claim': True}, expected, {})
|
|
self.assertNotEqual(scope, expected)
|
|
scope, _ = project({'state': 'partial', 'module_ref_count': 0}, expected, {'module_ref_count': 0})
|
|
self.assertNotEqual(scope, expected)
|
|
ambiguous = review['named_scope']({'native_boundaries': expected, 'native_boundary_declarations': {'state': 'complete'}}, ['native_boundaries', 'native_boundary_declarations'])
|
|
scope, _ = project(ambiguous, expected, {'module_ref_count': 0})
|
|
self.assertNotEqual(scope, expected)
|
|
|
|
|
|
class ContextCaptureTests(unittest.TestCase):
|
|
def test_filters_ambient_account_and_nested_usage_state(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
home = root / 'home'
|
|
sessions = home / 'sessions'
|
|
sessions.mkdir(parents=True)
|
|
output = root / 'output'
|
|
output.mkdir()
|
|
events = [
|
|
{'type': 'world_state', 'payload': {'state': {'env': 'ambient-marker'}}},
|
|
{'type': 'session_meta', 'payload': {'creator_account_id': 'account-marker'}},
|
|
{'type': 'response_item', 'payload': {
|
|
'type': 'custom_tool_call_output', 'call_id': 'original-call',
|
|
'output': 'caller-selected evidence and paths',
|
|
'internal_chat_message_metadata_passthrough': 'private-marker'}},
|
|
{'type': 'token_usage_record', 'payload': {
|
|
'response_id': 'original-response', 'usage': {
|
|
'input_tokens': 12, 'cached_input_tokens': 4,
|
|
'environment': 'nested-ambient-marker'},
|
|
'transport_auth': 'auth-marker'}},
|
|
]
|
|
(sessions / 'rollout.jsonl').write_text(''.join(json.dumps(x) + '\n' for x in events))
|
|
(home / 'auth.json').write_text('copied-credential-marker')
|
|
receipt = capture_tool_context(home, output)
|
|
text = (output / 'codex-tool-context.jsonl').read_text()
|
|
self.assertEqual(receipt['status'], 'captured')
|
|
self.assertEqual(receipt['retained_events'], 2)
|
|
for marker in ['ambient-marker', 'account-marker', 'private-marker',
|
|
'auth-marker', 'copied-credential-marker']:
|
|
self.assertNotIn(marker, text)
|
|
self.assertIn('caller-selected evidence and paths', text)
|
|
self.assertIn('original-response', text)
|
|
# Windows mode bits do not describe access control; this is a POSIX assertion.
|
|
if os.name == 'posix':
|
|
self.assertEqual((output / 'codex-tool-context.jsonl').stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_missing_rollouts_are_unknown_and_malformed_lines_remain_partial(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
output = root / 'missing'
|
|
output.mkdir()
|
|
self.assertEqual(capture_tool_context(root / 'home', output)['status'], 'unavailable')
|
|
sessions = root / 'home' / 'sessions'
|
|
sessions.mkdir(parents=True)
|
|
(sessions / 'rollout.jsonl').write_text('not-json\n' + json.dumps({
|
|
'type': 'event_msg', 'payload': {'type': 'token_count', 'info': None}}) + '\n')
|
|
partial = root / 'partial'
|
|
partial.mkdir()
|
|
receipt = capture_tool_context(root / 'home', partial)
|
|
self.assertEqual(receipt['status'], 'partial')
|
|
self.assertEqual(receipt['errors'][0]['line'], 1)
|
|
self.assertEqual(receipt['retained_events'], 1)
|
|
with self.assertRaises(ValueError):
|
|
selected_event({'type': 'token_usage_record', 'payload': {'usage': {'input_tokens': True}}})
|
|
with self.assertRaises(ValueError):
|
|
selected_event({'type': {'environment': 'unexpected'}, 'payload': {}})
|
|
with self.assertRaises(ValueError):
|
|
selected_event({'type': 'response_item', 'timestamp': {'environment': 'unexpected'}, 'payload': {}})
|
|
unknown = selected_event({'type': 'event_msg', 'payload': {'type': 'token_count', 'info': {'model_context_window': None}}})
|
|
self.assertIsNone(unknown['payload']['info']['model_context_window'])
|
|
|
|
def test_capture_failure_still_removes_trial_credentials_and_profile(self):
|
|
spec = importlib.util.spec_from_file_location('agent_trial', Path(__file__).with_name('agent-trial.py'))
|
|
agent = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(agent)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
runtime = root / 'runtime'
|
|
(runtime / 'scripts').mkdir(parents=True)
|
|
(runtime / 'scripts' / 'rea.mjs').write_text('')
|
|
(runtime / 'package-lock.json').write_text('{}')
|
|
auth_home = root / 'source-auth'
|
|
auth_home.mkdir()
|
|
(auth_home / 'auth.json').write_text('synthetic-credential')
|
|
question = root / 'question.txt'
|
|
question.write_text('Observe the declared fixture.')
|
|
output = root / 'trial'
|
|
child = Mock(stdin=io.StringIO())
|
|
child.wait.return_value = 0
|
|
arguments = ['agent-trial.py', '--runtime-root', str(runtime), '--revision', 'fixture',
|
|
'--prompt', str(question), '--output', str(output), '--capture-tool-context']
|
|
with patch.dict(os.environ, {'CODEX_HOME': str(auth_home)}, clear=True), \
|
|
patch.object(sys, 'argv', arguments), \
|
|
patch.object(agent.shutil, 'which', return_value='/fixture/node'), \
|
|
patch.object(agent.subprocess, 'check_output', return_value='v24.18.0\n'), \
|
|
patch.object(agent.subprocess, 'Popen', return_value=child), \
|
|
patch.object(agent, 'capture_tool_context', side_effect=OSError('fixture capture failure')), \
|
|
patch.object(agent.Path, 'home', side_effect=AssertionError('Explicit CODEX_HOME must not resolve the default home')), \
|
|
patch('builtins.print'):
|
|
with self.assertRaisesRegex(OSError, 'fixture capture failure'):
|
|
agent.main()
|
|
self.assertFalse(list(output.glob('profile-*')))
|
|
self.assertTrue(json.loads((output / 'cleanup.json').read_text())['profile_removed'])
|
|
self.assertEqual((auth_home / 'auth.json').read_text(), 'synthetic-credential')
|
|
|
|
def test_refuses_foreign_session_directory_and_symlink_records(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
home = root / 'home'
|
|
home.mkdir()
|
|
foreign = root / 'foreign'
|
|
foreign.mkdir()
|
|
(foreign / 'rollout.jsonl').write_text('private marker')
|
|
(home / 'sessions').symlink_to(foreign, target_is_directory=True)
|
|
output = root / 'directory-result'
|
|
output.mkdir()
|
|
receipt = capture_tool_context(home, output)
|
|
self.assertEqual(receipt['status'], 'partial')
|
|
self.assertEqual(receipt['retained_events'], 0)
|
|
(home / 'sessions').unlink()
|
|
(home / 'sessions').mkdir()
|
|
(home / 'sessions' / 'linked.jsonl').symlink_to(foreign / 'rollout.jsonl')
|
|
output = root / 'file-result'
|
|
output.mkdir()
|
|
receipt = capture_tool_context(home, output)
|
|
self.assertEqual(receipt['status'], 'partial')
|
|
self.assertEqual(receipt['errors'][0]['reason'], 'not_an_owned_regular_file')
|
|
self.assertEqual((foreign / 'rollout.jsonl').read_text(), 'private marker')
|
|
|
|
def test_guidance_requires_selected_tools_before_acquiring_a_profile(self):
|
|
spec = importlib.util.spec_from_file_location('agent_trial', Path(__file__).with_name('agent-trial.py'))
|
|
agent = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(agent)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
output = root / 'trial'
|
|
arguments = ['agent-trial.py', '--runtime-root', str(root), '--revision', 'fixture',
|
|
'--prompt', str(root / 'question'), '--output', str(output), '--guide-enabled-tools']
|
|
with patch.object(sys, 'argv', arguments), patch.object(sys, 'stderr', io.StringIO()):
|
|
with self.assertRaises(SystemExit) as raised:
|
|
agent.main()
|
|
self.assertEqual(raised.exception.code, 2)
|
|
self.assertFalse(output.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|