mirror of
https://github.com/ruvnet/RuView.git
synced 2026-10-21 22:22:33 -05:00
rUv
aac4155584
* feat(bfi): add validated capture and bounded research tooling * fix(deps): patch Wasmtime for RUSTSEC-2026-0316
77 lines
4.3 KiB
Python
77 lines
4.3 KiB
Python
import json
|
|
import io
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import codex_iteration as c
|
|
|
|
|
|
class ReviewBoundaryTests(unittest.TestCase):
|
|
def test_snapshot_allows_only_bfi_source_and_binds_hash(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td);(root/'.git').mkdir();(root/'AGENTS.md').write_text('contract')
|
|
tools=root/'tools/bfi';tools.mkdir(parents=True);(tools/'a.py').write_text('x=1')
|
|
self.assertEqual(c.snapshot(root,['tools/bfi/a.py'])[0]['content'],'x=1')
|
|
(root/'private.env').write_text('not for review')
|
|
with self.assertRaises(ValueError):c.snapshot(root,['private.env'])
|
|
with self.assertRaises(ValueError):c.snapshot(root,[])
|
|
def test_private_output_rejects_checkout(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td).resolve();(root/'.git').mkdir()
|
|
with self.assertRaises(ValueError):c.checked_output_root(root/'logs',root)
|
|
def test_child_policy_keeps_exec_rules_and_read_only(self):
|
|
cmd=c.command(Path('codex'),Path('repo'))
|
|
self.assertIn('read-only',cmd)
|
|
self.assertNotIn('--ignore-rules',cmd)
|
|
self.assertFalse(any('bypass' in a for a in cmd))
|
|
self.assertIn('shell_tool',cmd)
|
|
def test_review_rejects_tools_and_malformed_events(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
p=Path(td)/'events.jsonl'
|
|
p.write_text(json.dumps({'type':'item.completed','item':{'type':'agent_message','text':'review'}})+'\n'+json.dumps({'type':'turn.completed'})+'\n')
|
|
self.assertEqual(c.audit_events(p),(['review'],[]))
|
|
with p.open('a') as f:
|
|
f.write(json.dumps({'type':'item.started','item':{'type':'command_execution'}})+'\ninvalid\n')
|
|
findings,errors=c.audit_events(p)
|
|
self.assertEqual(findings,['review'])
|
|
self.assertEqual([e['type'] for e in errors],['prohibited_activity','invalid_json_event'])
|
|
p.write_text('{"type":null}\n')
|
|
self.assertEqual([e['type'] for e in c.audit_events(p)[1]],['invalid_event_type','missing_turn_completed'])
|
|
def test_source_bound_and_nonregular_file(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td);(root/'.git').mkdir();(root/'AGENTS.md').write_text('contract')
|
|
folder=root/'tools/bfi';folder.mkdir(parents=True)
|
|
(folder/'large.py').write_bytes(b'x'*(c.MAX_SOURCE+1))
|
|
with self.assertRaises(ValueError):c.snapshot(root,['tools/bfi/large.py'])
|
|
(folder/'directory.py').mkdir()
|
|
with self.assertRaises(ValueError):c.snapshot(root,['tools/bfi/directory.py'])
|
|
def test_output_failure_reaps_child(self):
|
|
class BrokenSink:
|
|
def write(self,data):raise OSError('simulated disk full')
|
|
original=c.subprocess.Popen;children=[]
|
|
def spawn(*a,**k):
|
|
child=original(*a,**k);children.append(child);return child
|
|
program="import sys,time;sys.stdout.buffer.write(b'x'*8192);sys.stdout.flush();time.sleep(20)"
|
|
with patch.object(c,'command',return_value=[sys.executable,'-c',program]), patch.object(c.subprocess,'Popen',side_effect=spawn):
|
|
with self.assertRaises(OSError):
|
|
c.run_with_sinks(Path(sys.executable),Path('.'),'input',5,{'stdout':BrokenSink(),'stderr':io.BytesIO()})
|
|
self.assertIsNotNone(children[0].poll())
|
|
def test_stdin_receipt_and_deadline(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
root=Path(td);ok=root/'ok';ok.mkdir()
|
|
program="import sys,json; x=sys.stdin.read(); print(json.dumps({'received':len(x)}))"
|
|
with patch.object(c,'command',return_value=[sys.executable,'-c',program]):
|
|
result=c.run(Path(sys.executable),root,ok,'abc',5)
|
|
self.assertEqual(result['exit_code'],0)
|
|
self.assertEqual(json.loads((ok/'events.jsonl').read_text())['received'],3)
|
|
slow=root/'slow';slow.mkdir()
|
|
with patch.object(c,'command',return_value=[sys.executable,'-c','import time;time.sleep(10)']):
|
|
result=c.run(Path(sys.executable),root,slow,'x'*240000,.25)
|
|
self.assertEqual(result['stop_reason'],'timeout')
|
|
self.assertLess(result['elapsed_seconds'],3)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|