ObsiGate/tests/test_services.py

# tests/test_services.py — Unit tests for the shared business-logic services (A2)
"""Tests for backend.services: vaults, files, search, and error mapping.

The index-dependent tests reuse the ``client`` fixture from conftest, which
builds the in-memory index for a ``TestVault``.
"""

from pathlib import Path

import pytest

from backend.services.errors import ServiceError
from backend.services.files import read_file_text, read_raw_file
from backend.services.search import list_tags, search_vaults
from backend.services.vaults import browse_directory, list_accessible_vaults

ADMIN = {"username": "admin", "role": "admin", "vaults": ["*"]}
LIMITED = {"username": "limited", "vaults": ["OtherVault"]}


# ═══════════════════════════════════════════════════════════════════
# Vaults
# ═══════════════════════════════════════════════════════════════════


class TestListAccessibleVaults:
    def test_admin_sees_vault(self, client):
        names = {v["name"] for v in list_accessible_vaults(ADMIN)}
        assert "TestVault" in names

    def test_limited_user_sees_none(self, client):
        assert list_accessible_vaults(LIMITED) == []

    def test_metadata_fields(self, client):
        vault = next(v for v in list_accessible_vaults(ADMIN) if v["name"] == "TestVault")
        assert set(vault) == {"name", "file_count", "tag_count", "type"}
        assert vault["file_count"] >= 1


class TestBrowseDirectory:
    def test_root_lists_children(self, client):
        data = browse_directory("TestVault", "")
        names = {i["name"] for i in data["items"]}
        assert "note1.md" in names
        assert "Projets" in names

    def test_missing_path_raises_not_found(self, client):
        with pytest.raises(ServiceError) as exc:
            browse_directory("TestVault", "nope")
        assert exc.value.code == "not_found"

    def test_unknown_vault_raises_not_found(self, client):
        with pytest.raises(ServiceError) as exc:
            browse_directory("MissingVault", "")
        assert exc.value.code == "not_found"


# ═══════════════════════════════════════════════════════════════════
# Files
# ═══════════════════════════════════════════════════════════════════


class TestReadFile:
    def test_read_text_redacts_secrets(self, client):
        from backend.indexer import get_vault_data

        root = Path(get_vault_data("TestVault")["path"])
        secret = root / "svc_secret.md"
        fake_jwt = "eyJ" + "a" * 30 + "." + "b" * 30 + "." + "c" * 30
        secret.write_text(f"token: {fake_jwt}\n", encoding="utf-8")
        try:
            data = read_file_text("TestVault", "svc_secret.md")
            assert "[JWT MASQUÉ]" in data["content"]
        finally:
            secret.unlink()

    def test_read_text_too_large(self, client):
        from backend.indexer import get_vault_data

        root = Path(get_vault_data("TestVault")["path"])
        big = root / "svc_big.txt"
        big.write_text("x" * 1000, encoding="utf-8")
        try:
            with pytest.raises(ServiceError) as exc:
                read_file_text("TestVault", "svc_big.txt", max_bytes=10)
            assert exc.value.code == "file_too_large"
        finally:
            big.unlink()

    def test_read_raw_no_redaction(self, client):
        data = read_raw_file("TestVault", "note1.md")
        assert "Python" in data["raw"]

    def test_read_missing_raises_not_found(self, client):
        with pytest.raises(ServiceError) as exc:
            read_raw_file("TestVault", "missing.md")
        assert exc.value.code == "not_found"

    def test_traversal_is_rejected(self, client):
        with pytest.raises(ServiceError) as exc:
            read_raw_file("TestVault", "../../etc/passwd")
        assert exc.value.code == "path_outside_vault"


# ═══════════════════════════════════════════════════════════════════
# Search
# ═══════════════════════════════════════════════════════════════════


class TestSearchService:
    def test_pagination(self, client):
        data = search_vaults("Python", limit=1, offset=0)
        assert data["limit"] == 1
        assert len(data["results"]) <= 1
        assert data["total"] >= 1

    def test_returns_api_shape(self, client):
        data = search_vaults("Python")
        assert set(data) >= {"query", "vault_filter", "tag_filter", "count", "total", "offset", "limit", "results"}

    def test_tags(self, client):
        tags = list_tags("TestVault")
        assert "python" in tags


# ═══════════════════════════════════════════════════════════════════
# Service → Tool error mapping
# ═══════════════════════════════════════════════════════════════════


class TestServiceErrorMapping:
    def test_not_found_maps_to_tool_not_found(self, client):
        from backend.tools.api import ToolContext, ToolNotFoundError, call_tool

        ctx = ToolContext(user=ADMIN, audit_enabled=False)
        with pytest.raises(ToolNotFoundError):
            call_tool("read_file", ctx, {"vault": "TestVault", "path": "nope.md"})

    def test_file_too_large_preserves_code(self, client, monkeypatch):
        from backend.indexer import get_vault_data
        from backend.tools import service as service_mod
        from backend.tools.api import ToolContext, ToolError, call_tool

        monkeypatch.setattr(service_mod, "TOOL_MAX_READ_BYTES", 10)
        root = Path(get_vault_data("TestVault")["path"])
        big = root / "svc_tool_big.txt"
        big.write_text("x" * 1000, encoding="utf-8")
        try:
            ctx = ToolContext(user=ADMIN, audit_enabled=False)
            with pytest.raises(ToolError) as exc:
                call_tool("read_file", ctx, {"vault": "TestVault", "path": "svc_tool_big.txt"})
            assert exc.value.code == "file_too_large"
        finally:
            big.unlink()