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()