"""Offline language, resource-boundary and agent continuation tests.""" from dataclasses import replace import json from pathlib import Path import tempfile import unittest from unittest.mock import Mock, patch import aegis from astra import Agent CASES = { "greeting.aeg": (["Astra"], "ok", "Hello, Astra\n"), "arithmetic.aeg": ([], "ok", "28.0\n512\n4\n3.5\n0.25\n980.0\n"), "branching.aeg": ([], "ok", "allow\n"), "two_inputs.aeg": (["Astra", "yes"], "ok", "Supply a name\nSupply yes to approve\nApproved: Astra\n"), "short_circuit.aeg": ([], "ok", "false\ntrue\nshield: \u2694 # literal\ntrue\n"), "reject_host_access.aeg": ([], "error", "syntax_error"), "reject_type.aeg": ([], "error", "type_error"), "reject_resource.aeg": ([], "error", "limit_error"), } class LanguageTests(unittest.TestCase): def check_error(self, source, kind, **limits): result = aegis.run(source, limits=replace(aegis.Limits(), **limits)) self.assertEqual(result["status"], "error", result) self.assertEqual(result["error"]["kind"], kind, result) self.assertEqual(result["output"], "") self.assertGreaterEqual(result["error"]["line"], 1) self.assertGreaterEqual(result["error"]["column"], 1) return result def test_every_example_file(self): directory = Path(__file__).parent / "aegis_test" self.assertEqual({p.name for p in directory.glob("*.aeg")}, set(CASES)) for name, (inputs, status, expected) in CASES.items(): with self.subTest(script=name): result = aegis.run((directory / name).read_bytes(), inputs) self.assertEqual(result["status"], status, result) self.assertEqual(result["output"] if status == "ok" else result["error"]["kind"], expected) def test_numeric_and_boolean_operators(self): source = '''print(10 - 2 - 3) print(2 + 3 * 4) print((2 + 3) * 4) print(2 ^ (3 ^ 2)) print(1 < 2) print(2 <= 2) print(3 > 2) print(3 >= 3) print(1 == 1) print(1 == 1.0) print(true == 1) print("a" != 1) print(not false and true or false) print("a" < "b") print(-(-2)) print(0 ^ 0) ''' self.assertEqual(aegis.run(source)["output"], "5\n14\n20\n512\ntrue\ntrue\ntrue\ntrue\ntrue\nfalse\nfalse\ntrue\ntrue\ntrue\n2\n1\n") def test_static_validation_before_any_effect(self): for source, kind in [ ('print(input())\nif false:\n print(missing)', "name_error"), ('print("hidden")\nprint(true + 1)', "type_error"), ('print(false and 1)', "type_error"), ('print(true or missing)', "name_error"), ('if 1:\n print(1)', "type_error"), ('if true:\n a = 1\nprint(a)', "name_error"), ('if true:\n a = 1\nelse:\n a = "s"\nprint(a + 1)', "type_error"), ('print("a" < 1)', "type_error"), ('print(-true)', "type_error"), ('print(not 1)', "type_error"), ('print(input() + 1)', "type_error"), ]: with self.subTest(source=source): self.check_error(source, kind) def test_branch_merges_and_reassignment(self): source = 'x = 1\nif true:\n x = "yes"\nelse:\n x = "no"\nprint(x)\nx = false\nprint(x)' self.assertEqual(aegis.run(source)["output"], "yes\nfalse\n") self.assertEqual(aegis.run('if false:\n print(input())\nprint("ok")')["output"], "ok\n") def test_lexical_rejections(self): for source in ['print(1); print(2)', 'Print(1)', 'x = 1abc', 'x = 1.2.3', 'print(1 < 2 < 3)', 'print = 1', 'input = 1', 'x = input(1)', 'print("unterminated)', 'print("\\q")', 'x = --1', 'x = .5', 'print(1)\rprint(2)', 'if true:\n # empty', 'else:\n print(1)', 'x = "a"\nprint(x[0])', 'print(__import__("os"))']: with self.subTest(source=source): result = aegis.run(source) self.assertEqual(result["status"], "error", result) self.assertIn(result["error"]["kind"], ("syntax_error", "indentation_error")) for source in ['if true:\n\tprint(1)', 'if true:\n print(1)', 'if true:\n if true:\n print(1)', ' print(1)', 'if true:\n print(1)\n print(2)']: self.check_error(source, "indentation_error") self.check_error(b'print("\xff")', "syntax_error") self.check_error('print("\\ud800")', "value_error") def test_resource_limits(self): for source, limits in [ ('print(1)', dict(source_bytes=4)), ('print(1)', dict(line_length=4)), ('abc = 1', dict(identifier_length=2)), ('print(1)', dict(tokens=2)), ('print(1)', dict(steps=2)), ('print(1)', dict(memory_bytes=16)), ('print("abcd")', dict(string_bytes=3)), ('print("abcd")', dict(output_bytes=4)), ('print(256)', dict(integer_bits=8)), ('print(2 ^ 4096)', {}), ('print(2 ^ (2 ^ 100))', {}), ('print("a" + "b")', dict(string_bytes=1)), ('print(1)', dict(seconds=0)), ('print(' + '(' * 40 + '1' + ')' * 40 + ')', {}), ('print(' + '+'.join(['1'] * 40) + ')', {}), ]: with self.subTest(source=source, limits=limits): self.check_error(source, "limit_error", **limits) def test_value_errors(self): for source in ['print(1 / 0)', 'print(0 ^ -1)', 'print(1e999)', 'print(1e308 * 1e308)', 'print((-1) ^ 0.5)', 'print((2 ^ 2000) + 1.0)']: self.check_error(source, "value_error") def test_resume_exactly_once_and_active_time(self): session = aegis.Session('print("start")\na = input()\nb = input()\nprint(a + b)') first = session.advance() self.assertEqual(first["request"], dict(index=1, line=2, column=5)) self.assertEqual(first["output"], "start\n") # Waiting on a model/user is not interpreter execution time. with patch("aegis.time.monotonic", return_value=10**12): second = session.advance("a") self.assertEqual(second["request"]["index"], 2) self.assertEqual(session.advance("b")["output"], "start\nab\n") self.assertEqual(session.advance("c")["status"], "error") def test_input_limits_and_fresh_environment(self): for inputs, limits, kind in [(["abcd"], dict(input_bytes=3), "limit_error"), ([True], {}, "type_error"), (["a", "b"], dict(input_requests=1), "limit_error")]: result = aegis.run('a = input()\nprint(input())', inputs, replace(aegis.Limits(), **limits)) self.assertEqual(result["error"]["kind"], kind) self.assertEqual(aegis.run('x = 1')["status"], "ok") self.check_error('print(x)', "name_error") def test_safe_host_failures(self): secret = 'C:\\private\\machine\\secret.txt terminal output secret' with patch("aegis.lex", side_effect=OSError(secret)): result = aegis.run("print(1)") self.assertNotIn(secret, json.dumps(result)) self.assertEqual(result["error"]["kind"], "host_error") session = aegis.Session("print(1)") with patch.object(session, "expression", side_effect=OSError(secret)): result = session.advance() self.assertEqual(result["error"]["kind"], "host_error") self.assertNotIn(secret, json.dumps(result)) class IntegrationTests(unittest.TestCase): def setUp(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.agent = Agent(Path(directory.name), dict(OPENAI_API_KEY="fake", AGENT_MAIL_API_KEY="fake", AGENT_MAIL_ADRESS="fake")) self.agent.mail = Mock() self.agent.mail.request.return_value = {"messages": []} self.agent.ai = Mock() def create(self, name, source): self.agent.files.execute("create_file", dict(filename=name, content=source)) @staticmethod def call(name, **args): return dict(output=[dict(type="function_call", name=name, arguments=json.dumps(args), call_id="call")]) def test_all_examples_through_virtual_tool(self): for name, (inputs, status, expected) in CASES.items(): virtual = "aegis_test/" + name self.create(virtual, (Path(__file__).parent / virtual).read_text(encoding="utf-8")) result = self.agent.dispatch("run_ag_file", dict(filename=virtual)) for value in inputs: self.assertEqual(result["status"], "input_required") result = self.agent.dispatch("provide_ag_input", dict(value=value)) self.assertEqual(result["status"], status, (name, result)) self.assertEqual(result["output"] if status == "ok" else result["error"]["kind"], expected) def test_docs_examples_execute_through_tools(self): # Documentation must teach runnable code and remain available without file reads. with patch.object(self.agent.files, "execute", side_effect=AssertionError("unexpected file access")): docs = self.agent.dispatch("aegis_docs", {}) self.assertEqual(docs["version"], "0.1") self.assertEqual(len(docs["examples"]), 6) for example in docs["examples"]: with self.subTest(filename=example["filename"]): self.create(example["filename"], example["content"]) result = self.agent.dispatch("run_ag_file", dict(filename=example["filename"])) for value in example["inputs"]: self.assertEqual(result["status"], "input_required", result) result = self.agent.dispatch("provide_ag_input", dict(value=value)) self.assertEqual(result["status"], "ok", result) self.assertEqual(result["output"], example["expected_output"]) def test_reuse_and_snapshot(self): self.create("hello.aeg", 'print(input())') self.assertEqual(self.agent.dispatch("run_ag_file", dict(filename="hello.aeg"))["status"], "input_required") self.agent.files.execute("write_file", dict(filename="hello.aeg", content='print("new")')) self.assertEqual(self.agent.dispatch("run_ag_file", dict(filename="hello.aeg"))["status"], "error") self.assertEqual(self.agent.dispatch("provide_ag_input", dict(value="old"))["output"], "old\n") self.assertEqual(self.agent.dispatch("run_ag_file", dict(filename="hello.aeg"))["output"], "new\n") def test_real_files_unreachable_and_error_redaction(self): real = self.agent.root / "real.aeg" real.write_text('print("private")') for name in (str(real), "../real.aeg", "missing.aeg", "real.py"): result = self.agent.dispatch("run_ag_file", dict(filename=name)) self.assertEqual(result["status"], "error") self.assertNotIn(str(real), json.dumps(result)) self.assertNotIn("private", json.dumps(result)) with patch.object(self.agent.files, "execute", side_effect=OSError("C:\\private\\secret")): result = self.agent.dispatch("run_ag_file", dict(filename="x.aeg")) self.assertNotIn("private", json.dumps(result)) def test_fifth_response_input_gets_extra_response(self): self.create("x.aeg", "print(input())") self.agent.ai.request.side_effect = ([self.call("list_files")] * 4 + [self.call("run_ag_file", filename="x.aeg"), self.call("provide_ag_input", value="yes"), dict(output=[])]) self.agent.cycle() self.assertEqual(self.agent.ai.request.call_count, 7) self.assertIsNone(self.agent.aegis_session) def test_twenty_total_responses_cannot_be_reset(self): self.create("x.aeg", "print(input())") self.agent.ai.request.side_effect = [self.call("run_ag_file", filename="x.aeg"), self.call("provide_ag_input", value="yes")] * 11 self.agent.cycle() self.assertEqual(self.agent.ai.request.call_count, 20) self.agent.ai.request.reset_mock(side_effect=True) self.agent.ai.request.return_value = self.call("list_files") self.agent.cycle() self.assertEqual(self.agent.ai.request.call_count, 5) def test_pending_run_discarded_at_cap(self): self.create("x.aeg", "\n".join('a = input()' for _ in range(20))) self.agent.ai.request.side_effect = [self.call("run_ag_file", filename="x.aeg")] + [self.call("provide_ag_input", value="x")] * 20 self.agent.cycle() self.assertEqual(self.agent.ai.request.call_count, 20) self.assertIsNone(self.agent.aegis_session) self.assertEqual(self.agent.dispatch("provide_ag_input", dict(value="x"))["status"], "error") def test_noninteractive_script_keeps_five_cap(self): self.create("x.aeg", "print(1)") self.agent.ai.request.return_value = self.call("run_ag_file", filename="x.aeg") self.agent.cycle() self.assertEqual(self.agent.ai.request.call_count, 5) if __name__ == "__main__": unittest.main()