504 lines
19 KiB
Python
504 lines
19 KiB
Python
"""Integration: the admin sources-sync API (phase 32, task 01; phase 35,
|
|
task 03 re-points the URL resolution at the shared resolver).
|
|
|
|
Covers the in-process sync runner end to end over HTTP: anonymous 403s
|
|
on both endpoints; admin idle → 202 → ``success`` with the full
|
|
ImportSummary detail; 409 on a double trigger while a run is in flight;
|
|
``GitSyncError`` → ``failed`` with the failing repo named and **zero**
|
|
import attempts; empty on *both* origins (``git_sources`` table +
|
|
``BOR_GIT_SOURCES``) → ``failed`` loudly; an embedding failure →
|
|
``failed`` with any credentials masked; the import always runs with
|
|
``prune=True``; and the phase-31 overview trigger is change-gated (no
|
|
``lite`` call on an unchanged KB).
|
|
|
|
Phase 35: the runner resolves the repos through
|
|
:func:`app.rag.git_sources.effective_git_sources` — the **real**
|
|
resolver against the **real** ``git_sources`` table (truncated around
|
|
every test), so DB-over-env and the env fallback go through the actual
|
|
indirection; the env list is driven by a fresh ``Settings`` on the
|
|
resolver's module (the dev ``.env`` never leaks in).
|
|
|
|
The git / import / overview layers are monkeypatched in ``app.api.sync``
|
|
(same fake style as ``test_import_docs_git.py``) — no real git, no LLM:
|
|
the runner's state machine and HTTP surface are under test.
|
|
|
|
The admin client is used **as a context manager** on purpose: the
|
|
background sync task lives on the app's event loop, so the loop must
|
|
survive across requests — exactly how the app runs under uvicorn.
|
|
(A TestClient without the context manager starts a fresh loop per
|
|
request and would cancel the task on request exit.)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from collections.abc import Iterator
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api import sync as sync_api
|
|
from app.config import Settings
|
|
from app.main import app as fastapi_app
|
|
from app.models import GitSource
|
|
from app.rag import git_sources as git_sources_resolver
|
|
from app.rag.importer import ImportSummary
|
|
from app.rag.llm import EmbeddingError, LLMClient
|
|
from scripts.git_sync import GitSyncError
|
|
from tests.conftest import ADMIN_PASSWORD
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_sync_state() -> Iterator[None]:
|
|
"""The module-level status object + task are process-global: reset them
|
|
around every test (both before — a previous test's terminal state
|
|
would leak into the idle assertion — and after)."""
|
|
sync_api._status = sync_api.SyncStatus()
|
|
sync_api._task = None
|
|
yield
|
|
sync_api._status = sync_api.SyncStatus()
|
|
sync_api._task = None
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_git_sources(db: Session) -> Iterator[None]:
|
|
"""Phase 35: the runner resolves through the real ``git_sources``
|
|
table — global state, truncated around every test (the ``db``
|
|
fixture skips the file when Postgres is down)."""
|
|
db.execute(text("TRUNCATE git_sources"))
|
|
db.commit()
|
|
yield
|
|
db.execute(text("TRUNCATE git_sources"))
|
|
db.commit()
|
|
|
|
|
|
@pytest.fixture()
|
|
def sync_client() -> Iterator[TestClient]:
|
|
"""Context-managed TestClient — one app event loop across requests
|
|
(the background task must survive between the POST and the polls)."""
|
|
with TestClient(fastapi_app) as client:
|
|
yield client
|
|
|
|
|
|
def _settings(sources_dir: str = "~/bor-sources") -> Settings:
|
|
"""Fresh settings (no .env file); explicit kwargs beat any env leaks.
|
|
|
|
(The ``git_sources`` kwarg is gone with phase 35 — the runner reads
|
|
the URLs from the resolver, not from its own settings; the env list
|
|
is stubbed on the resolver's module via :func:`_stub_env`.)
|
|
"""
|
|
return Settings(_env_file=None, sources_dir=sources_dir) # pyright: ignore[reportCallIssue]
|
|
|
|
|
|
def _stub_env(monkeypatch: pytest.MonkeyPatch, git_sources: str = "") -> None:
|
|
"""The resolver's env fallback, driven by a fresh ``Settings`` (the
|
|
dev ``.env`` never leaks in — the task-02 pattern)."""
|
|
monkeypatch.setattr(
|
|
git_sources_resolver,
|
|
"get_settings",
|
|
lambda: Settings(_env_file=None, git_sources=git_sources), # pyright: ignore[reportCallIssue]
|
|
)
|
|
|
|
|
|
def _seed(db: Session, url: str) -> None:
|
|
db.add(GitSource(url=url))
|
|
db.commit()
|
|
|
|
|
|
def _login(client: TestClient) -> None:
|
|
r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
|
|
assert r.status_code == 204, f"admin login failed: {r.status_code} {r.text}"
|
|
|
|
|
|
def _poll(client: TestClient, want: str, timeout: float = 5.0) -> dict:
|
|
"""Poll ``GET /api/sync/status`` until ``state == want`` (terminal).
|
|
|
|
Any state other than ``running`` before the deadline fails loudly —
|
|
an unexpected ``failed`` must never be masked by the wait.
|
|
"""
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
body = client.get("/api/sync/status").json()
|
|
if body["state"] == want:
|
|
return body
|
|
assert body["state"] == "running", (
|
|
f"unexpected state {body['state']!r} while waiting for {want!r}: {body}"
|
|
)
|
|
time.sleep(0.05)
|
|
raise AssertionError(f"sync did not reach {want!r} within {timeout}s")
|
|
|
|
|
|
class FakeImportSources:
|
|
"""Records every ``import_sources`` call; returns a canned summary."""
|
|
|
|
def __init__(self, summary: ImportSummary, delay: float = 0.0) -> None:
|
|
self.summary = summary
|
|
self.delay = delay
|
|
self.sources: list[list[Path]] = []
|
|
self.llms: list[LLMClient] = []
|
|
self.prune_flags: list[bool] = []
|
|
|
|
async def __call__(
|
|
self,
|
|
sources: list[Path],
|
|
llm: LLMClient,
|
|
*,
|
|
prune: bool = False,
|
|
limit: int | None = None,
|
|
session: Session | None = None,
|
|
) -> ImportSummary:
|
|
self.sources.append(list(sources))
|
|
self.llms.append(llm)
|
|
self.prune_flags.append(prune)
|
|
if self.delay:
|
|
await asyncio.sleep(self.delay)
|
|
return self.summary
|
|
|
|
|
|
class FakeOverview:
|
|
"""Records every ``regenerate_overview`` call; canned result."""
|
|
|
|
def __init__(self, ok: bool = True) -> None:
|
|
self.ok = ok
|
|
self.llms: list[LLMClient] = []
|
|
|
|
async def __call__(self, llm: LLMClient, session: Session | None = None) -> bool:
|
|
self.llms.append(llm)
|
|
return self.ok
|
|
|
|
|
|
def _fake_clone() -> tuple[list[tuple[str, Path]], object]:
|
|
"""A ``clone_or_pull`` that materialises a checkout with one .md file."""
|
|
calls: list[tuple[str, Path]] = []
|
|
|
|
def fake_clone_or_pull(url: str, dest: Path | str) -> Path:
|
|
dest = Path(dest)
|
|
dest.mkdir(parents=True, exist_ok=True)
|
|
(dest / "notes.md").write_text(f"# {dest.name}\ncontent for the KB\n", encoding="utf-8")
|
|
calls.append((url, dest))
|
|
return dest
|
|
|
|
return calls, fake_clone_or_pull
|
|
|
|
|
|
# --- anonymous -------------------------------------------------------------
|
|
|
|
|
|
def test_anonymous_gets_403_on_both_endpoints(client: TestClient) -> None:
|
|
r = client.get("/api/sync/status")
|
|
assert r.status_code == 403
|
|
assert r.json() == {"detail": "admin only"}
|
|
r = client.post("/api/sync")
|
|
assert r.status_code == 403
|
|
assert r.json() == {"detail": "admin only"}
|
|
|
|
|
|
# --- admin: success --------------------------------------------------------
|
|
|
|
|
|
def test_admin_sync_success_reports_full_detail(
|
|
sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
|
|
) -> None:
|
|
repo_url = f"file://{tmp_path / 'repo.git'}"
|
|
_seed(db, repo_url) # the phase-35 resolver picks the DB row up
|
|
_stub_env(monkeypatch) # env must not matter once the table has a row
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
clone_calls, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
summary = ImportSummary(
|
|
files=5, added=1, updated=2, unchanged=2, pruned=3, errors=0,
|
|
chunks=11, embed_batches=4, summaries=1, summary_errors=0,
|
|
)
|
|
fake_import = FakeImportSources(summary)
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
fake_overview = FakeOverview(ok=True)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.get("/api/sync/status").json() == {
|
|
"state": "idle",
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
"detail": {},
|
|
"error": None,
|
|
}
|
|
|
|
r = sync_client.post("/api/sync")
|
|
assert r.status_code == 202
|
|
assert r.json() == {"detail": "sync started"}
|
|
|
|
body = _poll(sync_client, "success")
|
|
assert body["error"] is None
|
|
# ISO-8601 timestamps round-trip; finished after started.
|
|
started = datetime.fromisoformat(body["started_at"])
|
|
finished = datetime.fromisoformat(body["finished_at"])
|
|
assert finished >= started
|
|
# Every ImportSummary field + the overview flag, verbatim.
|
|
assert body["detail"] == {
|
|
"files": 5, "added": 1, "updated": 2, "unchanged": 2, "pruned": 3,
|
|
"errors": 0, "chunks": 11, "summaries": 1, "summary_errors": 0,
|
|
"overview": True,
|
|
}
|
|
# Git: the configured repo was cloned into BOR_SOURCES_DIR/<repo-name>/.
|
|
assert clone_calls == [(repo_url, tmp_path / "bor" / "repo")]
|
|
# Import: exactly the checkouts, with prune=True (the button is the
|
|
# canonical "mirror the repos" action) and a real LLMClient.
|
|
assert fake_import.sources == [[tmp_path / "bor" / "repo"]]
|
|
assert fake_import.prune_flags == [True]
|
|
assert len(fake_import.llms) == 1
|
|
assert isinstance(fake_import.llms[0], LLMClient)
|
|
# Overview: refreshed (added + updated > 0) with the same client.
|
|
assert fake_overview.llms == [fake_import.llms[0]]
|
|
|
|
|
|
def test_unchanged_kb_skips_overview_refresh(
|
|
sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
|
|
) -> None:
|
|
"""Phase-31 trigger is change-gated: added + updated == 0 → no ``lite`` call."""
|
|
repo_url = f"file://{tmp_path / 'repo.git'}"
|
|
_seed(db, repo_url)
|
|
_stub_env(monkeypatch)
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
_, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
summary = ImportSummary(files=7, added=0, updated=0, unchanged=7, pruned=0)
|
|
fake_import = FakeImportSources(summary)
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
fake_overview = FakeOverview(ok=True)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
body = _poll(sync_client, "success")
|
|
|
|
assert body["detail"]["overview"] is False
|
|
assert fake_overview.llms == [] # no wasted model call
|
|
assert len(fake_import.llms) == 1 # the import itself ran
|
|
|
|
|
|
# --- admin: concurrency ----------------------------------------------------
|
|
|
|
|
|
def test_double_trigger_while_running_returns_409(
|
|
sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
|
|
) -> None:
|
|
repo_url = f"file://{tmp_path / 'repo.git'}"
|
|
_seed(db, repo_url)
|
|
_stub_env(monkeypatch)
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
_, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
# The in-flight run takes a while (asyncio.sleep) so the second POST
|
|
# lands while it is still running.
|
|
fake_import = FakeImportSources(ImportSummary(files=1, added=1), delay=0.5)
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
|
|
r = sync_client.post("/api/sync") # second trigger while running
|
|
assert r.status_code == 409
|
|
assert r.json() == {"detail": "a sync is already running"}
|
|
|
|
body = sync_client.get("/api/sync/status").json()
|
|
assert body["state"] == "running"
|
|
assert body["started_at"] is not None
|
|
assert body["finished_at"] is None
|
|
assert body["error"] is None
|
|
|
|
# The (single) run completes; the import ran exactly once.
|
|
_poll(sync_client, "success")
|
|
assert len(fake_import.sources) == 1
|
|
|
|
|
|
# --- admin: failures -------------------------------------------------------
|
|
|
|
|
|
def test_git_failure_marks_failed_and_skips_import(
|
|
sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
|
|
) -> None:
|
|
repo_url = f"file://{tmp_path / 'bad.git'}"
|
|
_seed(db, repo_url)
|
|
_stub_env(monkeypatch)
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
|
|
def failing_clone(url: str, dest: Path | str) -> Path:
|
|
raise GitSyncError(
|
|
f"git clone --depth 1 {url} failed (exit 128): "
|
|
"fatal: repository not found"
|
|
)
|
|
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", failing_clone)
|
|
fake_import = FakeImportSources(ImportSummary())
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
fake_overview = FakeOverview(ok=True)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", fake_overview)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
|
|
body = _poll(sync_client, "failed")
|
|
assert "bad.git" in body["error"] # the failing repo is named
|
|
assert "fatal: repository not found" in body["error"]
|
|
assert body["detail"] == {}
|
|
assert body["finished_at"] is not None
|
|
assert fake_import.sources == [] # no partial import
|
|
assert fake_overview.llms == []
|
|
|
|
# A failed run leaves the system restartable: a new POST is accepted.
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
_poll(sync_client, "failed")
|
|
|
|
|
|
def test_no_git_sources_configured_fails_loudly(
|
|
sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""Both origins empty (the truncate fixture + a blank env) → the
|
|
fail-loud error names *both* (phase 35)."""
|
|
# Whitespace-only is just as unconfigured as empty.
|
|
_stub_env(monkeypatch, " , ")
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
clone_calls, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
fake_import = FakeImportSources(ImportSummary())
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
|
|
body = _poll(sync_client, "failed")
|
|
assert body["error"] == (
|
|
"no git sources configured (git_sources table empty and BOR_GIT_SOURCES unset)"
|
|
)
|
|
assert clone_calls == [] # git is never touched
|
|
assert fake_import.sources == []
|
|
|
|
|
|
def test_db_rows_win_over_env(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
db: Session,
|
|
tmp_path: Path,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Phase 35: a stored row beats ``BOR_GIT_SOURCES`` — only the DB URL
|
|
is cloned, and the started log names the origin."""
|
|
db_url = "https://db.example.com/managed.git"
|
|
_seed(db, db_url)
|
|
_stub_env(monkeypatch, "https://env.example.com/ignored.git")
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
clone_calls, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
fake_import = FakeImportSources(ImportSummary(files=1, added=1))
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
|
|
|
|
_login(sync_client)
|
|
with caplog.at_level(logging.INFO, logger="app.api.sync"):
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
body = _poll(sync_client, "success")
|
|
|
|
assert clone_calls == [(db_url, tmp_path / "bor" / "managed")]
|
|
assert fake_import.sources == [[tmp_path / "bor" / "managed"]]
|
|
assert "env.example.com" not in str(body) # the env URL never reaches the UI
|
|
assert any("sync: started repos=1 origin=db" in r.getMessage() for r in caplog.records)
|
|
|
|
|
|
def test_env_fallback_when_table_empty(
|
|
sync_client: TestClient,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Phase 35: with the table empty (the truncate fixture), the
|
|
``BOR_GIT_SOURCES`` list is what gets cloned — origin ``env``."""
|
|
env_url = "https://env.example.com/fallback.git"
|
|
_stub_env(monkeypatch, env_url)
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
clone_calls, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
fake_import = FakeImportSources(ImportSummary(files=1, added=1))
|
|
monkeypatch.setattr(sync_api, "import_sources", fake_import)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
|
|
|
|
_login(sync_client)
|
|
with caplog.at_level(logging.INFO, logger="app.api.sync"):
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
_poll(sync_client, "success")
|
|
|
|
assert clone_calls == [(env_url, tmp_path / "bor" / "fallback")]
|
|
assert any("sync: started repos=1 origin=env" in r.getMessage() for r in caplog.records)
|
|
|
|
|
|
def test_import_error_is_reported_with_credentials_masked(
|
|
sync_client: TestClient, monkeypatch: pytest.MonkeyPatch, db: Session, tmp_path: Path
|
|
) -> None:
|
|
repo_url = f"file://{tmp_path / 'repo.git'}"
|
|
_seed(db, repo_url)
|
|
_stub_env(monkeypatch)
|
|
monkeypatch.setattr(
|
|
sync_api,
|
|
"get_settings",
|
|
lambda: _settings(sources_dir=str(tmp_path / "bor")),
|
|
)
|
|
_, fake_clone = _fake_clone()
|
|
monkeypatch.setattr(sync_api, "clone_or_pull", fake_clone)
|
|
|
|
async def failing_import(
|
|
sources: list[Path],
|
|
llm: LLMClient,
|
|
*,
|
|
prune: bool = False,
|
|
limit: int | None = None,
|
|
session: Session | None = None,
|
|
) -> ImportSummary:
|
|
raise EmbeddingError(
|
|
"embeddings request to https://user:secret@aipi.reeseapps.com/v1 "
|
|
"failed: connection refused"
|
|
)
|
|
|
|
monkeypatch.setattr(sync_api, "import_sources", failing_import)
|
|
monkeypatch.setattr(sync_api, "regenerate_overview", FakeOverview(ok=True))
|
|
|
|
_login(sync_client)
|
|
assert sync_client.post("/api/sync").status_code == 202
|
|
|
|
body = _poll(sync_client, "failed")
|
|
assert "*****@aipi.reeseapps.com" in body["error"] # credentials masked
|
|
assert "user:secret" not in body["error"]
|
|
assert "connection refused" in body["error"] # the reason survives
|