repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
Mirrored from https://github.com/yingqi-z20/Agent-libOS
stars
latest
clone command
git clone gitlawb://did:key:z6MkqRzA...RfoM/yingqi-z20-Agen...git clone gitlawb://did:key:z6MkqRzA.../yingqi-z20-Agen...d98dd2c9IPC1d ago| #1 | from __future__ import annotations |
| #2 | |
| #3 | import unittest |
| #4 | from uuid import uuid4 |
| #5 | |
| #6 | from agent_libos import Runtime |
| #7 | from agent_libos.api.cli import DEMO_PATCH_PREVIEW_CONTENT, DEMO_PATCH_PREVIEW_PATH, run_demo |
| #8 | |
| #9 | |
| #10 | class DemoContractTests(unittest.TestCase): |
| #11 | def setUp(self) -> None: |
| #12 | self.runtime = Runtime.open("local") |
| #13 | |
| #14 | def tearDown(self) -> None: |
| #15 | self.runtime.close() |
| #16 | |
| #17 | def test_run_demo_returns_auditable_contract(self) -> None: |
| #18 | result = run_demo(self.runtime) |
| #19 | |
| #20 | self.assertTrue(result["root"].startswith("pid_")) |
| #21 | self.assertTrue(result["worker"].startswith("pid_")) |
| #22 | self.assertTrue(result["checkpoint"].startswith("ckpt_")) |
| #23 | self.assertTrue(result["final_report_oid"].startswith("obj_")) |
| #24 | self.assertIsNotNone(result["approval_request"]) |
| #25 | self.assertGreater(result["audit_records"], 0) |
| #26 | |
| #27 | self.assertFalse(result["filesystem_write_denial"]["ok"]) |
| #28 | self.assertIn("lacks write", result["filesystem_write_denial"]["error"]) |
| #29 | self.assertTrue(result["write_result"]["ok"]) |
| #30 | self.assertEqual(result["write_result"]["payload"]["path"], DEMO_PATCH_PREVIEW_PATH) |
| #31 | self.assertTrue(result["target_file_exists"]) |
| #32 | self.assertTrue(result["target_file_content_matches"]) |
| #33 | |
| #34 | target = self.runtime.workspace_root / DEMO_PATCH_PREVIEW_PATH |
| #35 | self.assertEqual(target.read_text(encoding="utf-8"), DEMO_PATCH_PREVIEW_CONTENT) |
| #36 | |
| #37 | tool_names = [entry["tool"] for entry in result["tool_sequence"]] |
| #38 | self.assertIn("parse_pytest_log", tool_names) |
| #39 | if result["jit_validation_ok"]: |
| #40 | self.assertIn("extract_failed_tests", tool_names) |
| #41 | else: |
| #42 | self.assertTrue(result["jit_validation_errors"]) |
| #43 | self.assertGreaterEqual(tool_names.count("write_text_file"), 2) |
| #44 | |
| #45 | report = self.runtime.store.get_object(result["final_report_oid"]) |
| #46 | self.assertIsNotNone(report) |
| #47 | assert report is not None |
| #48 | payload = report.payload |
| #49 | self.assertEqual(payload["problem"]["failed_test"], "tests/test_math.py::test_add") |
| #50 | self.assertEqual(payload["authorization"]["filesystem_write_approval_request"], result["approval_request"]) |
| #51 | self.assertFalse(payload["authorization"]["filesystem_write_denied_before_grant"]["ok"]) |
| #52 | self.assertEqual(payload["external_side_effects"][0]["path"], DEMO_PATCH_PREVIEW_PATH) |
| #53 | self.assertTrue(payload["target_file"]["content_matches"]) |
| #54 | self.assertIn("not a production automatic repair system", payload["limits"]) |
| #55 | |
| #56 | audit_actions = [record.action for record in self.runtime.audit.trace()] |
| #57 | for action in [ |
| #58 | "checkpoint.create", |
| #59 | "human.query", |
| #60 | "human.response", |
| #61 | "primitive.filesystem.write_text", |
| #62 | "tool.call", |
| #63 | "process.exit", |
| #64 | ]: |
| #65 | self.assertIn(action, audit_actions) |
| #66 | |
| #67 | event_types = [event.type.value for event in self.runtime.events.list()] |
| #68 | self.assertIn("external_write", event_types) |
| #69 | self.assertIn("human_query", event_types) |
| #70 | self.assertIn("human_response", event_types) |
| #71 | |
| #72 | def test_tool_outside_process_tool_table_is_denied_without_human_approval(self) -> None: |
| #73 | pid = self.runtime.process.spawn(image="toolmaker-agent:v0", goal="write a demo file") |
| #74 | path = f"agent_outputs/demo_missing_tool_{uuid4().hex}.txt" |
| #75 | |
| #76 | result = self.runtime.tools.call(pid, "write_text_file", {"path": path, "content": "denied"}) |
| #77 | |
| #78 | self.assertFalse(result.ok) |
| #79 | self.assertIn("not in process tool table", result.error or "") |
| #80 | self.assertFalse((self.runtime.workspace_root / path).exists()) |
| #81 | self.assertNotIn("human.query", [record.action for record in self.runtime.audit.trace()]) |
| #82 | |
| #83 | |
| #84 | if __name__ == "__main__": |
| #85 | unittest.main() |
| #86 |