| import json |
| from pathlib import Path |
| import tempfile |
| import time |
| import unittest |
| from unittest.mock import Mock, patch |
|
|
| from astra import Agent, Audit, MODEL, VirtualFiles |
|
|
|
|
| class AgentTests(unittest.TestCase): |
| def reload_agent(self): |
| return Agent(self.root, {"OPENAI_API_KEY": "test-openai", "AGENT_MAIL_API_KEY": "test-mail", "AGENT_MAIL_ADRESS": "astra@example.com"}) |
|
|
| def test_resume_preserves_history_and_suppresses_onboarding(self): |
| self.agent.state.update(cycle=22, next_run=12345) |
| self.agent.persist() |
| resumed = self.reload_agent() |
| resumed.prepare_launch(no_prompt=True) |
| self.assertEqual(resumed.state["next_run"], 12345) |
| resumed.mail = self.agent.mail |
| resumed.ai = Mock() |
| resumed.ai.request.return_value = {"output": []} |
| resumed.cycle() |
| context = json.loads(resumed.ai.request.call_args.args[1]["input"][0]["content"]) |
| self.assertEqual(context["cycle"], 23) |
| self.assertEqual(context["online_confirmations_remaining"], []) |
|
|
| def test_instruction_prompt_runs_once_and_wakes_immediately(self): |
| self.agent.state["cycle"] = 2 |
| self.agent.persist() |
| resumed = self.reload_agent() |
| with patch("astra.sys.stdin.isatty", return_value=True), patch("builtins.input", return_value="Review my project") as prompt: |
| resumed.prepare_launch() |
| prompt.assert_called_once() |
| self.assertEqual(resumed.state["next_run"], 0) |
| resumed.mail = self.agent.mail |
| resumed.ai = Mock() |
| resumed.ai.request.return_value = {"output": []} |
| resumed.cycle() |
| context = json.loads(resumed.ai.request.call_args.args[1]["input"][0]["content"]) |
| self.assertEqual(context["operator_instruction"], "Review my project") |
| self.assertNotIn("operator_instruction", self.reload_agent().state) |
|
|
| def test_instruction_survives_failed_wake(self): |
| self.agent.prepare_launch("Check notes") |
| self.agent.ai.request.side_effect = TimeoutError() |
| with self.assertRaises(TimeoutError): |
| self.agent.cycle() |
| self.assertEqual(self.reload_agent().state["operator_instruction"], "Check notes") |
|
|
| def test_clearing_one_file_does_not_reset_but_both_does(self): |
| self.agent.state["cycle"] = 22 |
| self.agent.persist() |
| self.agent.state_path.write_text("") |
| recovered = self.reload_agent() |
| self.assertEqual(recovered.state["cycle"], 22) |
| recovered.persist() |
| self.agent.audit.path.write_text("") |
| self.assertEqual(self.reload_agent().state["cycle"], 22) |
| self.agent.state_path.write_text("{}") |
| fresh = self.reload_agent() |
| self.assertFalse(fresh.resuming) |
| self.assertEqual(fresh.state["cycle"], 0) |
|
|
| def test_legacy_log_without_state_does_not_restart(self): |
| self.agent.audit.log("startup") |
| with self.assertRaisesRegex(ValueError, "BOTH"): |
| self.reload_agent() |
|
|
| def test_first_wake_has_confirmations_and_no_prompt(self): |
| with patch("builtins.input") as prompt: |
| self.agent.prepare_launch() |
| prompt.assert_not_called() |
| self.agent.ai.request.return_value = {"output": []} |
| self.agent.cycle() |
| context = json.loads(self.agent.ai.request.call_args.args[1]["input"][0]["content"]) |
| self.assertEqual(len(context["online_confirmations_remaining"]), 2) |
|
|
| def setUp(self): |
| self.temp = tempfile.TemporaryDirectory() |
| self.addCleanup(self.temp.cleanup) |
| self.root = Path(self.temp.name) |
| self.agent = Agent(self.root, {"OPENAI_API_KEY": "test-openai", "AGENT_MAIL_API_KEY": "test-mail", "AGENT_MAIL_ADRESS": "astra@example.com"}) |
| self.agent.mail = Mock() |
| self.agent.mail.request.return_value = {"messages": []} |
| self.agent.ai = Mock() |
|
|
| def test_virtual_paths_never_touch_real_files(self): |
| outside = self.root / "outside.txt" |
| outside.write_text("protected") |
| for name in (str(outside), "../../outside.txt", "C:\\Windows\\test.txt"): |
| self.agent.files.execute("create_file", {"filename": name, "content": "virtual"}) |
| self.assertEqual(outside.read_text(), "protected") |
| self.assertEqual(len(self.agent.files.execute("list_files", {})), 3) |
|
|
| def test_file_edits(self): |
| fs = self.agent.files |
| fs.execute("create_file", {"filename": "notes", "content": "one\ntwo\nthree\n"}) |
| fs.execute("rewrite_lines", {"filename": "notes", "start_line": 2, "end_line": 2, "content": "new"}) |
| self.assertEqual(fs.execute("read_file", {"filename": "notes"}), "one\nnew\nthree\n") |
| with self.assertRaises(ValueError): |
| fs.execute("rewrite_lines", {"filename": "notes", "start_line": 0, "end_line": 1, "content": "bad"}) |
| fs.execute("rename_file", {"filename": "notes", "newname": "next"}) |
| fs.execute("append_file", {"filename": "next", "content": "four"}) |
| fs.execute("delete_file", {"filename": "next"}) |
| self.assertEqual(fs.execute("list_files", {}), []) |
|
|
| def test_scheduling_bounds_and_persistence(self): |
| for value in (9, 61, True, 10.5): |
| self.assertIn("error", self.agent.dispatch("schedule_next_turn", {"minutes": value})) |
| for value in (10, 60): |
| before = time.time() |
| self.agent.dispatch("schedule_next_turn", {"minutes": value}) |
| saved = json.loads(self.agent.state_path.read_text()) |
| self.assertAlmostEqual(saved["next_run"], before + value * 60, delta=2) |
|
|
| def test_five_response_cap_and_default_schedule(self): |
| self.agent.ai.request.return_value = {"output": [{"type": "function_call", "name": "list_files", "arguments": "{}", "call_id": "call"}]} |
| self.agent.cycle() |
| self.assertEqual(self.agent.ai.request.call_count, 5) |
| self.assertAlmostEqual(self.agent.state["next_run"], time.time() + 1800, delta=2) |
| self.assertTrue(self.agent.mail.request.called) |
| self.assertTrue(all(c.args[1]["model"] == MODEL for c in self.agent.ai.request.call_args_list)) |
|
|
| def test_schedule_survives_cycle_end(self): |
| self.agent.ai.request.side_effect = [ |
| {"output": [{"type": "function_call", "name": "schedule_next_turn", "arguments": '{"minutes":10}', "call_id": "c"}]}, |
| {"output": []}] |
| self.agent.cycle() |
| self.assertAlmostEqual(self.agent.state["next_run"], time.time() + 600, delta=2) |
|
|
| def test_pagination_and_newest_first(self): |
| self.agent.mail.request.side_effect = [ |
| {"messages": [{"message_id": "new"}], "next_page_token": "page2"}, |
| {"message_id": "new", "timestamp": "2026-09-16T10:00:00Z", "text": "new body"}, |
| {"messages": [{"message_id": "old"}]}, |
| {"message_id": "old", "timestamp": "2026-09-15T10:00:00Z", "text": "old body"}] |
| result = self.agent.sync_inbox() |
| self.assertEqual([m["message_id"] for m in result], ["new", "old"]) |
| self.assertEqual(result[0]["text"], "new body") |
| self.assertIn("page_token=page2", self.agent.mail.request.call_args_list[2].args[0]) |
|
|
| def test_send_deduplication_and_uncertainty(self): |
| self.agent.mail.request.side_effect = TimeoutError("timeout") |
| args = {"to": "someone@example.com", "subject": "test", "body": "hello"} |
| self.assertIn("error", self.agent.dispatch("send_email", args)) |
| self.assertTrue(self.agent.dispatch("send_email", args)["duplicate_suppressed"]) |
| self.assertEqual(self.agent.mail.request.call_count, 1) |
| self.assertEqual(next(iter(self.agent.state["outbox"].values()))["status"], "delivery_uncertain") |
|
|
| def test_reply_tracking(self): |
| self.agent.state["messages"]["m1"] = {"message_id": "m1", "from": "person@example.com", "subject": "Hello"} |
| self.agent.mail.request.return_value = {"message_id": "reply1"} |
| result = self.agent.dispatch("reply_email", {"message_id": "m1", "body": "Hi"}) |
| self.assertEqual(result["status"], "sent") |
| self.assertEqual(self.agent.state["handled"]["m1"], "replied") |
| self.assertIn("error", self.agent.dispatch("reply_email", {"message_id": "m1", "body": "Hi again"})) |
|
|
| def test_audit_redacts_secrets(self): |
| self.agent.audit.log("test", detail="test-openai and test-mail") |
| content = (self.root / "astra_log.txt").read_text() |
| self.assertNotIn("test-openai", content) |
| self.assertNotIn("test-mail", content) |
| self.assertIn("timestamp", content) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
