Download tests/test_loop.py from hkblockchaintechnology/tiny-computer-agent: direct link, hf CLI and curl.
- Browser
- Download file 6.5 kB
-
https://huggingface.co/hkblockchaintechnology/tiny-computer-agent/resolve/main/tests/test_loop.py
- Command line
-
hf download hf://hkblockchaintechnology/tiny-computer-agent/tests/test_loop.py
-
curl -L -o test_loop.py https://huggingface.co/hkblockchaintechnology/tiny-computer-agent/resolve/main/tests/test_loop.py
6.5 kB
| import pytest | |
| from tiny_computer_agent import ( | |
| Action, AgentLoop, DryRunExecutor, Element, Episode, FakeScreen, | |
| GoalUnreachable, MockBrain, PlannerBrain, Policy, | |
| ) | |
| from tiny_computer_agent.__main__ import demo_screen | |
| def test_scripted_episode_succeeds_and_records_every_step(): | |
| screen = FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]) | |
| brain = MockBrain([ | |
| {"kind": "type", "text": "report"}, | |
| {"kind": "key", "keys": "ctrl+s"}, | |
| {"kind": "done", "reason": "file saved"}, | |
| ]) | |
| episode = AgentLoop(screen=screen, brain=brain).run("save the report") | |
| assert episode.success | |
| assert [s.action.kind for s in episode.steps] == ["type", "key", "done"] | |
| assert "file saved" in episode.stop_reason | |
| assert len(episode.to_rows()) == 3 | |
| def test_planner_solves_a_real_click_task_with_no_model(): | |
| episode = AgentLoop(screen=demo_screen(), brain=PlannerBrain()).run( | |
| "click 'Save' to save the document" | |
| ) | |
| assert episode.success | |
| clicks = [s.action for s in episode.steps if s.action.kind == "click"] | |
| assert [c.label for c in clicks] == ["Save"] | |
| def test_screen_is_reread_after_every_executed_action(): | |
| """The verify half of the loop: never decide twice on the same frame.""" | |
| screen = FakeScreen([(Element("Save"),)]) | |
| brain = MockBrain([{"kind": "click", "label": "Save"}, {"kind": "click", "label": "Save"}]) | |
| episode = AgentLoop(screen=screen, brain=brain).run("click Save") | |
| executed = [s for s in episode.steps if s.result is not None and s.action.kind != "done"] | |
| assert len(executed) == 2 | |
| assert screen.captures == len(executed) + 1 # one initial read, one per action | |
| assert episode.steps[0].observation.index == 0 | |
| assert episode.steps[1].observation.index == 1 | |
| def test_step_budget_is_enforced(): | |
| screen = FakeScreen([(Element("Save"),)]) | |
| # distinct actions, so the budget is what stops it rather than the no-progress guard | |
| brain = MockBrain([{"kind": "wait", "seconds": 0.5 + i} for i in range(10)]) | |
| episode = AgentLoop( | |
| screen=screen, brain=brain, policy=Policy(step_budget=3) | |
| ).run("click Save") | |
| assert episode.success is False | |
| assert len(episode.steps) == 3 | |
| assert "budget of 3" in episode.stop_reason | |
| def test_a_flailing_agent_is_stopped_instead_of_burning_the_budget(): | |
| """Repeating one action against an unchanged screen is stuck, not persistent.""" | |
| brain = MockBrain([{"kind": "click", "label": "Save"}] * 10) | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save"),)]), brain=brain, | |
| policy=Policy(step_budget=10), | |
| ).run("click Save") | |
| assert episode.success is False | |
| assert "no progress" in episode.stop_reason | |
| assert len(episode.steps) < 10 | |
| assert "unchanged screen" in episode.steps[-1].error | |
| def test_progress_resets_the_repeat_counter(): | |
| """Two different actions on an unchanged screen is not flailing.""" | |
| brain = MockBrain([ | |
| {"kind": "wait", "seconds": 0.5}, | |
| {"kind": "click", "label": "Save"}, | |
| {"kind": "wait", "seconds": 0.5}, | |
| {"kind": "click", "label": "Save"}, | |
| {"kind": "done", "reason": "assumed saved"}, | |
| ]) | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save"),)]), brain=brain, | |
| policy=Policy(step_budget=10), | |
| ).run("save") | |
| assert episode.success | |
| assert len(episode.steps) == 5 | |
| def test_a_dead_screen_is_reported_as_an_episode_not_a_traceback(): | |
| class Blind: | |
| name = "blind" | |
| def capture(self): | |
| raise RuntimeError("X connection failed") | |
| episode = AgentLoop(screen=Blind(), brain=MockBrain([{"kind": "done"}])).run("anything") | |
| assert episode.success is False | |
| assert episode.stop_reason.startswith("capture failed: RuntimeError") | |
| assert episode.steps == [] | |
| def test_policy_refusal_stops_the_run_and_nothing_executes(): | |
| executor = DryRunExecutor() | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Terminal"),)]), | |
| brain=MockBrain([{"kind": "type", "text": "sudo rm -rf /"}]), | |
| executor=executor, | |
| ).run("clean up") | |
| assert episode.success is False | |
| assert "policy refusal" in episode.stop_reason | |
| assert episode.steps[-1].error.startswith("refused:") | |
| assert executor.performed == [] | |
| def test_kill_switch_aborts_before_the_first_action(tmp_path): | |
| marker = tmp_path / "STOP" | |
| marker.write_text("stop") | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save"),)]), | |
| brain=MockBrain([{"kind": "click", "label": "Save"}]), | |
| policy=Policy(kill_switch_path=str(marker)), | |
| ).run("click Save") | |
| assert episode.steps == [] | |
| assert episode.stop_reason == "kill switch engaged" | |
| def test_a_broken_brain_stops_the_run_instead_of_crashing_it(): | |
| class Exploding: | |
| name = "exploding" | |
| def decide(self, goal, observation): | |
| raise RuntimeError("model died") | |
| episode = AgentLoop(screen=FakeScreen([(Element("Save"),)]), brain=Exploding()).run("save") | |
| assert episode.success is False | |
| assert episode.stop_reason == "brain error" | |
| assert "RuntimeError: model died" in episode.steps[-1].error | |
| def test_failed_action_ends_the_episode(): | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save"),)]), | |
| brain=MockBrain([{"kind": "click", "label": "Nowhere"}]), | |
| ).run("click Nowhere") | |
| assert episode.success is False | |
| assert "action failed" in episode.stop_reason | |
| def test_on_step_callback_sees_every_step_and_render_mentions_the_goal(): | |
| seen = [] | |
| episode = AgentLoop( | |
| screen=demo_screen(), brain=PlannerBrain(), on_step=seen.append | |
| ).run("press Save") | |
| assert len(seen) == len(episode.steps) | |
| assert "press Save" in episode.render() | |
| assert "success" in episode.render() | |
| def test_a_brain_that_gives_up_is_a_failure_not_a_success(): | |
| """The harness owns success: "I give up" must never read as "goal met".""" | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Files"),)]), | |
| brain=PlannerBrain(max_waits=0), | |
| ).run("click 'Save'") | |
| assert episode.success is False | |
| assert episode.stop_reason.startswith("brain gave up:") | |
| assert episode.steps[-1].error.startswith("gave up:") | |
| def test_planner_raises_rather_than_inventing_a_target(): | |
| with pytest.raises(GoalUnreachable, match="no element name"): | |
| PlannerBrain().decide("do the thing", FakeScreen([(Element("Save"),)]).capture()) | |