311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""Unit: the total-removal helpers (phase 69, task 01).
|
|
|
|
Covers ``app.rag.source_removal`` with plain objects and ``tmp_path``
|
|
(no FastAPI, no database):
|
|
|
|
* ``resolve_source_name`` — exactly the sync/importer document labels:
|
|
git URL shapes (https ``.git`` / bare, scp-style ``git@host:repo.git``,
|
|
``ssh://``) and local paths (``~`` expansion, trailing slash, the
|
|
``path or url`` fallback);
|
|
* ``managed_dir_for`` — git → ``sources_dir/<repo>/``; local → the
|
|
stored dir only when it is ``upload_dir`` itself or nested under it
|
|
(the containment check, so a sibling named ``uploads-foo`` never
|
|
counts); any other local path → ``None`` (owner's own dir, never
|
|
touched);
|
|
* ``remove_managed_dir`` — ``None``/absent no-op (no filesystem write),
|
|
present tree removed → ``True``, ``OSError`` logged (``logger.
|
|
exception``) and returned as ``False`` — never raises;
|
|
* ``has_sibling`` — same-name sibling (git ``…/r`` vs ``…/r.git``, and
|
|
across kinds) → ``True``; different names → ``False``; self-excluded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import pytest
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import GitSource
|
|
from app.rag import source_removal
|
|
from app.rag.source_removal import (
|
|
has_sibling,
|
|
managed_dir_for,
|
|
remove_managed_dir,
|
|
resolve_source_name,
|
|
)
|
|
|
|
|
|
def _git(url: str) -> GitSource:
|
|
return GitSource(id=uuid.uuid4(), url=url, kind="git")
|
|
|
|
|
|
def _local(path: str, path_column: str | None = None) -> GitSource:
|
|
# Phase 38 mirrors the expanded path in the NOT-NULL ``url`` column;
|
|
# ``path_column=None`` exercises the ``row.path or row.url`` fallback.
|
|
return GitSource(id=uuid.uuid4(), url=path, kind="local", path=path_column)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resolve_source_name
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_resolve_git_url_https_dot_git_suffix() -> None:
|
|
assert resolve_source_name(_git("https://example.com/reese/homelab.git")) == "homelab"
|
|
|
|
|
|
def test_resolve_git_url_https_bare() -> None:
|
|
assert resolve_source_name(_git("https://example.com/reese/homelab")) == "homelab"
|
|
|
|
|
|
def test_resolve_git_url_scp_style_git_at() -> None:
|
|
"""``git@host:repo.git`` — the ``:`` basename split (the phase-28
|
|
``repo_name`` behavior, reused not re-implemented)."""
|
|
assert resolve_source_name(_git("git@github.com:reese/deployments.git")) == "deployments"
|
|
assert resolve_source_name(_git("git@github.com:reese/deployments")) == "deployments"
|
|
|
|
|
|
def test_resolve_git_url_ssh_scheme() -> None:
|
|
assert resolve_source_name(_git("ssh://git@example.com/reese/ops.git")) == "ops"
|
|
|
|
|
|
def test_resolve_git_url_strips_whitespace() -> None:
|
|
assert resolve_source_name(_git(" https://example.com/reese/x.git ")) == "x"
|
|
|
|
|
|
def test_resolve_local_expands_tilde(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
monkeypatch.setenv("HOME", str(tmp_path / "home"))
|
|
(tmp_path / "home" / "notes").mkdir(parents=True)
|
|
assert resolve_source_name(_local("~/notes")) == "notes"
|
|
|
|
|
|
def test_resolve_local_trailing_slash_and_nested() -> None:
|
|
assert resolve_source_name(_local("/srv/docs/notes/")) == "notes"
|
|
assert resolve_source_name(_local("/srv/a/b/c")) == "c"
|
|
|
|
|
|
def test_resolve_local_falls_back_to_url_column_when_path_null() -> None:
|
|
"""Phase 38 mirrors the path in ``url`` — the ``or`` fallback keeps a
|
|
NULL ``path`` row resolvable the same way."""
|
|
assert resolve_source_name(_local("/srv/docs/notes", path_column=None)) == "notes"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# managed_dir_for
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_managed_dir_git_maps_to_sources_dir_repo_name(tmp_path: Path) -> None:
|
|
sources = tmp_path / "sources"
|
|
row = _git("https://example.com/reese/homelab.git")
|
|
assert managed_dir_for(row, sources, tmp_path / "uploads") == sources / "homelab"
|
|
|
|
|
|
def test_managed_dir_git_expands_tilde_sources_dir(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
monkeypatch.setenv("HOME", str(tmp_path / "home"))
|
|
(tmp_path / "home" / "bor-sources").mkdir(parents=True)
|
|
row = _git("git@github.com:reese/ops.git")
|
|
got = managed_dir_for(row, Path("~/bor-sources"), tmp_path / "uploads")
|
|
assert got == tmp_path / "home" / "bor-sources" / "ops"
|
|
|
|
|
|
def test_managed_dir_local_upload_dir_itself(tmp_path: Path) -> None:
|
|
upload = tmp_path / "uploads"
|
|
upload.mkdir()
|
|
row = _local(str(upload))
|
|
assert managed_dir_for(row, tmp_path / "sources", upload) == upload
|
|
|
|
|
|
def test_managed_dir_local_nested_under_upload_dir(tmp_path: Path) -> None:
|
|
upload = tmp_path / "uploads"
|
|
folder = upload / "my-notes"
|
|
folder.mkdir(parents=True)
|
|
row = _local(str(folder))
|
|
assert managed_dir_for(row, tmp_path / "sources", upload) == folder
|
|
|
|
|
|
def test_managed_dir_local_deeply_nested_under_upload_dir(tmp_path: Path) -> None:
|
|
upload = tmp_path / "uploads"
|
|
folder = upload / "a" / "b"
|
|
folder.mkdir(parents=True)
|
|
row = _local(str(folder))
|
|
assert managed_dir_for(row, tmp_path / "sources", upload) == folder
|
|
|
|
|
|
def test_managed_dir_unresolvable_path_is_none(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""Containment that cannot be established (an unresolvable path) is
|
|
``None`` — never delete when in doubt (the defensive branch)."""
|
|
row = _local("/srv/docs/notes")
|
|
|
|
def boom(self: Path, strict: bool = False) -> Path: # noqa: ARG001
|
|
raise OSError("simulated resolve failure")
|
|
|
|
monkeypatch.setattr(Path, "resolve", boom)
|
|
assert managed_dir_for(row, tmp_path / "sources", tmp_path / "uploads") is None
|
|
|
|
|
|
def test_managed_dir_local_foreign_path_is_none(tmp_path: Path) -> None:
|
|
"""The owner's own directory — never app-managed, never touched."""
|
|
foreign = tmp_path / "own" / "docs"
|
|
foreign.mkdir(parents=True)
|
|
row = _local(str(foreign))
|
|
assert managed_dir_for(row, tmp_path / "sources", tmp_path / "uploads") is None
|
|
|
|
|
|
def test_managed_dir_local_prefix_sibling_never_counts(tmp_path: Path) -> None:
|
|
"""The containment edge: ``…/u-evil`` is NOT under ``…/u`` — a
|
|
prefix-sharing sibling name must never map into the upload dir."""
|
|
evil = tmp_path / "u-evil"
|
|
evil.mkdir()
|
|
row = _local(str(evil))
|
|
assert managed_dir_for(row, tmp_path / "sources", tmp_path / "u") is None
|
|
|
|
|
|
def test_managed_dir_local_symlink_escaping_upload_dir_is_none(tmp_path: Path) -> None:
|
|
"""A symlink stored under the upload dir that points at the owner's
|
|
dir resolves OUTSIDE — containment fails → ``None`` (never delete
|
|
through a link)."""
|
|
upload = tmp_path / "uploads"
|
|
foreign = tmp_path / "foreign"
|
|
upload.mkdir()
|
|
foreign.mkdir()
|
|
link = upload / "sneaky"
|
|
link.symlink_to(foreign)
|
|
row = _local(str(link))
|
|
assert managed_dir_for(row, tmp_path / "sources", upload) is None
|
|
|
|
|
|
def test_managed_dir_local_under_upload_dir_symlink_still_maps(tmp_path: Path) -> None:
|
|
"""Mirror image: a link that stays under the upload dir resolves
|
|
inside it — the stored path is still the app-managed dir."""
|
|
upload = tmp_path / "uploads"
|
|
real = upload / "real"
|
|
upload.mkdir()
|
|
real.mkdir()
|
|
link = upload / "alias"
|
|
link.symlink_to(real)
|
|
row = _local(str(link))
|
|
assert managed_dir_for(row, tmp_path / "sources", upload) == link
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# remove_managed_dir
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_remove_managed_dir_none_is_noop_false() -> None:
|
|
assert remove_managed_dir(None) is False
|
|
|
|
|
|
def test_remove_managed_dir_absent_is_noop_false(tmp_path: Path) -> None:
|
|
missing = tmp_path / "never-created"
|
|
assert remove_managed_dir(missing) is False
|
|
assert not missing.exists() # no filesystem write
|
|
|
|
|
|
def test_remove_managed_dir_present_tree_removed_true(tmp_path: Path) -> None:
|
|
tree = tmp_path / "homelab"
|
|
(tree / "sub").mkdir(parents=True)
|
|
(tree / "alpha.md").write_text("one")
|
|
(tree / "sub" / "bravo.md").write_text("two")
|
|
assert remove_managed_dir(tree) is True
|
|
assert not tree.exists()
|
|
|
|
|
|
def test_remove_managed_dir_oserror_logged_not_fatal(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""A permission/busy-dir failure is ``logger.exception`` + ``False``
|
|
— removal never raises (the DB removal is already committed)."""
|
|
stuck = tmp_path / "stuck"
|
|
stuck.mkdir()
|
|
(stuck / "alpha.md").write_text("one")
|
|
|
|
def boom(directory: Path) -> None:
|
|
raise OSError(13, "Permission denied", str(directory))
|
|
|
|
monkeypatch.setattr(source_removal.shutil, "rmtree", boom)
|
|
with caplog.at_level(logging.ERROR, logger="app.rag.source_removal"):
|
|
assert remove_managed_dir(stuck) is False
|
|
# The exception was logged (with the traceback) and the dir is as it
|
|
# was (the rmtree never ran) — inert, self-heals on re-add.
|
|
assert stuck.is_dir()
|
|
errors = [r for r in caplog.records if r.levelno >= logging.ERROR]
|
|
assert len(errors) == 1
|
|
assert "could not remove the on-disk directory" in errors[0].getMessage()
|
|
assert errors[0].exc_info is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# has_sibling
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeScalars:
|
|
def __init__(self, rows: list[GitSource]) -> None:
|
|
self._rows = rows
|
|
|
|
def all(self) -> list[GitSource]:
|
|
return list(self._rows)
|
|
|
|
|
|
class _FakeSession:
|
|
"""The duck-typed seam: ``has_sibling`` only calls
|
|
``db.scalars(select(GitSource))`` (the statement builds fine without
|
|
a connection) — the fake returns the registry rows it was given."""
|
|
|
|
def __init__(self, rows: list[GitSource]) -> None:
|
|
self._rows = rows
|
|
|
|
def scalars(self, statement: object) -> _FakeScalars: # noqa: ARG002
|
|
return _FakeScalars(self._rows)
|
|
|
|
|
|
def _sibling(rows: list[GitSource], row: GitSource) -> bool:
|
|
"""``has_sibling`` against a fake registry session — the duck-typed
|
|
fake goes through the seam via ``cast`` (the ``test_agent.py``
|
|
pattern)."""
|
|
return has_sibling(cast("Session", _FakeSession(rows)), row)
|
|
|
|
|
|
def test_has_sibling_same_name_git_dot_git_pair() -> None:
|
|
a = _git("https://example.com/reese/r")
|
|
b = _git("https://example.com/reese/r.git")
|
|
assert _sibling([a, b], a) is True
|
|
assert _sibling([a, b], b) is True
|
|
|
|
|
|
def test_has_sibling_different_names_is_false() -> None:
|
|
a = _git("https://example.com/reese/one.git")
|
|
b = _git("https://example.com/reese/two.git")
|
|
assert _sibling([a, b], a) is False
|
|
assert _sibling([a, b], b) is False
|
|
|
|
|
|
def test_has_sibling_self_excluded() -> None:
|
|
a = _git("https://example.com/reese/only.git")
|
|
assert _sibling([a], a) is False
|
|
|
|
|
|
def test_has_sibling_empty_registry_is_false() -> None:
|
|
a = _git("https://example.com/reese/only.git")
|
|
assert _sibling([], a) is False
|
|
|
|
|
|
def test_has_sibling_across_kinds_same_resolved_name() -> None:
|
|
"""A git URL whose ``repo_name`` equals a local dir name is a
|
|
sibling too — both index under the same label."""
|
|
git_row = _git("https://example.com/reese/notes.git")
|
|
local_row = _local("/srv/docs/notes")
|
|
assert _sibling([git_row, local_row], git_row) is True
|
|
assert _sibling([git_row, local_row], local_row) is True
|