"""Tests for JARVIS memory system — task queue, conversations, memories.""" import json import os import sqlite3 import tempfile import unittest from unittest.mock import patch import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) class TestMemoryTaskQueue(unittest.TestCase): """Test deferred task queue in Memory.""" def setUp(self): self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) self.tmp.close() import memory self._orig_path = memory.DB_PATH memory.DB_PATH = self.tmp.name self.mem = memory.Memory() def tearDown(self): import memory memory.DB_PATH = self._orig_path os.unlink(self.tmp.name) def test_add_task(self): task_id = self.mem.add_task("Fix bug in server.py", "Handle 404 errors", "code", 2) self.assertIsInstance(task_id, int) self.assertGreater(task_id, 0) def test_get_pending_tasks(self): self.mem.add_task("Task A", priority=0) self.mem.add_task("Task B", priority=2) self.mem.add_task("Task C", priority=1) tasks = self.mem.get_pending_tasks() self.assertEqual(len(tasks), 3) # Should be ordered by priority DESC self.assertEqual(tasks[0]["title"], "Task B") self.assertEqual(tasks[1]["title"], "Task C") self.assertEqual(tasks[2]["title"], "Task A") def test_get_task(self): task_id = self.mem.add_task("Test task", "Description here", "general", 1) task = self.mem.get_task(task_id) self.assertIsNotNone(task) self.assertEqual(task["title"], "Test task") self.assertEqual(task["description"], "Description here") self.assertEqual(task["task_type"], "general") self.assertEqual(task["priority"], 1) self.assertEqual(task["status"], "pending") def test_get_nonexistent_task(self): task = self.mem.get_task(9999) self.assertIsNone(task) def test_update_task_status(self): task_id = self.mem.add_task("Status test") self.mem.update_task_status(task_id, "in_progress") task = self.mem.get_task(task_id) self.assertEqual(task["status"], "in_progress") self.mem.update_task_status(task_id, "completed") task = self.mem.get_task(task_id) self.assertEqual(task["status"], "completed") self.assertIsNotNone(task["completed_at"]) def test_delete_task(self): task_id = self.mem.add_task("Delete me") self.mem.delete_task(task_id) task = self.mem.get_task(task_id) self.assertIsNone(task) def test_get_all_tasks_excludes_completed(self): id1 = self.mem.add_task("Active task") id2 = self.mem.add_task("Done task") self.mem.update_task_status(id2, "completed") tasks = self.mem.get_all_tasks(include_completed=False) self.assertEqual(len(tasks), 1) self.assertEqual(tasks[0]["title"], "Active task") def test_get_all_tasks_includes_completed(self): id1 = self.mem.add_task("Active task") id2 = self.mem.add_task("Done task") self.mem.update_task_status(id2, "completed") tasks = self.mem.get_all_tasks(include_completed=True) self.assertEqual(len(tasks), 2) def test_pending_tasks_summary(self): self.mem.add_task("Fix tests", "Unit tests failing", priority=2) self.mem.add_task("Update docs", priority=0) summary = self.mem.get_pending_tasks_summary() self.assertIn("2 pending task(s)", summary) self.assertIn("Fix tests", summary) self.assertIn("[HIGH]", summary) self.assertIn("Update docs", summary) def test_pending_tasks_summary_empty(self): summary = self.mem.get_pending_tasks_summary() self.assertEqual(summary, "") def test_task_metadata(self): meta = {"source": "copilot", "file": "server.py"} task_id = self.mem.add_task("Meta task", metadata=meta) task = self.mem.get_task(task_id) stored_meta = json.loads(task["metadata"]) self.assertEqual(stored_meta["source"], "copilot") self.assertEqual(stored_meta["file"], "server.py") class TestMemoryConversations(unittest.TestCase): """Test conversation and memory storage.""" def setUp(self): self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) self.tmp.close() import memory self._orig_path = memory.DB_PATH memory.DB_PATH = self.tmp.name self.mem = memory.Memory() def tearDown(self): import memory memory.DB_PATH = self._orig_path os.unlink(self.tmp.name) def test_save_and_search_memory(self): self.mem.save_memory("pref", "language", "Python") results = self.mem.search_memories("Python") self.assertEqual(len(results), 1) self.assertEqual(results[0]["value"], "Python") def test_conversation_flow(self): conv_id = self.mem.create_conversation("Test Chat") self.mem.add_message(conv_id, "user", "Hello") self.mem.add_message(conv_id, "assistant", "Hi there!") messages = self.mem.get_messages(conv_id) self.assertEqual(len(messages), 2) self.assertEqual(messages[0]["role"], "user") self.assertEqual(messages[1]["role"], "assistant") def test_context_summary(self): self.mem.save_memory("fact", "name", "Tony Stark") summary = self.mem.get_context_summary() self.assertIn("Tony Stark", summary) self.assertIn("[LONG-TERM MEMORY]", summary) class TestClapDetector(unittest.TestCase): """Test clap detector impulse classification.""" def test_classify_impulse_fallback(self): """Test that classify_impulse works even without numpy (returns 'clap').""" from clap_detector import ClapDetector detector = ClapDetector() # Create silence data data = b'\x00\x00' * 1024 # Should not crash result = detector._classify_impulse(data) self.assertIn(result, ["clap", "snap"]) def test_is_impulse_below_threshold(self): from clap_detector import ClapDetector detector = ClapDetector() detector.ambient_level = 100 # Silence should not be an impulse data = b'\x00\x00' * 1024 self.assertFalse(detector._is_impulse(data)) def test_get_amplitude_silence(self): from clap_detector import ClapDetector detector = ClapDetector() data = b'\x00\x00' * 1024 amp = detector._get_amplitude(data) self.assertEqual(amp, 0.0) def test_suppress_unsuppress(self): from clap_detector import ClapDetector detector = ClapDetector() self.assertFalse(detector.suppressed) detector.suppress() self.assertTrue(detector.suppressed) class TestBuiltinTools(unittest.TestCase): """Test builtin tool functions.""" def test_add_task_tool(self): """Test the add_task tool function.""" import tempfile tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) tmp.close() import memory orig = memory.DB_PATH memory.DB_PATH = tmp.name try: mem = memory.Memory() from tools.builtin import add_task result = add_task("Test task", "Description", "code", 1) self.assertIn("Task #", result) self.assertIn("Test task", result) self.assertIn("medium", result) finally: memory.DB_PATH = orig os.unlink(tmp.name) def test_list_tasks_empty(self): import tempfile tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) tmp.close() import memory orig = memory.DB_PATH memory.DB_PATH = tmp.name try: mem = memory.Memory() from tools.builtin import list_tasks result = list_tasks() self.assertIn("No tasks", result) finally: memory.DB_PATH = orig os.unlink(tmp.name) def test_complete_task_tool(self): import tempfile tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) tmp.close() import memory orig = memory.DB_PATH memory.DB_PATH = tmp.name try: mem = memory.Memory() task_id = mem.add_task("Complete me") from tools.builtin import complete_task result = complete_task(task_id) self.assertIn("completed", result) self.assertIn("Complete me", result) finally: memory.DB_PATH = orig os.unlink(tmp.name) def test_complete_nonexistent_task(self): import tempfile tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) tmp.close() import memory orig = memory.DB_PATH memory.DB_PATH = tmp.name try: mem = memory.Memory() from tools.builtin import complete_task result = complete_task(9999) self.assertIn("not found", result) finally: memory.DB_PATH = orig os.unlink(tmp.name) class TestVSCodeTools(unittest.TestCase): """Test VS Code tool parameter validation (no actual execution).""" def test_vscode_tools_registered(self): from tools import TOOL_REGISTRY import tools.vscode_tools vscode_tools = [k for k in TOOL_REGISTRY if k.startswith("vscode_") or k.startswith("copilot_")] self.assertGreaterEqual(len(vscode_tools), 4) self.assertIn("vscode_open", TOOL_REGISTRY) self.assertIn("copilot_chat", TOOL_REGISTRY) self.assertIn("copilot_inline", TOOL_REGISTRY) def test_tool_registry_has_task_tools(self): from tools import TOOL_REGISTRY import tools.builtin self.assertIn("add_task", TOOL_REGISTRY) self.assertIn("list_tasks", TOOL_REGISTRY) self.assertIn("complete_task", TOOL_REGISTRY) self.assertIn("delete_task", TOOL_REGISTRY) self.assertIn("open_terminal", TOOL_REGISTRY) if __name__ == "__main__": unittest.main()