basicmachines-co-basic-memory/tests/cli/test_watch.py

"""Tests for the watch CLI command."""

import asyncio
import os
import signal
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
import typer

from basic_memory.cli.commands.watch import run_watch
from basic_memory.config import BasicMemoryConfig


@pytest.fixture
def mock_config():
    """Create a mock config for testing."""
    return BasicMemoryConfig()


@pytest.fixture
def mock_container(mock_config):
    """Create a mock CLI container."""
    container = MagicMock()
    container.config = mock_config
    return container


class TestRunWatch:
    """Tests for run_watch async function."""

    @pytest.mark.asyncio
    async def test_initializes_app(self, mock_container):
        """run_watch calls initialize_app with the container's config."""
        mock_init = AsyncMock()

        with (
            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
            patch("basic_memory.cli.commands.watch.initialize_app", mock_init),
            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
            patch("basic_memory.cli.commands.watch.db") as mock_db,
        ):
            # Make coordinator.start() set the shutdown event so we don't block
            mock_coordinator = AsyncMock()
            mock_coordinator_cls.return_value = mock_coordinator

            async def start_then_shutdown():
                # Simulate immediate shutdown after start
                pass

            mock_coordinator.start = start_then_shutdown
            mock_coordinator.stop = AsyncMock()
            mock_db.shutdown_db = AsyncMock()

            # Patch signal handlers and make shutdown_event fire immediately
            with patch("asyncio.get_running_loop") as mock_loop:
                mock_loop_instance = MagicMock()
                mock_loop.return_value = mock_loop_instance

                # Capture the signal handler so we can trigger it
                signal_handlers = {}

                def capture_handler(sig, handler):
                    signal_handlers[sig] = handler

                mock_loop_instance.add_signal_handler.side_effect = capture_handler

                # Run in a task so we can trigger shutdown
                async def run_and_shutdown():
                    task = asyncio.create_task(run_watch())
                    # Give it a moment to start
                    await asyncio.sleep(0.01)
                    # Trigger shutdown via captured signal handler
                    import signal

                    if signal.SIGINT in signal_handlers:
                        signal_handlers[signal.SIGINT]()
                    await task

                await run_and_shutdown()

            mock_init.assert_called_once_with(mock_container.config)

    @pytest.mark.asyncio
    async def test_creates_coordinator_with_quiet_false(self, mock_container):
        """SyncCoordinator is created with should_sync=True and quiet=False."""
        with (
            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
            patch("basic_memory.cli.commands.watch.db") as mock_db,
        ):
            mock_coordinator = AsyncMock()
            mock_coordinator_cls.return_value = mock_coordinator
            mock_db.shutdown_db = AsyncMock()

            with patch("asyncio.get_running_loop") as mock_loop:
                mock_loop_instance = MagicMock()
                mock_loop.return_value = mock_loop_instance

                signal_handlers = {}

                def capture_handler(sig, handler):
                    signal_handlers[sig] = handler

                mock_loop_instance.add_signal_handler.side_effect = capture_handler

                async def run_and_shutdown():
                    task = asyncio.create_task(run_watch())
                    await asyncio.sleep(0.01)
                    import signal

                    if signal.SIGINT in signal_handlers:
                        signal_handlers[signal.SIGINT]()
                    await task

                await run_and_shutdown()

            mock_coordinator_cls.assert_called_once_with(
                config=mock_container.config,
                should_sync=True,
                quiet=False,
            )

    @pytest.mark.asyncio
    async def test_project_sets_env_var(self, mock_container):
        """--project validates and sets BASIC_MEMORY_MCP_PROJECT env var."""
        mock_config_manager = MagicMock()
        mock_config_manager.get_project.return_value = ("my-project", "/some/path")

        with (
            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
            patch(
                "basic_memory.cli.commands.watch.ConfigManager",
                return_value=mock_config_manager,
            ),
            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
            patch("basic_memory.cli.commands.watch.db") as mock_db,
            patch.dict(os.environ, {}, clear=False),
        ):
            mock_coordinator = AsyncMock()
            mock_coordinator_cls.return_value = mock_coordinator
            mock_db.shutdown_db = AsyncMock()

            with patch("asyncio.get_running_loop") as mock_loop:
                mock_loop_instance = MagicMock()
                mock_loop.return_value = mock_loop_instance

                signal_handlers = {}

                def capture_handler(sig, handler):
                    signal_handlers[sig] = handler

                mock_loop_instance.add_signal_handler.side_effect = capture_handler

                async def run_and_shutdown():
                    task = asyncio.create_task(run_watch(project="my-project"))
                    await asyncio.sleep(0.01)
                    import signal

                    if signal.SIGINT in signal_handlers:
                        signal_handlers[signal.SIGINT]()
                    await task

                await run_and_shutdown()

            assert os.environ.get("BASIC_MEMORY_MCP_PROJECT") == "my-project"

        # Clean up env var
        os.environ.pop("BASIC_MEMORY_MCP_PROJECT", None)

    @pytest.mark.asyncio
    async def test_invalid_project_exits_with_error(self, mock_container):
        """--project with unknown name exits with error and still cleans up DB."""
        mock_config_manager = MagicMock()
        mock_config_manager.get_project.return_value = (None, None)

        with (
            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
            patch(
                "basic_memory.cli.commands.watch.ConfigManager",
                return_value=mock_config_manager,
            ),
            patch("basic_memory.cli.commands.watch.db") as mock_db,
        ):
            mock_db.shutdown_db = AsyncMock()

            with pytest.raises(typer.Exit) as exc_info:
                await run_watch(project="nonexistent")

            assert exc_info.value.exit_code == 1
            # DB should still be cleaned up even on early exit
            mock_db.shutdown_db.assert_called_once()

    @pytest.mark.asyncio
    async def test_shutdown_stops_coordinator_and_db(self, mock_container):
        """On shutdown, coordinator.stop() and db.shutdown_db() are called."""
        with (
            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
            patch("basic_memory.cli.commands.watch.db") as mock_db,
        ):
            mock_coordinator = AsyncMock()
            mock_coordinator_cls.return_value = mock_coordinator
            mock_db.shutdown_db = AsyncMock()

            with patch("asyncio.get_running_loop") as mock_loop:
                mock_loop_instance = MagicMock()
                mock_loop.return_value = mock_loop_instance

                signal_handlers = {}

                def capture_handler(sig, handler):
                    signal_handlers[sig] = handler

                mock_loop_instance.add_signal_handler.side_effect = capture_handler

                async def run_and_shutdown():
                    task = asyncio.create_task(run_watch())
                    await asyncio.sleep(0.01)
                    import signal

                    if signal.SIGINT in signal_handlers:
                        signal_handlers[signal.SIGINT]()
                    await task

                await run_and_shutdown()

            mock_coordinator.stop.assert_called_once()
            mock_db.shutdown_db.assert_called_once()

    @pytest.mark.asyncio
    async def test_falls_back_to_signal_module_on_windows(self, mock_container):
        """When add_signal_handler raises NotImplementedError, falls back to signal.signal()."""
        with (
            patch("basic_memory.cli.commands.watch.get_container", return_value=mock_container),
            patch("basic_memory.cli.commands.watch.initialize_app", AsyncMock()),
            patch("basic_memory.cli.commands.watch.SyncCoordinator") as mock_coordinator_cls,
            patch("basic_memory.cli.commands.watch.db") as mock_db,
        ):
            mock_coordinator = AsyncMock()
            mock_coordinator_cls.return_value = mock_coordinator
            mock_db.shutdown_db = AsyncMock()

            with patch("asyncio.get_running_loop") as mock_loop:
                mock_loop_instance = MagicMock()
                mock_loop.return_value = mock_loop_instance

                # Simulate Windows: add_signal_handler raises NotImplementedError
                mock_loop_instance.add_signal_handler.side_effect = NotImplementedError

                with patch("basic_memory.cli.commands.watch.signal.signal") as mock_signal:
                    # Track calls to signal.signal for the fallback path
                    registered_handlers = {}

                    def capture_signal(sig, handler):
                        registered_handlers[sig] = handler

                    mock_signal.side_effect = capture_signal

                    async def run_and_shutdown():
                        task = asyncio.create_task(run_watch())
                        await asyncio.sleep(0.01)
                        # Trigger shutdown via the fallback handler
                        if signal.SIGINT in registered_handlers:
                            registered_handlers[signal.SIGINT](signal.SIGINT, None)
                        await task

                    await run_and_shutdown()

                # Verify fallback signal.signal was called for both signals
                assert mock_signal.call_count == 2
                called_signals = {call.args[0] for call in mock_signal.call_args_list}
                assert signal.SIGINT in called_signals
                assert signal.SIGTERM in called_signals