| import pytest |
|
|
| """A refusal from the operator is a step, not a traceback.""" |
|
|
| from tiny_computer_agent import ( |
| Action, AgentLoop, AllOf, AllowAll, AllowOnlyLabels, AnyOf, Approval, DenyAll, Element, |
| FakeScreen, InteractiveGate, MockBrain, Policy, gate_from_spec, mutating, |
| ) |
|
|
|
|
| def survey(): |
| answers = ["y", "n", "a", "v"] |
| asked = [] |
|
|
| def reader(question: str) -> str: |
| asked.append(question) |
| return answers.pop(0) if answers else "" |
|
|
| return reader, asked |
|
|
|
|
| def test_read_only_verbs_never_need_consent(): |
| assert not mutating(Action("wait", seconds=1)) |
| assert not mutating(Action("done", reason="finished")) |
| assert mutating(Action("click", label="Save")) |
| assert mutating(Action("type", text="hello")) |
|
|
|
|
| def test_deny_all_kills_a_live_loop_but_lets_it_wait(): |
| gate = DenyAll() |
| assert gate.approve(Action("wait", seconds=1)).approved |
| refused = gate.approve(Action("click", label="Save")) |
| assert refused.approved is False and "said no" in refused.reason |
|
|
|
|
| def test_the_whitelist_gate_allows_a_renamed_button_and_refuses_the_rest(): |
| gate = AllowOnlyLabels(["Save", "Cancel"]) |
|
|
| assert gate.approve(Action("click", label="Save")).approved |
| assert gate.approve(Action("click", label="Save.")).approved, "the index's comparison applies" |
| assert not gate.approve(Action("click", label="Terminal")).approved |
| assert gate.approve(Action("type", text="rm -rf /")).approved is False, "text is not whitelisted" |
|
|
|
|
| def test_the_interactive_gate_asks_and_remembers(): |
| reader, asked = survey() |
| gate = InteractiveGate(reader=reader, writer=lambda line: None) |
|
|
| assert gate.approve(Action("click", label="Save")).approved |
| assert not gate.approve(Action("click", label="Save")).approved |
| assert gate.approve(Action("click", label="Save")).approved |
| assert gate.approve(Action("click", label="Save")).approved |
| assert len(asked) == 3 |
| assert gate.approve(Action("wait", seconds=1)).approved, "and it is never asked about a wait" |
|
|
|
|
| def test_the_interactive_gate_remembers_a_never(): |
| reader, _ = survey() |
| gate = InteractiveGate(reader=reader, writer=lambda line: None) |
| gate.approve(Action("click", label="Save")) |
| gate.approve(Action("click", label="Save")) |
| gate.approve(Action("click", label="Save")) |
| assert gate.approve(Action("click", label="Save")).approved |
| gate._always = False |
| answers = ["v", ""] |
| gate.reader = lambda question: answers.pop(0) |
| assert not gate.approve(Action("click", label="Save")).approved |
| assert not gate.approve(Action("click", label="Save")).approved, "never sticks" |
|
|
|
|
| def test_an_absent_operator_is_a_refusal_not_a_hang(): |
| def no_one(question: str) -> str: |
| raise EOFError("no stdin") |
|
|
| assert not InteractiveGate(reader=no_one).approve(Action("click", label="Save")).approved |
|
|
|
|
| def test_all_of_is_strict_and_any_of_is_permissive(): |
| allowed = AllowOnlyLabels(["Save"]) |
| assert not AllOf([allowed, DenyAll()]).approve(Action("click", label="Save")).approved |
| assert AllOf([allowed, AllowAll()]).approve(Action("click", label="Save")).approved |
| assert AnyOf([DenyAll(), allowed]).approve(Action("click", label="Save")).approved |
| assert not AnyOf([DenyAll()]).approve(Action("click", label="Save")).approved |
| assert isinstance(AllowAll().approve(Action("click", label="Save")), Approval) |
|
|
|
|
| def test_a_gate_refusal_stops_the_episode_and_is_recorded(): |
| screen = FakeScreen([(Element("Save", x=10, y=10, width=60, height=20),)]) |
| brain = MockBrain([{"kind": "click", "label": "Save"}, {"kind": "done", "reason": "saved"}]) |
| loop = AgentLoop(screen=screen, brain=brain, policy=Policy(step_budget=4), |
| gate=AllowOnlyLabels(["Cancel"])) |
|
|
| episode = loop.run("click 'Save' to save the document") |
|
|
| assert episode.success is False |
| assert "operator refused" in episode.stop_reason |
| assert episode.steps[0].error is not None and episode.steps[0].error.startswith("not approved") |
| assert episode.steps[0].result is None, "nothing was executed" |
| assert len(episode.steps) == 1, "the episode ended there" |
|
|
|
|
| |
|
|
| def test_gate_from_spec_read_only_allows_a_read_and_refuses_a_change(): |
| gate = gate_from_spec("read-only") |
| assert gate.approve(Action("wait", seconds=1)).approved |
| refusal = gate.approve(Action("click", label="Save")) |
| assert refusal.approved is False |
| assert "read-only gate" in refusal.reason |
|
|
|
|
| def test_gate_from_spec_labels_normalise_what_the_operator_wrote(): |
| gate = gate_from_spec("labels: save , Cancel") |
| assert gate.approve(Action("click", label="Save")).approved |
| assert gate.approve(Action("key", keys="ctrl+s")).approved is False |
|
|
|
|
| def test_gate_from_spec_interactive_asks_the_operator(): |
| assert type(gate_from_spec("interactive")).__name__ == "InteractiveGate" |
|
|
|
|
| def test_gate_from_spec_allow_all_approves_whatever_policy_allowed(): |
| assert gate_from_spec("allow-all").approve(Action("click", label="Save")).approved |
|
|
|
|
| def test_an_unknown_spec_says_what_it_accepts(): |
| """A typo in a safety switch must never quietly mean "allow everything".""" |
| with pytest.raises(ValueError, match="allow-all"): |
| gate_from_spec("yes-please") |
|
|
|
|
| def test_labels_with_no_labels_is_a_configuration_error(): |
| with pytest.raises(ValueError, match="at least one label"): |
| gate_from_spec("labels: , ") |
|
|
