Files
brain-of-reese/tests/integration/test_git_sources_api.py
T
ducoterra 8c706259e9
Build and Push Containers / build-and-push-app (push) Successful in 1m44s
Build and Push Containers / build-and-push-db (push) Successful in 13s
phase: 89_source_ignore_paths
All verification complete — TODO.md was already cleared in the roadmap commit; the two extra unit-test diffs are necessary fake-signature adaptations for the new keywords. Everything is green, no fixes were needed.

## Phase 89 — final verification pass: ALL GREEN

**Verified (all 6 task files present in `complete/`):**
- `git_sources.ignore_paths` JSONB column + migration 0013; `alembic downgrade -1 && upgrade head` round-trips (head `0013`)
- Importer: `normalize_ignore_path`/`is_ignored`/`_ignore_for_root`, `ignore` in walk + progress pre-walk, `ignore_by_root` in `import_sources`
- API: GET/POST carry list; admin-only `PATCH` (replace, 404/422 fixed details, anonymous 403)
- Pipelines wired: `_run_sync`, `_run_upload` re-upload, `scripts/import_docs.py`
- Sources-page box: dialog, §7.4 save lifecycle, `N ignored` tag, a11y; env rows get no box

**Test/lint results:**
- `uv run pytest` → 1808 passed
- `uv run pytest --cov=app --cov-report=term-missing` → TOTAL **99%** (>90%)
- `uv run pytest tests/e2e/test_source_ignore_paths.py -v --no-cov` → 6 passed (isolated, DB up)
- Regressions in isolation: `test_git_sources_admin` 6, `test_archive_upload_sources` 5, `test_sync_button` 3, `test_smoke` 3 — all passed
- `uv run ruff check . && uv run pyright` → clean (0 errors)

**Completion criteria:** box→PATCH 200→count+GET round-trip ✅ · sync excludes `ignore/` (no docs/chunks/embeddings/summaries) + prunes newly-ignored (pruned==2) ✅ · no-mid-path rule E2E ✅ · PATCH 404/422/replace/clear/403 ✅ · full gate green ✅ · commit + phase move left to harness per rules.

