diff --git a/.getzilla/getzilla/python_test_runner.py b/.getzilla/getzilla/python_test_runner.py index cd20b5b6..685c168c 100644 --- a/.getzilla/getzilla/python_test_runner.py +++ b/.getzilla/getzilla/python_test_runner.py @@ -337,9 +337,11 @@ def run_named_tests(root: Path, targets: list[str], *, budget: int = 180) -> Pro for target in targets ): raise RunnerError('named tests require nonempty existing tests.test_module[.Class[.test_method]] targets') - with tempfile.TemporaryDirectory(prefix='grok-named-smoke-') as directory: - return execute([sys.executable, '-m', 'unittest', *targets], root, - _environment(root, Path(directory) / '.coverage'), timeout=budget) + retained: dict[str, ProcessResult] = {} + with _run_directory('grok-named-smoke-', retained) as directory: + retained['tests'] = execute([sys.executable, '-m', 'unittest', *targets], root, + _environment(root, Path(directory) / '.coverage'), timeout=budget) + return retained['tests'] def _tool_versions(required: list[str]) -> dict[str, str]: @@ -617,10 +619,13 @@ def run_trust_tests(root: Path, workers: int) -> ProcessResult: command = (_pytest_command(workers, 'loadfile') + ['tests']) if workers else [ sys.executable, '-m', 'unittest', 'discover', '-s', 'tests', ] - with tempfile.TemporaryDirectory(prefix='grok-trust-tests-') as directory: + retained: dict[str, ProcessResult] = {} + # Child temp files now live in this folder, so its cleanup must not replace the verdict (#134). + with _run_directory('grok-trust-tests-', retained) as directory: environment = _environment(suite, Path(directory) / '.coverage') environment['PYTHONPATH'] = os.pathsep.join(str(path) for path in (suite / 'src', suite / 'tests', suite)) - return execute(command, suite, environment) + retained['tests'] = execute(command, suite, environment) + return retained['tests'] def main() -> int: diff --git a/scripts/getzilla_protected_write.py b/scripts/getzilla_protected_write.py index a4780b3e..35b4a7fb 100755 --- a/scripts/getzilla_protected_write.py +++ b/scripts/getzilla_protected_write.py @@ -3,6 +3,7 @@ import argparse import json +import os import sys from pathlib import Path @@ -10,6 +11,7 @@ sys.path.insert(0, str(ROOT / '.getzilla')) from getzilla.protected_write import ProtectedWriteError, apply_manifest +from getzilla.state import bind_session from getzilla.util import find_root @@ -24,6 +26,8 @@ def main() -> int: parser.add_argument('--manifest', required=True, type=Path) parser.add_argument('--dry-run', action='store_true') args = parser.parse_args() + # The caller's session, as in getzilla_approve.py, so a session-bound grant follows its later routes (#134). + bind_session(os.environ.get('CLAUDE_CODE_SESSION_ID') or os.environ.get('GETZILLA_SESSION_ID')) try: result = apply_manifest(find_root(ROOT), args.manifest, dry_run=args.dry_run) except ProtectedWriteError as exc: diff --git a/scripts/getzilla_report.py b/scripts/getzilla_report.py index 6391d066..62361937 100755 --- a/scripts/getzilla_report.py +++ b/scripts/getzilla_report.py @@ -38,6 +38,11 @@ def _cell(value: object) -> str: return '-' if value is None else str(value) +def _printable(value: object) -> str: + """Journal text for the terminal: control and format characters become `?` (#134).""" + return ''.join(char if char.isprintable() else '?' for char in str(value)) + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog='getzilla report', description=__doc__.split('\n')[0]) parser.add_argument('--since', default='7d', help='period: Nd, Nh or all (default 7d)') @@ -56,7 +61,7 @@ def main(argv: list[str] | None = None) -> int: if summary['profiles']: print(f'{"profile":<10} {"routes":>6} {"median min to pass":>19} {"verify runs/route":>18} {"reviews":>8} {"failed outcomes":>20}') for name, group in sorted(summary['profiles'].items()): - print(f'{name:<10} {group["routes"]:>6} {_cell(group["median_minutes_to_pass"]):>19} ' + print(f'{_printable(name):<10} {group["routes"]:>6} {_cell(group["median_minutes_to_pass"]):>19} ' f'{group["verify_runs_per_route"]:>18} {group["reviews"]:>8} {group["failed_check_or_review_outcomes"]:>20}') else: print('No routes recorded yet: metrics start with the next route and verification run.') @@ -68,7 +73,7 @@ def main(argv: list[str] | None = None) -> int: print(f'Route observation gaps ({observed["routes_with_gaps"]} of {observed["routes"]} routes; unknown is not zero):') gapped = [route for route in summary['routes'] if route.get('gaps')] for route in gapped[:MAX_GAP_ROUTES]: - print(f' {route["route_id"]}: ' + '; '.join(route['gaps'])) + print(f' {_printable(route["route_id"])}: ' + '; '.join(route['gaps'])) if len(gapped) > MAX_GAP_ROUTES: print(f' ... {len(gapped) - MAX_GAP_ROUTES} more; use --json for all.') hints = summary['counters'].get('path_hints', {}) diff --git a/tests/test_grant_lifecycle.py b/tests/test_grant_lifecycle.py index 252f26eb..eef795a5 100644 --- a/tests/test_grant_lifecycle.py +++ b/tests/test_grant_lifecycle.py @@ -193,6 +193,44 @@ def test_the_cli_records_the_caller_session(self) -> None: set_active_route(root, build_route(root, 'Next prompt', 's1').to_dict()) self.assertTrue(valid(root)) + def test_the_cli_falls_back_to_getzilla_session_id_and_prefers_claude_code(self) -> None: + approve = [sys.executable, str(ROOT / 'scripts' / 'getzilla_approve.py'), 'protected-path', + '--reason', 'one file', '--action', 'protected-path-write', '--resource', TARGET] + for env_extra, expected in (({'GETZILLA_SESSION_ID': 's1'}, 's1'), + ({'GETZILLA_SESSION_ID': 's2', 'CLAUDE_CODE_SESSION_ID': 's1'}, 's1')): + with self.subTest(env=env_extra), github_project() as root: + env = {key: value for key, value in os.environ.items() + if key not in {'CLAUDE_CODE_SESSION_ID', 'GETZILLA_SESSION_ID'}} + subprocess.run(approve, cwd=root, env={**env, **env_extra}, check=True, capture_output=True) + recorded = json.loads(approvals_path(root).read_text(encoding='utf-8'))[-1]['session_id'] + self.assertEqual(recorded, expected) + + def test_protected_write_cli_follows_the_caller_session(self) -> None: + import base64 + import shutil + from getzilla.util import file_sha256 + with github_project() as root: + (root / 'scripts').mkdir(exist_ok=True) + shutil.copy2(ROOT / 'scripts' / 'getzilla_protected_write.py', root / 'scripts') + target = root / 'AGENTS.md' + add_approval(root, 'protected-path', 'one file', 15, actions=['protected-path-write'], + resources=['AGENTS.md']) + set_active_route(root, build_route(root, 'Next prompt', 's1').to_dict()) + manifest = root.parent / 'protected-write.json' + manifest.write_text(json.dumps({'schema_version': 1, 'operations': [{ + 'path': 'AGENTS.md', 'expected_sha256': file_sha256(target), + 'content_base64': base64.b64encode(target.read_bytes() + b'\n# edit\n').decode('ascii')}]}), + encoding='utf-8') + cli = [sys.executable, str(root / 'scripts' / 'getzilla_protected_write.py'), '--manifest', str(manifest), + '--dry-run'] + env = {key: value for key, value in os.environ.items() + if key not in {'CLAUDE_CODE_SESSION_ID', 'GETZILLA_SESSION_ID'}} + unbound = subprocess.run(cli, cwd=root, env=env, capture_output=True, text=True) + self.assertEqual(unbound.returncode, 2, unbound.stdout) + bound = subprocess.run(cli, cwd=root, env={**env, 'CLAUDE_CODE_SESSION_ID': 's1'}, + capture_output=True, text=True) + self.assertEqual(bound.returncode, 0, bound.stderr) + def test_without_a_known_session_the_exact_route_is_still_required(self) -> None: with github_project() as root: approval = add_approval(root, 'protected-path', 'one file', 15, diff --git a/tests/test_python_test_runner.py b/tests/test_python_test_runner.py index 78a771b9..b90c6df3 100644 --- a/tests/test_python_test_runner.py +++ b/tests/test_python_test_runner.py @@ -134,6 +134,27 @@ def test_temp_files_a_killed_child_leaves_are_removed_with_the_run(self): self.assertNotEqual(leaked.parent, Path(tempfile.gettempdir())) self.assertFalse(leaked.exists()) + def test_a_cleanup_error_keeps_the_named_and_trust_verdicts(self): + left: list[tempfile.TemporaryDirectory] = [] + real_cleanup = tempfile.TemporaryDirectory.cleanup + + def broken_cleanup(self): + left.append(self) + raise PermissionError('locked file') + + # The real cleanup runs afterwards, so the test leaves no run folders behind. + self.addCleanup(lambda: [real_cleanup(directory) for directory in left]) + with fixture() as root, patch.object(tempfile.TemporaryDirectory, 'cleanup', broken_cleanup): + named = python_test_runner.run_named_tests(root, ['tests.test_sample'], budget=30) + self.assertEqual(named.returncode, 0, named.stderr) + self.assertIn('PermissionError: locked file', named.cleanup_error) + (root / 'trust-ci' / 'tests').mkdir(parents=True) + (root / 'trust-ci' / 'tests' / 'test_trust.py').write_text( + 'import unittest\nclass Trust(unittest.TestCase):\n def test_ok(self):\n pass\n') + trust = python_test_runner.run_trust_tests(root, 0) + self.assertEqual(trust.returncode, 0, trust.stderr) + self.assertIn('PermissionError: locked file', trust.cleanup_error) + def test_existing_cli_requires_explicit_observation_mode_and_no_record(self): script = Path(__file__).resolve().parents[1] / 'scripts/getzilla_verify.py' for arguments in ( diff --git a/tests/test_route_metrics.py b/tests/test_route_metrics.py index b306da2f..37f7e61d 100644 --- a/tests/test_route_metrics.py +++ b/tests/test_route_metrics.py @@ -243,6 +243,21 @@ def test_report_cli_bounds_the_route_gap_list(self) -> None: self.assertEqual(sum(line.startswith(' r') for line in text.splitlines()), 10) self.assertIn('2 more; use --json for all.', text) + def test_report_cli_never_prints_terminal_control_characters_from_the_journal(self) -> None: + stamp = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ') + (self.root / '.getzilla' / 'runtime').mkdir() + hostile = 'r1\x1b]0;owned\x07‮' + (self.root / METRICS).write_text('\n'.join(json.dumps(item) for item in [ + {'at': stamp, 'event': 'route', 'route_id': hostile}, + {'at': stamp, 'event': 'verify', 'route_id': hostile, 'status': 'pass', 'profile': 'std\x1b[2J'}, + ]) + '\n', encoding='utf-8') + text = subprocess.run([sys.executable, str(ROOT / 'scripts' / 'getzilla_report.py'), '--since', '1d'], + cwd=self.root, capture_output=True, text=True, check=True).stdout + self.assertNotIn('\x1b', text) + self.assertNotIn('\x07', text) + self.assertNotIn('‮', text) + self.assertIn('r1?]0;owned??:', text) + if __name__ == '__main__': unittest.main()