Newer
Older
navi-1 / tests / unit / api / test_admin_tools_reload.py
@Eugene Sukhodolskiy Eugene Sukhodolskiy 10 hours ago 4 KB admin: POST /admin/tools/reload
"""POST /admin/tools/reload — the button in settings for admins whose profile
carries no reload_tools, and the same reload_all() underneath it."""

import textwrap
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from navi.auth import User
from navi.config import Settings
from navi.context_providers._loader import ContextProviderRegistry
from navi.core.registry import ToolRegistry


def _write_tool(path: Path, name: str) -> None:
    path.write_text(
        textwrap.dedent(
            f'''
            name = "{name}"
            description = "d"
            parameters = {{"type": "object", "properties": {{}}}}

            async def execute(params):
                return "{name}"
            '''
        ).lstrip()
    )


@pytest.fixture(autouse=True)
def clean_dependency_overrides():
    """Start from a known auth state on the shared app.

    Other suites install overrides on navi.main.app and leave them behind, and
    an overridden require_admin silently outranks this module's own
    get_current_user override — the 403 test then gets a 200. Their overrides
    are put back afterwards.
    """
    import navi.api.deps as deps_mod
    from navi.main import app

    saved = dict(app.dependency_overrides)
    for dep in (
        deps_mod.require_admin,
        deps_mod.require_user,
        deps_mod.require_permission,
        deps_mod.get_current_user,
    ):
        app.dependency_overrides.pop(dep, None)
    yield
    app.dependency_overrides.clear()
    app.dependency_overrides.update(saved)


@pytest.fixture
def admin_client(tmp_path, monkeypatch):
    """Anonymous-admin client (auth disabled) with stub registries."""
    import navi.api.deps as deps_mod
    import navi.auth.deps as auth_deps
    import navi.core.reload as reload_mod
    from navi.main import app

    monkeypatch.setattr(
        auth_deps, "settings",
        Settings(_env_file=None, navi_persona_file="", navi_auth_enabled=False),
    )

    tools_dir = tmp_path / "tools"
    tools_dir.mkdir()
    _write_tool(tools_dir / "alpha.py", "alpha")
    providers_dir = tmp_path / "providers"
    providers_dir.mkdir()

    registry = ToolRegistry()
    monkeypatch.setattr(
        reload_mod, "settings",
        reload_mod.settings.model_copy(
            update={"tools_dir": str(tools_dir), "context_providers_dir": str(providers_dir)}
        ),
    )
    monkeypatch.setattr("navi.core.tool_utils.load_user_enabled_tools", list)
    monkeypatch.setattr(deps_mod, "get_tool_registry", lambda: registry)
    monkeypatch.setattr(deps_mod, "get_cp_registry", lambda: ContextProviderRegistry())
    monkeypatch.setattr(deps_mod, "get_mcp_manager", lambda: None)

    return TestClient(app), registry, tools_dir


def test_reload_reports_what_it_loaded(admin_client):
    client, _, _ = admin_client

    response = client.post("/admin/tools/reload")

    assert response.status_code == 200
    data = response.json()
    assert data["ok"] is True
    assert data["tools"] == ["alpha"]
    assert data["total_tools"] == 1
    assert data["tool_errors"] == {}
    assert data["missing_enabled"] == []
    assert data["mcp_tools"] == []


def test_the_registry_itself_is_updated(admin_client):
    """Not just a report — the running registry gains the new tool, so the
    next turn of any live session resolves it."""
    client, registry, _ = admin_client

    client.post("/admin/tools/reload")

    assert registry.get("alpha") is not None


def test_a_broken_tool_is_reported_without_hiding_the_rest(admin_client):
    client, _, tools_dir = admin_client
    (tools_dir / "broken.py").write_text("not python (")

    data = client.post("/admin/tools/reload").json()

    assert data["ok"] is False
    assert data["tools"] == ["alpha"]
    assert list(data["tool_errors"]) == ["broken.py"]


def test_non_admin_gets_403(monkeypatch):
    """The reload rewrites the global toolset — it is not a user action."""
    import navi.api.deps as deps_mod
    import navi.auth.deps as auth_deps
    from navi.main import app

    monkeypatch.setattr(
        auth_deps, "settings",
        Settings(_env_file=None, navi_persona_file="", navi_auth_enabled=True),
    )
    app.dependency_overrides[deps_mod.get_current_user] = lambda: User(
        id="u1", email="u@test.com", role="user"
    )
    try:
        response = TestClient(app).post("/admin/tools/reload")
    finally:
        app.dependency_overrides.pop(deps_mod.get_current_user, None)

    assert response.status_code == 403


def test_anonymous_gets_401_when_auth_is_on(monkeypatch):
    import navi.auth.deps as auth_deps
    from navi.main import app

    monkeypatch.setattr(
        auth_deps, "settings",
        Settings(_env_file=None, navi_persona_file="", navi_auth_enabled=True),
    )
    app.dependency_overrides.pop(auth_deps.get_current_user, None)

    assert TestClient(app).post("/admin/tools/reload").status_code == 401