**Deviations:** none blocking — E2E pins `files == 4` (overview's "5" was an off-by-one vs its own 6-file tree, documented in-test); `tests/unit/test_importer.py` + `test_sync_button.py` test-double fakes extended for the new keywords (needed for the suite to stay green).

**Next pending phase:** none — `todo/` holds only this phase.
2026-09-09 01:45:42 -04:00

775 lines
31 KiB
Python

"""Integration: the admin sources CRUD API (phase 35, task 02; local
kind, phase 38, task 02).
Real Postgres (``podman compose up -d db``); the ``BOR_GIT_SOURCES``
fallback is exercised deterministically by monkeypatching the router's
``get_settings`` with a fresh ``Settings(_env_file=None, git_sources=…)``
(same pattern as ``test_sync_api.py`` — the dev ``.env`` never leaks in).
Contract under test:
* anonymous → 403 ``{"detail": "admin only"}`` on GET, POST (git and
local), and DELETE (phase 16 pattern, same as ``/api/sync``);
* GET — rows carry ``kind`` + ``path`` (phase 38); empty table + env
set → the git-only env rows (``kind: "git"``, ``path: null``) with
``from_env: true`` and null ``id``/``added_at``; empty table + empty
env → ``sources: []`` with ``from_env: true``; any DB rows →
``from_env: false`` and the env var is ignored (the phase's locked
decision); DB rows ordered by ``(added_at, id)``;
* POST ``kind=git`` (default) — 201 stored trimmed; duplicate (even with
different surrounding whitespace) → 409 with a generic detail that
never echoes the URL (credential safety), including when only the DB
unique index catches it; bad shape / blank / >500 chars → 422, also
input-free (the phase-35 contract, unchanged);
* POST ``kind=local`` (phase 38) — existing directory → 201, stored row
carries ``kind=local`` + the path expanded (``~`` resolved) and
trimmed; relative / missing / not-a-directory path → 422 naming the
path (not a secret); duplicate path → 409 naming the path (unique
index as backstop); wrong field combinations (git without url, local
without path, both kinds' fields) → 422;
* DELETE — 204 and gone; an emptied table falls back to the env list
again; unknown id → 404;
* ignore paths (phase 89) — POST accepts the RAW box lines for both
kinds (optional, absent → ``[]``), stored normalized (A1) with the A4
fixed-detail 422s (shared gate with PATCH); GET reports each row's
stored list (env rows ``[]``); ``PATCH /{source_id}`` (admin-only)
replaces the list wholesale (A5 — an empty list clears all), 404
unknown id, 422 fixed details for the A4 limits, the row otherwise
unchanged.
``git_sources`` is global state: truncated around every test.
"""
from __future__ import annotations
import uuid
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select, text
from sqlalchemy.orm import Session
from app.api import git_sources as git_sources_api
from app.config import Settings
from app.models import GitSource
@pytest.fixture(autouse=True)
def clean_git_sources(db: Session) -> Iterator[None]:
"""The stored list is global state: reset around every test."""
db.execute(text("TRUNCATE git_sources"))
db.commit()
yield
db.execute(text("TRUNCATE git_sources"))
db.commit()
def _settings(git_sources: str = "", **overrides: object) -> Settings:
"""Fresh settings with the ``.env`` file ignored; the explicit
kwargs beat any process env leaks (test_sync_api pattern).
``**overrides`` carries the per-test dir pins (``sources_dir`` /
``upload_dir`` — the phase-69 total-removal DELETE must never aim a
rmtree at the operator's real ``~/bor-sources``).
"""
return Settings(
_env_file=None, # pyright: ignore[reportCallIssue]
git_sources=git_sources,
**overrides, # pyright: ignore[reportCallIssue]
)
# --- anonymous -------------------------------------------------------------
def test_anonymous_gets_403_on_all_routes(client: TestClient, db: Session) -> None:
r = client.get("/api/git-sources")
assert r.status_code == 403
assert r.json() == {"detail": "admin only"}
r = client.post("/api/git-sources", json={"url": "https://anon.example.com/x.git"})
assert r.status_code == 403
assert r.json() == {"detail": "admin only"}
# The phase-38 local kind is gated the same way.
r = client.post("/api/git-sources", json={"kind": "local", "path": "/tmp"})
assert r.status_code == 403
assert r.json() == {"detail": "admin only"}
r = client.delete(f"/api/git-sources/{uuid.uuid4()}")
assert r.status_code == 403
assert r.json() == {"detail": "admin only"}
# The phase-89 ignore-list PATCH is gated the same way.
r = client.patch(f"/api/git-sources/{uuid.uuid4()}", json={"ignore_paths": ["a"]})
assert r.status_code == 403
assert r.json() == {"detail": "admin only"}
# Nothing landed in the table.
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0
# --- GET: env fallback -----------------------------------------------------
def test_get_empty_table_with_env_returns_env_rows(
admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
git_sources_api,
"get_settings",
lambda: _settings("https://a.example.com/one.git, git@b.example.com:two.git ,"),
)
r = admin_client.get("/api/git-sources")
assert r.status_code == 200
body = r.json()
assert body["from_env"] is True
# Whitespace-trimmed, empty entries dropped, order preserved; null
# ids; the env fallback is git-only (phase 38: kind + path fields).
assert body["sources"] == [
{
"id": None,
"kind": "git",
"url": "https://a.example.com/one.git",
"path": None,
"added_at": None,
# Env rows have no DB row to store a list on (phase 89).
"ignore_paths": [],
},
{
"id": None,
"kind": "git",
"url": "git@b.example.com:two.git",
"path": None,
"added_at": None,
"ignore_paths": [],
},
]
def test_get_empty_table_with_empty_env_returns_empty_list(
admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(git_sources_api, "get_settings", lambda: _settings())
r = admin_client.get("/api/git-sources")
assert r.status_code == 200
assert r.json() == {"sources": [], "from_env": True}
def test_get_orders_db_rows_by_added_at_then_id(admin_client: TestClient, db: Session) -> None:
base = datetime.now(UTC) - timedelta(hours=3)
db.add_all(
[
GitSource(url="https://example.com/oldest.git", added_at=base),
GitSource(url="https://example.com/second.git", added_at=base + timedelta(hours=1)),
GitSource(url="https://example.com/newest.git", added_at=base + timedelta(hours=2)),
# Same transaction → identical server-stamped added_at (≈ now,
# after the explicit rows): the id tie-break decides their order.
GitSource(url="https://example.com/tie-a.git"),
GitSource(url="https://example.com/tie-b.git"),
]
)
db.commit()
tie_rows = db.scalars(
select(GitSource).where(GitSource.url.like("https://example.com/tie-%"))
).all()
tie_ids = sorted(row.id for row in tie_rows)
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is False
assert [s["url"] for s in body["sources"][:3]] == [
"https://example.com/oldest.git",
"https://example.com/second.git",
"https://example.com/newest.git",
]
assert [s["id"] for s in body["sources"][3:]] == [str(i) for i in tie_ids]
# DB rows carry real ids + timestamps (the env shape has neither).
for s in body["sources"]:
assert s["id"] is not None
assert s["added_at"] is not None
# --- POST: create ----------------------------------------------------------
def test_post_creates_trimmed_and_list_stops_using_env(
admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
git_sources_api,
"get_settings",
lambda: _settings("https://env.example.com/env.git"),
)
r = admin_client.post("/api/git-sources", json={"url": " https://new.example.com/repo.git "})
assert r.status_code == 201
body = r.json()
assert body["url"] == "https://new.example.com/repo.git" # trimmed
uuid.UUID(body["id"])
assert body["added_at"] is not None
# Phase 89: the response gains ``ignore_paths`` — absent at create
# time → ``[]``.
assert set(body) == {"id", "url", "added_at", "ignore_paths"}
assert body["ignore_paths"] == []
# The DB row now wins: from_env False, the env URL is gone from the list.
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is False
assert [s["url"] for s in body["sources"]] == ["https://new.example.com/repo.git"]
def test_post_accepts_accepted_url_shapes(admin_client: TestClient) -> None:
for url in (
"https://github.com/owner/repo.git",
"http://git.local/repo.git",
"ssh://git@example.com/repo.git",
"git@github.com:owner/repo.git",
):
r = admin_client.post("/api/git-sources", json={"url": url})
assert r.status_code == 201, f"{url} must be accepted: {r.text}"
assert r.json()["url"] == url
def test_post_duplicate_url_returns_409_without_echoing_url(
admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The 409 detail is a fixed generic string — URLs may embed
``user:pass@`` credentials (phase 32's masking discipline)."""
monkeypatch.setattr(git_sources_api, "get_settings", lambda: _settings())
url = "https://user:secret@example.com/creds.git"
assert admin_client.post("/api/git-sources", json={"url": url}).status_code == 201
# Same URL with different surrounding whitespace: the trim makes it a
# duplicate too.
r = admin_client.post("/api/git-sources", json={"url": f" {url}\t"})
assert r.status_code == 409
detail = r.json()["detail"]
assert detail == "a git source with this URL already exists"
assert url not in detail
assert "user:secret" not in detail
# Exactly one row stored.
assert len(admin_client.get("/api/git-sources").json()["sources"]) == 1
def test_post_concurrent_insert_backstop_still_409(
admin_client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
"""If the duplicate pre-check misses (a concurrent insert lands
between the check and the commit), the DB unique index still yields
the generic 409 — never a 500."""
monkeypatch.setattr(git_sources_api, "get_settings", lambda: _settings())
url = "https://example.com/backstop.git"
assert admin_client.post("/api/git-sources", json={"url": url}).status_code == 201
real_select = git_sources_api.select
def blind_select(*args: Any, **kwargs: Any) -> Any:
if args and args[0] is GitSource: # the duplicate pre-check
# …now never matches — only the unique index can catch it.
return real_select(GitSource).where(GitSource.url == "zz-never-matches")
return real_select(*args, **kwargs)
monkeypatch.setattr(git_sources_api, "select", blind_select)
r = admin_client.post("/api/git-sources", json={"url": url})
assert r.status_code == 409
detail = r.json()["detail"]
assert detail == "a git source with this URL already exists"
assert url not in detail
def test_post_rejects_invalid_shapes_without_echoing_input(
admin_client: TestClient, db: Session
) -> None:
"""Bad shapes are 422 with a detail that never repeats the submitted
value."""
for bad in ("not a url", "host:repo", "ftp://example.com/x.git"):
r = admin_client.post("/api/git-sources", json={"url": bad})
assert r.status_code == 422, f"{bad!r} must be rejected"
assert bad not in r.text, f"422 detail must not echo the input ({bad!r})"
# Nothing stored.
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0
def test_post_rejects_blank_and_oversized_urls(admin_client: TestClient, db: Session) -> None:
"""Whitespace-only (trim → empty) and >500-char URLs are 422; the
500-char boundary passes."""
assert admin_client.post("/api/git-sources", json={"url": " \t\n "}).status_code == 422
assert admin_client.post("/api/git-sources", json={"url": "x" * 501}).status_code == 422
boundary = "https://" + "x" * 492 # exactly 500 chars
assert len(boundary) == 500
assert admin_client.post("/api/git-sources", json={"url": boundary}).status_code == 201
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 1
# --- POST: local kind (phase 38, task 02) ----------------------------------
def test_post_local_creates_stored_row_with_expanded_path(
admin_client: TestClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""``kind=local`` + an existing directory → 201; the stored row
carries ``kind=local`` and the path expanded (``~`` resolved via the
server's ``HOME``, whitespace trimmed)."""
monkeypatch.setenv("HOME", str(tmp_path / "home"))
real_dir = tmp_path / "home" / "notes"
real_dir.mkdir(parents=True)
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": " ~/notes\t"})
assert r.status_code == 201, r.text
body = r.json()
# The phase-35 response shape is unchanged — the local row reports
# its (expanded) path in ``url``; ``kind`` + ``path`` via GET;
# phase 89 adds ``ignore_paths`` (absent → ``[]``).
assert set(body) == {"id", "url", "added_at", "ignore_paths"}
assert body["ignore_paths"] == []
uuid.UUID(body["id"])
assert body["url"] == str(real_dir)
assert body["added_at"] is not None
row = admin_client.get("/api/git-sources").json()["sources"][0]
assert row["kind"] == "local"
assert row["path"] == str(real_dir)
assert row["url"] == str(real_dir)
assert row["id"] is not None
assert row["added_at"] is not None
def test_post_local_stores_trimmed_normalized_path(
admin_client: TestClient, tmp_path: Path
) -> None:
"""Absolute path with surrounding whitespace + trailing slash →
stored clean (trimmed, normalized)."""
real_dir = tmp_path / "plain"
real_dir.mkdir()
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": f" {real_dir}/ "})
assert r.status_code == 201, r.text
row = admin_client.get("/api/git-sources").json()["sources"][0]
assert row["path"] == str(real_dir)
def test_post_local_relative_path_returns_422_naming_path(
admin_client: TestClient, db: Session
) -> None:
"""A relative path fails loud at add-time — 422 naming the path
(relative or not, it is never stored)."""
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": "relative/dir"})
assert r.status_code == 422
assert r.json()["detail"] == "local source path is not a directory: relative/dir"
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0
def test_post_local_missing_path_returns_422_naming_path(
admin_client: TestClient, db: Session
) -> None:
"""A missing (absolute) path is a user error → 422 naming the path so
the owner sees exactly which directory failed."""
missing = f"/nonexistent/bor-test-{uuid.uuid4()}"
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": missing})
assert r.status_code == 422
assert r.json()["detail"] == f"local source path is not a directory: {missing}"
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0
def test_post_local_file_not_dir_returns_422(
admin_client: TestClient, tmp_path: Path
) -> None:
"""An existing *file* is not a directory → 422 naming the path."""
a_file = tmp_path / "a-file.md"
a_file.write_text("not a directory")
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": str(a_file)})
assert r.status_code == 422
assert r.json()["detail"] == f"local source path is not a directory: {a_file}"
def test_post_local_duplicate_path_returns_409_naming_path(
admin_client: TestClient, tmp_path: Path
) -> None:
"""Duplicate path (even with different surrounding whitespace) → 409
naming the path (a path is not a secret, unlike a git URL)."""
real_dir = tmp_path / "dups"
real_dir.mkdir()
assert admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": f" {real_dir}\t"})
assert r.status_code == 409
detail = r.json()["detail"]
assert detail == f"a local source with this path already exists: {real_dir}"
# Exactly one row stored.
assert len(admin_client.get("/api/git-sources").json()["sources"]) == 1
def test_post_local_concurrent_insert_backstop_still_409(
admin_client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""If the duplicate pre-check misses (a concurrent insert lands
between the check and the commit), the DB unique index on ``path``
still yields the 409 naming the path — never a 500."""
real_dir = tmp_path / "backstop"
real_dir.mkdir()
assert (
admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
.status_code
== 201
)
real_select = git_sources_api.select
def blind_select(*args: Any, **kwargs: Any) -> Any:
if args and args[0] is GitSource: # the duplicate pre-check
# …now never matches — only the unique index can catch it.
return real_select(GitSource).where(GitSource.url == "zz-never-matches")
return real_select(*args, **kwargs)
monkeypatch.setattr(git_sources_api, "select", blind_select)
r = admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
assert r.status_code == 409
assert r.json()["detail"] == f"a local source with this path already exists: {real_dir}"
def test_post_wrong_field_combinations_return_422(
admin_client: TestClient, db: Session, tmp_path: Path
) -> None:
"""git without url, local without path, and both kinds' fields are
422 with fixed details — nothing is stored."""
real_dir = tmp_path / "combo"
real_dir.mkdir()
assert admin_client.post("/api/git-sources", json={"kind": "git"}).status_code == 422
r = admin_client.post(
"/api/git-sources",
json={"kind": "git", "url": "https://example.com/both.git", "path": str(real_dir)},
)
assert r.status_code == 422
assert r.json()["detail"] == "a git source takes a url, not a path"
assert admin_client.post("/api/git-sources", json={"kind": "local"}).status_code == 422
# Whitespace-only path trims to empty → 422 as well (the schema's
# min-length guard).
assert (
admin_client.post("/api/git-sources", json={"kind": "local", "path": " "}).status_code
== 422
)
r = admin_client.post(
"/api/git-sources",
json={"kind": "local", "url": "https://example.com/both.git", "path": str(real_dir)},
)
assert r.status_code == 422
assert r.json()["detail"] == "a local source takes a path, not a url"
# Unknown kind and an oversized path are 422 too.
assert (
admin_client.post(
"/api/git-sources", json={"kind": "svn", "url": "https://example.com/x.git"}
).status_code
== 422
)
assert (
admin_client.post(
"/api/git-sources", json={"kind": "local", "path": "/" + "x" * 2000}
).status_code
== 422
)
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0
def test_get_mixed_kinds_in_added_order(
admin_client: TestClient, db: Session, tmp_path: Path
) -> None:
"""GET mixes git + local rows in ``(added_at, id)`` order; git rows
report ``path: null``, local rows their stored path."""
git_url = "https://example.com/mixed.git"
db.add(GitSource(url=git_url, kind="git", added_at=datetime.now(UTC) - timedelta(hours=1)))
db.commit()
real_dir = tmp_path / "mixed"
real_dir.mkdir()
assert admin_client.post("/api/git-sources", json={"kind": "local", "path": str(real_dir)})
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is False
assert [s["url"] for s in body["sources"]] == [git_url, str(real_dir)]
git_row, local_row = body["sources"]
assert git_row["kind"] == "git"
assert git_row["path"] is None
assert git_row["id"] is not None
assert local_row["kind"] == "local"
assert local_row["path"] == str(real_dir)
assert local_row["id"] is not None
# --- DB rows win over env ---------------------------------------------------
def test_db_rows_win_over_env(admin_client: TestClient, db: Session, monkeypatch) -> None:
"""Seed a row AND set the env: the GET returns only the DB rows and
``from_env: false`` — the env var is ignored once the table has rows."""
monkeypatch.setattr(
git_sources_api, "get_settings", lambda: _settings("https://env.example.com/env.git")
)
db.add(GitSource(url="https://db.example.com/db.git"))
db.commit()
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is False
assert [s["url"] for s in body["sources"]] == ["https://db.example.com/db.git"]
for s in body["sources"]:
assert s["id"] is not None
assert s["added_at"] is not None
# --- DELETE -----------------------------------------------------------------
def test_delete_removes_row_and_falls_back_to_env(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
# Phase 69: the DELETE is a total removal — the (absent) checkout
# dir it would rmtree is pinned at fresh tmp dirs, so the test never
# depends on the operator's real ``~/bor-sources`` being clean; the
# document prune it performs owns the KB tables the same way the
# other suites do (no row's source name may shadow real docs).
db.execute(text("TRUNCATE chunks, documents"))
db.commit()
monkeypatch.setattr(
git_sources_api,
"get_settings",
lambda: _settings(
"https://env.example.com/env.git",
sources_dir=str(tmp_path / "sources"),
upload_dir=str(tmp_path / "uploads"),
),
)
created = admin_client.post("/api/git-sources", json={"url": "https://new.example.com/x.git"})
assert created.status_code == 201
assert admin_client.delete(f"/api/git-sources/{created.json()['id']}").status_code == 204
# The table is empty again → the env fallback is live once more
# (git-only rows, phase 38).
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is True
assert body["sources"] == [
{
"id": None,
"kind": "git",
"url": "https://env.example.com/env.git",
"path": None,
"added_at": None,
"ignore_paths": [], # env rows: no DB row to store a list on
}
]
db.execute(text("TRUNCATE chunks, documents"))
db.commit()
def test_delete_unknown_id_returns_404(admin_client: TestClient) -> None:
r = admin_client.delete(f"/api/git-sources/{uuid.uuid4()}")
assert r.status_code == 404
assert r.json() == {"detail": "git source not found"}
def test_delete_invalid_id_returns_422(admin_client: TestClient) -> None:
assert admin_client.delete("/api/git-sources/not-a-uuid").status_code == 422
# --- ignore paths (phase 89) ----------------------------------------------
def test_post_stores_normalized_ignore_paths_both_kinds(
admin_client: TestClient, tmp_path: Path
) -> None:
"""POST accepts the RAW box lines for both kinds (A1 normalization
in the API layer) — the normalized list is stored and round-trips
through GET."""
r = admin_client.post(
"/api/git-sources",
json={
"url": "https://example.com/ig.git",
"ignore_paths": ["/my/files/", " my/files2 ", "x"],
},
)
assert r.status_code == 201, r.text
assert r.json()["ignore_paths"] == ["my/files", "my/files2", "x"]
real_dir = tmp_path / "ig"
real_dir.mkdir()
r = admin_client.post(
"/api/git-sources",
json={"kind": "local", "path": str(real_dir), "ignore_paths": ["//skip/", "keep"]},
)
assert r.status_code == 201, r.text
assert r.json()["ignore_paths"] == ["skip", "keep"]
# Round-trip through GET (keyed by url — (added_at, id) order of two
# same-millisecond inserts is not the point under test).
by_url = {
s["url"]: s["ignore_paths"] for s in admin_client.get("/api/git-sources").json()["sources"]
}
assert by_url["https://example.com/ig.git"] == ["my/files", "my/files2", "x"]
assert by_url[str(real_dir)] == ["skip", "keep"]
def test_post_without_ignore_paths_reports_empty_list(
admin_client: TestClient, db: Session
) -> None:
"""Absent ``ignore_paths`` at create time → ``[]`` — both in the 201
response and in the GET round-trip (the migration's server default)."""
r = admin_client.post("/api/git-sources", json={"url": "https://example.com/none.git"})
assert r.status_code == 201
assert r.json()["ignore_paths"] == []
body = admin_client.get("/api/git-sources").json()
assert body["sources"][0]["ignore_paths"] == []
# The stored value is the JSONB server default, not Python-only.
row = db.scalars(select(GitSource)).one()
assert row.ignore_paths == []
def test_post_rejects_invalid_ignore_paths_like_patch(
admin_client: TestClient, db: Session
) -> None:
"""POST shares ``_validate_ignore_paths`` with PATCH — the same A4
fixed 422 details; nothing is stored."""
for payload, detail in (
([" "], "ignore paths must be non-empty"),
([f"e{i}" for i in range(201)], "a source has at most 200 ignore paths"),
(["a" * 501], "an ignore path exceeds 500 characters"),
):
r = admin_client.post(
"/api/git-sources",
json={"url": "https://example.com/bad-ignore.git", "ignore_paths": payload},
)
assert r.status_code == 422, f"{detail!r}: {r.text}"
assert r.json()["detail"] == detail
assert db.execute(text("SELECT count(*) FROM git_sources")).scalar_one() == 0
def test_get_reports_stored_ignore_paths_and_env_rows_empty(
admin_client: TestClient, db: Session, monkeypatch: pytest.MonkeyPatch
) -> None:
"""GET reports each DB row's stored list; env-fallback rows (no DB
row to store a list on) report ``[]`` with ``from_env: true``."""
db.add(GitSource(url="https://example.com/stored.git", ignore_paths=["docs/drafts"]))
db.commit()
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is False
assert body["sources"][0]["ignore_paths"] == ["docs/drafts"]
db.execute(text("DELETE FROM git_sources"))
db.commit()
monkeypatch.setattr(
git_sources_api,
"get_settings",
lambda: _settings("https://a.example.com/env.git"),
)
body = admin_client.get("/api/git-sources").json()
assert body["from_env"] is True
assert body["sources"] == [
{
"id": None,
"kind": "git",
"url": "https://a.example.com/env.git",
"path": None,
"added_at": None,
"ignore_paths": [],
}
]
def test_patch_replaces_ignore_paths_including_clear(admin_client: TestClient) -> None:
"""PATCH 200 — REPLACE semantics (A5): the body list, normalized,
becomes the row's whole list; an empty list clears all; ``id``/
``url``/``added_at`` are unchanged; the response is the
``GitSourceOut`` shape incl. the new list."""
created = admin_client.post(
"/api/git-sources",
json={"url": "https://example.com/patch.git", "ignore_paths": ["old/"]},
)
assert created.status_code == 201
before = created.json()
r = admin_client.patch(f"/api/git-sources/{before['id']}", json={"ignore_paths": ["a/", "b"]})
assert r.status_code == 200, r.text
body = r.json()
assert set(body) == {"id", "url", "added_at", "ignore_paths"}
assert body["ignore_paths"] == ["a", "b"] # normalized
for key in ("id", "url", "added_at"):
assert body[key] == before[key]
# Round-trip through GET.
assert (
admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == ["a", "b"]
)
# An empty list clears all.
r = admin_client.patch(f"/api/git-sources/{before['id']}", json={"ignore_paths": []})
assert r.status_code == 200, r.text
assert r.json()["ignore_paths"] == []
assert admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == []
def test_patch_missing_field_is_422(admin_client: TestClient) -> None:
"""``ignore_paths`` is REQUIRED (replace semantics, A5) — an absent
field is a 422 with the model's own detail."""
created = admin_client.post("/api/git-sources", json={"url": "https://example.com/req.git"})
assert created.status_code == 201
r = admin_client.patch(f"/api/git-sources/{created.json()['id']}", json={})
assert r.status_code == 422
# The model's own (Pydantic) detail — the missing required field.
assert any(
item.get("loc") == ["body", "ignore_paths"] for item in r.json()["detail"]
)
def test_patch_unknown_id_returns_404(admin_client: TestClient) -> None:
r = admin_client.patch(f"/api/git-sources/{uuid.uuid4()}", json={"ignore_paths": ["a"]})
assert r.status_code == 404
assert r.json() == {"detail": "git source not found"}
def test_patch_invalid_id_returns_422(admin_client: TestClient) -> None:
r = admin_client.patch("/api/git-sources/not-a-uuid", json={"ignore_paths": []})
assert r.status_code == 422
def test_patch_422s_are_fixed_details(admin_client: TestClient) -> None:
"""The A4 422s are exact fixed strings (never echoing the input),
and a rejected PATCH leaves the row's list unchanged."""
created = admin_client.post(
"/api/git-sources", json={"url": "https://example.com/v.git", "ignore_paths": ["keep"]}
)
assert created.status_code == 201
sid = created.json()["id"]
for payload, detail in (
# Whitespace-only → empty after normalization: 422, not a silent
# drop (the UI drops blank lines client-side; the API is
# defensive).
([" "], "ignore paths must be non-empty"),
(list(f"e{i}" for i in range(201)), "a source has at most 200 ignore paths"),
(["a" * 501], "an ignore path exceeds 500 characters"),
):
r = admin_client.patch(f"/api/git-sources/{sid}", json={"ignore_paths": payload})
assert r.status_code == 422, f"{detail!r}: {r.text}"
assert r.json()["detail"] == detail
# Nothing changed by the failed PATCHes.
assert admin_client.get("/api/git-sources").json()["sources"][0]["ignore_paths"] == ["keep"]
def test_patch_accepts_a4_boundaries(admin_client: TestClient) -> None:
"""Exactly 200 entries and a 500-char entry (post-normalization)
are the accepted edge of A4."""
created = admin_client.post("/api/git-sources", json={"url": "https://example.com/bnd.git"})
assert created.status_code == 201
sid = created.json()["id"]
r = admin_client.patch(
f"/api/git-sources/{sid}", json={"ignore_paths": [f"e/{i}" for i in range(200)]}
)
assert r.status_code == 200, r.text
assert len(r.json()["ignore_paths"]) == 200
long_entry = "a" * 500
assert len(long_entry) == 500
r = admin_client.patch(f"/api/git-sources/{sid}", json={"ignore_paths": [long_entry]})
assert r.status_code == 200, r.text
assert r.json()["ignore_paths"] == [long_entry]