yingqi-z20-Agent-libOS

#1

from __future__ import annotations

#2

#3

import tempfile

#4

import unittest

#5

from pathlib import Path

#6

#7

from agent_libos import Runtime

#8

from agent_libos.models import CapabilityRight

#9

from agent_libos.substrate import CommandResult, LocalResourceProviderSubstrate

#10

#11

#12

class RecordingShellProvider:

#13

def __init__(self) -> None:

#14

self.calls: list[tuple[list[str], str | None]] = []

#15

#16

def run(self, argv: list[str], *, timeout: float = 30.0, cwd: str | None = None) -> CommandResult:

#17

self.calls.append((list(argv), cwd))

#18

return CommandResult(argv=list(argv), returncode=0, stdout="ok", stderr="")

#19

#20

#21

class ProcessWorkingDirectoryTests(unittest.TestCase):

#22

def test_filesystem_tools_resolve_paths_from_process_working_directory(self) -> None:

#23

with tempfile.TemporaryDirectory() as temp_dir:

#24

root = Path(temp_dir)

#25

(root / "pkg").mkdir()

#26

(root / "pkg" / "module.py").write_text("print('pkg')\n", encoding="utf-8")

#27

runtime = Runtime.open("local", substrate=LocalResourceProviderSubstrate(root))

#28

try:

#29

pid = runtime.process.spawn(image="review-agent:v0", goal="read from cwd")

#30

runtime.filesystem.grant_directory(pid, "pkg", [CapabilityRight.READ, CapabilityRight.WRITE], issued_by="test")

#31

#32

changed = runtime.tools.call(pid, "set_working_directory", {"path": "pkg"})

#33

read = runtime.tools.call(pid, "read_text_file", {"path": "module.py"})

#34

written = runtime.tools.call(pid, "write_text_file", {"path": "created.txt", "content": "ok"})

#35

#36

self.assertTrue(changed.ok, changed.error)

#37

self.assertEqual(changed.payload["working_directory"], "pkg")

#38

self.assertTrue(read.ok, read.error)

#39

self.assertEqual(read.payload["path"], "pkg/module.py")

#40

self.assertTrue(written.ok, written.error)

#41

self.assertTrue((root / "pkg" / "created.txt").exists())

finally:

runtime.close()

def test_children_inherit_parent_working_directory_by_default(self) -> None:

#46

with tempfile.TemporaryDirectory() as temp_dir:

#47

root = Path(temp_dir)

#48

(root / "child-cwd").mkdir()

#49

runtime = Runtime.open("local", substrate=LocalResourceProviderSubstrate(root))

#50

try:

#51

parent = runtime.process.spawn(image="review-agent:v0", goal="spawn child")

#52

self.assertTrue(runtime.tools.call(parent, "set_working_directory", {"path": "child-cwd"}).ok)

#53

#54

spawned = runtime.tools.call(parent, "spawn_child_process", {"goal": "inherit cwd"})

#55

forked = runtime.tools.call(parent, "fork_child_process", {"goal": "inherit cwd"})

#56

#57

self.assertTrue(spawned.ok, spawned.error)

#58

self.assertTrue(forked.ok, forked.error)

#59

self.assertEqual(runtime.process.get(spawned.payload["child_pid"]).working_directory, "child-cwd")

#60

self.assertEqual(runtime.process.get(forked.payload["child_pid"]).working_directory, "child-cwd")

finally:

runtime.close()

def test_process_working_directory_persists_in_sqlite(self) -> None:

#65

with tempfile.TemporaryDirectory() as temp_dir:

#66

root = Path(temp_dir)

#67

(root / "persisted").mkdir()

#68

db_path = root / "runtime.sqlite"

#69

runtime = Runtime.open(db_path, substrate=LocalResourceProviderSubstrate(root))

#70

try:

#71

pid = runtime.process.spawn(image="review-agent:v0", goal="persist cwd")

#72

self.assertTrue(runtime.tools.call(pid, "set_working_directory", {"path": "persisted"}).ok)

finally:

runtime.close()

reopened = Runtime.open(db_path, substrate=LocalResourceProviderSubstrate(root))

#77

try:

#78

self.assertEqual(reopened.process.get(pid).working_directory, "persisted")

finally:

reopened.close()

def test_shell_tool_runs_from_process_working_directory(self) -> None:

#83

with tempfile.TemporaryDirectory() as temp_dir:

#84

root = Path(temp_dir)

#85

(root / "commands").mkdir()

#86

shell = RecordingShellProvider()

#87

substrate = LocalResourceProviderSubstrate(root)

#88

substrate.shell = shell

#89

runtime = Runtime.open("local", substrate=substrate)

#90

try:

#91

pid = runtime.process.spawn(image="review-agent:v0", goal="run from cwd")

#92

runtime.shell.grant_policy(pid, "always_allow", issued_by="test")

#93

#94

self.assertTrue(runtime.tools.call(pid, "set_working_directory", {"path": "commands"}).ok)

#95

result = runtime.tools.call(pid, "run_shell_command", {"argv": ["echo", "hello"]})

#96

#97

self.assertTrue(result.ok, result.error)

#98

self.assertEqual(shell.calls, [(["echo", "hello"], "commands")])

finally:

runtime.close()

if __name__ == "__main__":

#104

unittest.main()

#105

z6MkqRzA/yingqi-z20-Agent-libOS