tests/test_memory_tasks.py

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