111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
"""Unit: the shared git-source resolver (phase 35, task 03).
|
|
|
|
``effective_git_sources`` is driven with a stubbed session (no Postgres)
|
|
and a fresh ``Settings(_env_file=None)`` env (monkeypatched into the
|
|
resolver's module — the dev ``.env`` never leaks in, same pattern as the
|
|
integration suites): DB rows win in ``(added_at, id)`` order (env
|
|
ignored), the ``BOR_GIT_SOURCES`` list is a fallback only while the
|
|
table is empty (phase-28 CSV parse reused), and both-empty yields
|
|
``([], "env")`` so the callers keep their fail-loud behavior.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from app.config import Settings
|
|
from app.models import GitSource
|
|
from app.rag import git_sources as resolver
|
|
|
|
|
|
class _FakeScalars:
|
|
"""The ``.scalars(stmt).all()`` tail of the resolver's query."""
|
|
|
|
def __init__(self, rows: list[GitSource]) -> None:
|
|
self._rows = rows
|
|
|
|
def all(self) -> list[GitSource]:
|
|
return self._rows
|
|
|
|
|
|
class _FakeSession:
|
|
"""Just enough of a SQLAlchemy session for the resolver (test_steering
|
|
pattern). Records the statement so the ordering can be asserted."""
|
|
|
|
def __init__(self, rows: list[GitSource]) -> None:
|
|
self._rows = rows
|
|
self.statements: list[Any] = []
|
|
|
|
def scalars(self, stmt: Any) -> _FakeScalars:
|
|
self.statements.append(stmt)
|
|
return _FakeScalars(self._rows)
|
|
|
|
|
|
def _stub_env(monkeypatch: pytest.MonkeyPatch, git_sources: str) -> None:
|
|
"""Point the resolver's env fallback at a fresh Settings (no .env)."""
|
|
monkeypatch.setattr(
|
|
resolver,
|
|
"get_settings",
|
|
lambda: Settings(_env_file=None, git_sources=git_sources), # pyright: ignore[reportCallIssue]
|
|
)
|
|
|
|
|
|
# --- the three branches ----------------------------------------------------
|
|
|
|
|
|
def test_db_rows_win_over_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Seeded table + env set → the DB list, origin ``db`` (env ignored)."""
|
|
_stub_env(monkeypatch, "https://env.example/ignored.git")
|
|
session = _FakeSession([GitSource(url="https://db.example/a.git"), GitSource(url="https://db.example/b.git")])
|
|
|
|
urls, origin = resolver.effective_git_sources(session) # pyright: ignore[reportArgumentType]
|
|
|
|
assert urls == ["https://db.example/a.git", "https://db.example/b.git"]
|
|
assert origin == "db"
|
|
|
|
|
|
def test_env_fallback_while_table_empty(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Empty table + env set → the env list, origin ``env``.
|
|
|
|
The CSV parse is ``Settings.git_source_list`` itself (phase 28):
|
|
whitespace-trimmed, empty entries dropped, order preserved.
|
|
"""
|
|
_stub_env(monkeypatch, " https://env.example/one.git , ,git@env.example:two.git ")
|
|
session = _FakeSession([])
|
|
|
|
urls, origin = resolver.effective_git_sources(session) # pyright: ignore[reportArgumentType]
|
|
|
|
assert urls == ["https://env.example/one.git", "git@env.example:two.git"]
|
|
assert origin == "env"
|
|
|
|
|
|
def test_both_empty_returns_empty_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Empty table + unset env → ``([], "env")`` — the callers fail loudly."""
|
|
_stub_env(monkeypatch, " , ") # whitespace-only is just as unconfigured as empty
|
|
session = _FakeSession([])
|
|
|
|
urls, origin = resolver.effective_git_sources(session) # pyright: ignore[reportArgumentType]
|
|
|
|
assert urls == []
|
|
assert origin == "env"
|
|
|
|
|
|
# --- the ordering ----------------------------------------------------------
|
|
|
|
|
|
def test_db_rows_ordered_by_added_at_then_id(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""The statement orders by ``(added_at, id)`` — oldest first, the id
|
|
tie-break deciding same-timestamp inserts (matches the API's GET)."""
|
|
_stub_env(monkeypatch, "")
|
|
session = _FakeSession([GitSource(url="https://db.example/a.git")])
|
|
|
|
resolver.effective_git_sources(session) # pyright: ignore[reportArgumentType]
|
|
|
|
assert len(session.statements) == 1
|
|
compiled = str(session.statements[0].compile(compile_kwargs={"literal_binds": True}))
|
|
assert re.search(
|
|
r"ORDER BY\s+git_sources\.added_at ASC,\s+git_sources\.id ASC", compiled
|
|
), compiled
|