ObsiGate/tests/conftest.py

# tests/conftest.py — Shared fixtures for ObsiGate test suite
import os
import sys
import shutil
import json
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

# Add project root to path so we can import backend modules
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))


@pytest.fixture(autouse=True)
def _reset_tool_ratelimit():
    """Isolate the per-identity tool rate limiter between tests."""
    from backend.tools import ratelimit

    ratelimit.reset()
    yield
    ratelimit.reset()


@pytest.fixture(autouse=True)
def _clean_env():
    """Ensure no vault env vars leak between tests — but preserve test vault config."""
    saved = {}
    # Only clean VAULT_N/DIR_N vars, NOT OBSIGATE_ vars needed by the app
    for key in list(os.environ.keys()):
        if key.startswith(("VAULT_", "DIR_")):
            saved[key] = os.environ.pop(key)
    yield
    # Restore
    for key in list(os.environ.keys()):
        if key.startswith(("VAULT_", "DIR_")):
            if key not in saved:
                os.environ.pop(key)
    os.environ.update(saved)


@pytest.fixture
def test_vault_dir(tmp_path: Path) -> str:
    """Create a temporary Obsidian-style vault with markdown files."""
    vault = tmp_path / "TestVault"
    vault.mkdir()

    # Simple markdown file
    (vault / "note1.md").write_text(
        "---\ntags:\n  - python\n  - tutorial\ntitle: Introduction à Python\n---\n"
        "# Introduction à Python\nPython est un langage de programmation moderne.\n"
        "Il supporte la programmation orientée objet et fonctionnelle.\n"
        "La syntaxe de Python est claire et lisible.\n",
        encoding="utf-8",
    )

    # File with inline tags and wikilinks
    (vault / "note2.md").write_text(
        "---\ntags:\n  - docker\nstatut: actif\nauteur: Jean Dupont\ntitle: Docker Guide\n---\n"
        "# Docker Guide\nDocker est une plateforme de conteneurisation.\n"
        "Voir aussi [[Introduction à Python]] pour les scripts.\n"
        "Et aussi [[Proxmox Setup]] pour l'infrastructure.\n"
        "Un tag inline #devops pour le fun.\n",
        encoding="utf-8",
    )

    # File in subdirectory
    sub = vault / "Projets"
    sub.mkdir()
    (sub / "projet.md").write_text(
        "---\ntags:\n  - projet\n  - python\ntitle: Mon Projet\n---\n"
        "# Mon Projet\nUtilise Docker et Python.\n"
        "Voir [[Introduction à Python]].\n",
        encoding="utf-8",
    )

    # Non-markdown file
    (vault / "config.json").write_text('{"key": "value"}', encoding="utf-8")

    # File with accents in title
    (vault / "café_crème.md").write_text(
        "---\ntitle: Café Crème\n---\n# Café Crème\nUn bon café.\n",
        encoding="utf-8",
    )

    return str(vault)


@pytest.fixture
def app_with_vault(test_vault_dir: str):
    """Create a FastAPI TestClient with a test vault configured.

    Imports app lazily to avoid side-effects at module load time.
    Disables auth and watcher for testing.
    """
    os.environ["VAULT_1_NAME"] = "TestVault"
    os.environ["VAULT_1_PATH"] = test_vault_dir
    os.environ["OBSIGATE_AUTH_ENABLED"] = "false"

    # Prevent watcher from starting (not needed for tests)
    import backend.main
    backend.main._load_config = lambda: {"watcher_enabled": False}

    from backend.main import app
    from backend.indexer import build_index
    import asyncio

    # Build the index
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(build_index())

    # Build inverted index for search
    from backend.search import init_inverted_index
    init_inverted_index()

    # Build semantic (embedding) index — uses the dependency-free hash fallback
    from backend.semantic_search import init_semantic_index
    init_semantic_index()

    client = TestClient(app)
    return client


@ pytest.fixture
def client(app_with_vault):
    """Alias for app_with_vault."""
    return app_with_vault


@ pytest.fixture
def admin_client(tmp_path):
    """TestClient with auth enabled, isolated temp data, admin user provisioned.
    
    Used by tests that need admin authentication (e.g., /api/health/detailed).
    """
    data_dir = tmp_path / "data"
    data_dir.mkdir()

    import json
    from backend.auth.password import hash_password

    pw_hash = hash_password("chab30")
    users = {
        "version": 1,
        "users": {
            "admin": {
                "id": "admin-1",
                "username": "admin",
                "display_name": "admin",
                "password_hash": pw_hash,
                "role": "admin",
                "vaults": ["*"],
                "active": True,
                "created_at": "2026-01-01T00:00:00",
            },
            "normaluser": {
                "id": "user-1",
                "username": "normaluser",
                "display_name": "normal",
                "password_hash": hash_password("normal123"),
                "role": "user",
                "vaults": ["TestVault"],
                "active": True,
                "created_at": "2026-01-01T00:00:00",
            },
        },
    }
    (data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")

    src_secret = Path("data/secret.key")
    if src_secret.exists():
        shutil.copy2(str(src_secret), str(data_dir / "secret.key"))

    orig_cwd = os.getcwd()
    test_vault_path = os.path.abspath("test-vault")
    os.chdir(str(tmp_path))

    os.environ["VAULT_1_NAME"] = "TestVault"
    os.environ["VAULT_1_PATH"] = test_vault_path
    os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
    os.environ["OBSIGATE_ADMIN_USER"] = "admin"
    os.environ["OBSIGATE_ADMIN_PASSWORD"] = "chab30"
    os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"

    import backend.main
    backend.main._load_config = lambda: {"watcher_enabled": False}

    from backend.main import app
    from backend.indexer import build_index, index
    import asyncio

    for key in list(index.keys()):
        del index[key]

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(build_index())

    from backend.search import init_inverted_index
    init_inverted_index()

    from backend.semantic_search import init_semantic_index
    init_semantic_index()

    from fastapi.testclient import TestClient
    client = TestClient(app)
    yield client

    if hasattr(client, "close"):
        client.close()
    loop.run_until_complete(asyncio.sleep(0))

    os.chdir(orig_cwd)
    shutil.rmtree(str(tmp_path), ignore_errors=True)
    for k in [
        "VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
        "OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED",
    ]:
        os.environ.pop(k, None)