yingqi-z20-Agent-libOS

repository

loading code, commits, and activity

repositories

loading repo index

#1	from __future__ import annotations
#2
#3	import unittest
#4	from uuid import uuid4
#5
#6	from agent_libos import Runtime
#7
#8
#9	class GranularPermissionTests(unittest.TestCase):
#10	def setUp(self) -> None:
#11	self.runtime = Runtime.open("local")
#12
#13	def tearDown(self) -> None:
#14	self.runtime.close()
#15
#16	def test_filesystem_file_and_directory_allow_lists(self) -> None:
#17	allowed_file = self._write_fixture("allowed file")
#18	denied_file = self._write_fixture("denied file")
#19	allowed_dir = f"agent_outputs/granular_dir_{uuid4().hex}"
#20	allowed_dir_file = self._write_fixture("allowed directory file", f"{allowed_dir}/nested/readme.txt")
#21	exact_write = f"agent_outputs/granular_write_{uuid4().hex}/exact.txt"
#22	write_dir = f"agent_outputs/granular_write_dir_{uuid4().hex}"
#23	denied_write = f"agent_outputs/granular_denied_write_{uuid4().hex}.txt"
#24	pid = self.runtime.process.spawn(image="review-agent:v0", goal="granular filesystem")
#25
#26	self.runtime.filesystem.grant_path_list(
#27	pid,
#28	read_files=[allowed_file],
#29	read_dirs=[allowed_dir],
#30	write_files=[exact_write],
#31	write_dirs=[write_dir],
#32	issued_by="test",
#33	)
#34
#35	read_file = self.runtime.tools.call(pid, "read_text_file", {"path": allowed_file})
#36	read_dir_file = self.runtime.tools.call(pid, "read_text_file", {"path": allowed_dir_file})
#37	read_denied = self.runtime.tools.call(pid, "read_text_file", {"path": denied_file})
#38	write_exact = self.runtime.tools.call(pid, "write_text_file", {"path": exact_write, "content": "exact"})
#39	write_dir_file = self.runtime.tools.call(
#40	pid,
#41	"write_text_file",
#42	{"path": f"{write_dir}/nested/out.txt", "content": "dir"},
#43	)
#44	write_denied = self.runtime.tools.call(pid, "write_text_file", {"path": denied_write, "content": "no"})
#45
#46	self.assertTrue(read_file.ok, read_file.error)
#47	self.assertEqual(read_file.payload["content"], "allowed file")
#48	self.assertTrue(read_dir_file.ok, read_dir_file.error)
#49	self.assertEqual(read_dir_file.payload["content"], "allowed directory file")
#50	self.assertFalse(read_denied.ok)
#51	self.assertIn("lacks read", read_denied.error or "")
#52	self.assertTrue(write_exact.ok, write_exact.error)
#53	self.assertTrue(write_dir_file.ok, write_dir_file.error)
#54	self.assertFalse(write_denied.ok)
#55	self.assertIn("lacks write", write_denied.error or "")
#56
#57	def test_child_inherits_only_explicit_filesystem_subset(self) -> None:
#58	allowed_dir = f"agent_outputs/inherit_allowed_{uuid4().hex}"
#59	allowed_file = self._write_fixture("allowed child read", f"{allowed_dir}/data.txt")
#60	denied_file = self._write_fixture("denied child read")
#61	parent_write = f"agent_outputs/inherit_parent_write_{uuid4().hex}.txt"
#62	parent = self.runtime.process.spawn(image="review-agent:v0", goal="parent")
#63	self.runtime.filesystem.grant_path_list(
#64	parent,
#65	read_dirs=[allowed_dir],
#66	write_files=[parent_write],
#67	issued_by="test",
#68	)
#69
#70	forked = self.runtime.tools.call(
#71	parent,
#72	"fork_child_process",
#73	{"goal": "child", "inherit_read_dirs": [allowed_dir]},
#74	)
#75	child = forked.payload["child_pid"]
#76	child_read = self.runtime.tools.call(child, "read_text_file", {"path": allowed_file})
#77	child_read_denied = self.runtime.tools.call(child, "read_text_file", {"path": denied_file})
#78	child_write_denied = self.runtime.tools.call(
#79	child,
#80	"write_text_file",
#81	{"path": parent_write, "content": "child should not write"},
#82	)
#83	parent_write_allowed = self.runtime.tools.call(
#84	parent,
#85	"write_text_file",
#86	{"path": parent_write, "content": "parent can write"},
#87	)
#88
#89	self.assertTrue(forked.ok, forked.error)
#90	self.assertTrue(child_read.ok, child_read.error)
#91	self.assertFalse(child_read_denied.ok)
#92	self.assertIn("lacks read", child_read_denied.error or "")
#93	self.assertFalse(child_write_denied.ok)
#94	self.assertIn("lacks write", child_write_denied.error or "")
#95	self.assertTrue(parent_write_allowed.ok, parent_write_allowed.error)
#96
#97	def test_child_cannot_inherit_broader_permission_than_parent_has(self) -> None:
#98	allowed_file = self._write_fixture("one file")
#99	parent = self.runtime.process.spawn(image="review-agent:v0", goal="parent")
#100	self.runtime.filesystem.grant_path_list(parent, read_files=[allowed_file], issued_by="test")
#101	requested_dir = "/".join(allowed_file.split("/")[:-1])
#102
#103	forked = self.runtime.tools.call(
#104	parent,
#105	"fork_child_process",
#106	{"goal": "child", "inherit_read_dirs": [requested_dir]},
#107	)
#108	children = self.runtime.process.list_children(parent)
#109
#110	self.assertFalse(forked.ok)
#111	self.assertIn("cannot inherit", forked.error or "")
#112	self.assertEqual(children, [])
#113
#114	def _write_fixture(self, content: str, path: str \| None = None) -> str:
#115	relative = path or f"agent_outputs/granular_fixture_{uuid4().hex}.txt"
#116	target = self.runtime.workspace_root / relative
#117	target.parent.mkdir(parents=True, exist_ok=True)
#118	target.write_text(content, encoding="utf-8")
#119	return relative
#120
#121
#122	if __name__ == "__main__":
#123	unittest.main()
#124

z6MkqRzA/yingqi-z20-Agent-libOS