basicmachines-co-basic-memory/tests/api/test_management_router.py

"""Tests for management router API endpoints (minimal mocking).

These endpoints are mostly simple state checks and wiring; we use stub objects
and pytest monkeypatch instead of standard-library mocks.
"""

from __future__ import annotations

import pytest
from fastapi import FastAPI

from basic_memory.api.routers.management_router import (
    WatchStatusResponse,
    get_watch_status,
    start_watch_service,
    stop_watch_service,
)


class _Request:
    def __init__(self, app: FastAPI):
        self.app = app


class _Task:
    def __init__(self, *, done: bool):
        self._done = done
        self.cancel_called = False

    def done(self) -> bool:
        return self._done

    def cancel(self) -> None:
        self.cancel_called = True


@pytest.fixture
def app_with_state() -> FastAPI:
    app = FastAPI()
    app.state.watch_task = None
    return app


@pytest.mark.asyncio
async def test_get_watch_status_not_running(app_with_state: FastAPI):
    app_with_state.state.watch_task = None
    resp = await get_watch_status(_Request(app_with_state))
    assert isinstance(resp, WatchStatusResponse)
    assert resp.running is False


@pytest.mark.asyncio
async def test_get_watch_status_running(app_with_state: FastAPI):
    app_with_state.state.watch_task = _Task(done=False)
    resp = await get_watch_status(_Request(app_with_state))
    assert resp.running is True


@pytest.mark.asyncio
async def test_start_watch_service_when_not_running(monkeypatch, app_with_state: FastAPI):
    app_with_state.state.watch_task = None

    created = {"watch_service": None, "task": None}

    class _StubWatchService:
        def __init__(self, *, app_config, project_repository):
            self.app_config = app_config
            self.project_repository = project_repository
            created["watch_service"] = self

    def _create_background_sync_task(sync_service, watch_service):
        created["task"] = _Task(done=False)
        return created["task"]

    # start_watch_service imports these inside the function, so patch at the source modules.
    monkeypatch.setattr("basic_memory.sync.WatchService", _StubWatchService)
    monkeypatch.setattr(
        "basic_memory.sync.background_sync.create_background_sync_task",
        _create_background_sync_task,
    )

    project_repository = object()
    sync_service = object()

    resp = await start_watch_service(_Request(app_with_state), project_repository, sync_service)
    assert resp.running is True
    assert app_with_state.state.watch_task is created["task"]
    assert created["watch_service"] is not None
    assert created["watch_service"].project_repository is project_repository


@pytest.mark.asyncio
async def test_start_watch_service_already_running(monkeypatch, app_with_state: FastAPI):
    existing = _Task(done=False)
    app_with_state.state.watch_task = existing

    def _should_not_be_called(*_args, **_kwargs):
        raise AssertionError("create_background_sync_task should not be called if already running")

    monkeypatch.setattr(
        "basic_memory.sync.background_sync.create_background_sync_task",
        _should_not_be_called,
    )

    resp = await start_watch_service(_Request(app_with_state), object(), object())
    assert resp.running is True
    assert app_with_state.state.watch_task is existing


@pytest.mark.asyncio
async def test_stop_watch_service_not_running(app_with_state: FastAPI):
    app_with_state.state.watch_task = None
    resp = await stop_watch_service(_Request(app_with_state))
    assert resp.running is False


@pytest.mark.asyncio
async def test_stop_watch_service_already_done(app_with_state: FastAPI):
    app_with_state.state.watch_task = _Task(done=True)
    resp = await stop_watch_service(_Request(app_with_state))
    assert resp.running is False