596 lines
21 KiB
Python
596 lines
21 KiB
Python
"""Sources screens: list/detail/form render real registry data, mutations go
|
|
through the management service, and unauthorized principals see nothing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field, replace
|
|
from datetime import UTC, datetime
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from ragcore.api.v1.control.dependencies import get_principal
|
|
from ragcore.application.documents import (
|
|
DocumentListFilter,
|
|
DocumentListPage,
|
|
DocumentRecord,
|
|
DocumentStatusRecord,
|
|
DocumentVersionState,
|
|
)
|
|
from ragcore.application.jobs import Job
|
|
from ragcore.application.sources import (
|
|
SourceItemRecord,
|
|
SourceListFilter,
|
|
SourceListPage,
|
|
SourceOverview,
|
|
SourceRecord,
|
|
SourceRegistryError,
|
|
SourceStatus,
|
|
SourceSyncOutcome,
|
|
SourceSyncRunRecord,
|
|
)
|
|
from ragcore.domain.authorization import (
|
|
AuthorizationError,
|
|
PermissionDenied,
|
|
Principal,
|
|
PrincipalType,
|
|
ResourceNotFound,
|
|
)
|
|
from ragcore.domain.control import Role
|
|
from ragcore.domain.sources import ConnectorKind, SourceSyncState
|
|
from ragcore.main import create_app
|
|
from ragcore.web.sources import get_source_management
|
|
|
|
WORKSPACE_ID = uuid4()
|
|
SPACE_ID = uuid4()
|
|
|
|
|
|
@dataclass
|
|
class FakeSourceManagement:
|
|
sources: dict[UUID, SourceRecord] = field(default_factory=dict)
|
|
runs: dict[UUID, list[SourceSyncRunRecord]] = field(default_factory=dict)
|
|
documents: dict[UUID, list[DocumentRecord]] = field(default_factory=dict)
|
|
failed_items: dict[UUID, list[SourceItemRecord]] = field(default_factory=dict)
|
|
manage: bool = True
|
|
manage_filesystem: bool = True
|
|
actions: list[tuple[str, UUID]] = field(default_factory=list)
|
|
created: list[dict[str, object]] = field(default_factory=list)
|
|
|
|
def can_manage(self, principal: Principal) -> bool:
|
|
return self.manage
|
|
|
|
def can_manage_filesystem_sources(self, principal: Principal) -> bool:
|
|
return self.manage and self.manage_filesystem
|
|
|
|
def _check(self) -> None:
|
|
if not self.manage:
|
|
raise PermissionDenied("no")
|
|
|
|
async def list_sources(self, principal: Principal, filters: SourceListFilter):
|
|
items = [
|
|
source
|
|
for source in self.sources.values()
|
|
if (filters.kind is None or source.kind is filters.kind)
|
|
and (filters.search is None or filters.search.lower() in source.name.lower())
|
|
and source.status is not SourceStatus.DELETED
|
|
]
|
|
page = SourceListPage(items=tuple(items), total=len(items), limit=filters.limit, offset=0)
|
|
counts = {source.id: len(self.documents.get(source.id, [])) for source in items}
|
|
return page, counts, {SPACE_ID: "Runbooks space"}
|
|
|
|
async def get_source(self, principal: Principal, source_id: UUID) -> SourceOverview:
|
|
source = self.sources.get(source_id)
|
|
if source is None:
|
|
raise ResourceNotFound("source not found")
|
|
return SourceOverview(
|
|
source=source,
|
|
document_count=len(self.documents.get(source_id, [])),
|
|
space_name="Runbooks space",
|
|
failed_items=tuple(self.failed_items.get(source_id, [])),
|
|
)
|
|
|
|
async def sync_history(self, principal: Principal, source_id: UUID, *, limit: int = 20):
|
|
return tuple(self.runs.get(source_id, []))
|
|
|
|
async def documents_of(
|
|
self, principal: Principal, source_id: UUID, filters: DocumentListFilter
|
|
):
|
|
items = tuple(
|
|
DocumentStatusRecord(
|
|
document=document,
|
|
latest_state=DocumentVersionState.PUBLISHED,
|
|
parser_version="text-parser-1",
|
|
chunk_profile_version="prose-v1",
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
for document in self.documents.get(source_id, [])
|
|
)
|
|
return DocumentListPage(items=items, total=len(items), limit=filters.limit, offset=0)
|
|
|
|
async def create_source(
|
|
self, principal, *, name, kind, space_id, config_values, trace_id,
|
|
sync_interval_seconds=None,
|
|
):
|
|
self._check()
|
|
if kind is ConnectorKind.FOLDER and not config_values.get("root", "").startswith("/"):
|
|
raise SourceRegistryError("root must be an absolute non-traversing path")
|
|
record = SourceRecord(
|
|
id=uuid4(),
|
|
workspace_id=WORKSPACE_ID,
|
|
space_id=space_id,
|
|
name=name,
|
|
kind=kind,
|
|
config={"root": config_values.get("root")},
|
|
)
|
|
self.sources[record.id] = record
|
|
self.created.append(
|
|
{
|
|
"name": name,
|
|
"kind": kind,
|
|
"config": dict(config_values),
|
|
"sync_interval_seconds": sync_interval_seconds,
|
|
}
|
|
)
|
|
return record
|
|
|
|
async def update_source(
|
|
self, principal, source_id, *, name, config_values, trace_id,
|
|
sync_interval_seconds=None,
|
|
):
|
|
self._check()
|
|
self.actions.append(("update", source_id))
|
|
self.updated_intervals.append(sync_interval_seconds)
|
|
return self.sources[source_id]
|
|
|
|
async def set_status(self, principal, source_id, status, *, trace_id):
|
|
self._check()
|
|
if source_id not in self.sources:
|
|
raise ResourceNotFound("source not found")
|
|
self.actions.append((status.value, source_id))
|
|
return self.sources[source_id]
|
|
|
|
async def request_sync(self, principal, source_id, *, trace_id):
|
|
self._check()
|
|
source = self.sources.get(source_id)
|
|
if source is None:
|
|
raise ResourceNotFound("source not found")
|
|
if not source.syncable:
|
|
raise SourceRegistryError("this source kind is fed by its callers, not synced")
|
|
self.actions.append(("sync", source_id))
|
|
return Job(
|
|
id=uuid4(),
|
|
job_key="k",
|
|
operation="sync_source",
|
|
target_key=str(source_id),
|
|
profile_version="sync-v1",
|
|
)
|
|
|
|
|
|
def _principal(role: Role = Role.WORKSPACE_ADMIN) -> Principal:
|
|
return Principal(uuid4(), WORKSPACE_ID, PrincipalType.HUMAN, frozenset({role}))
|
|
|
|
|
|
def _fixture(
|
|
service: FakeSourceManagement,
|
|
principal: Principal | None = None,
|
|
*,
|
|
resolve_failed_jobs: bool = True,
|
|
) -> TestClient:
|
|
app = create_app()
|
|
app.dependency_overrides[get_principal] = lambda: principal or _principal()
|
|
app.dependency_overrides[get_source_management] = lambda: service
|
|
|
|
class OwnedJobs:
|
|
async def get_owned(
|
|
self, job_id: UUID, *, workspace_id: UUID | None
|
|
) -> Job | None:
|
|
known = any(
|
|
item.last_failure_job_id == job_id
|
|
for items in service.failed_items.values()
|
|
for item in items
|
|
)
|
|
if not resolve_failed_jobs or not known or workspace_id != WORKSPACE_ID:
|
|
return None
|
|
return Job(
|
|
id=job_id,
|
|
workspace_id=WORKSPACE_ID,
|
|
job_key=f"failed-{job_id}",
|
|
operation="ingest_upload",
|
|
target_key="failed-item",
|
|
profile_version="upload-v1",
|
|
)
|
|
|
|
app.state.job_repository = OwnedJobs()
|
|
return TestClient(app)
|
|
|
|
|
|
def _folder_source(name: str = "Infra runbooks") -> SourceRecord:
|
|
return SourceRecord(
|
|
id=uuid4(),
|
|
workspace_id=WORKSPACE_ID,
|
|
space_id=SPACE_ID,
|
|
name=name,
|
|
kind=ConnectorKind.FOLDER,
|
|
config={"root": "/mnt/runbooks", "include": ["*.md"], "trust_level": "trusted_internal"},
|
|
last_sync_at=datetime(2026, 8, 20, 10, 0, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
def test_sources_list_shows_empty_state_with_action_and_grouped_navigation() -> None:
|
|
client = _fixture(FakeSourceManagement())
|
|
response = client.get("/admin/sources")
|
|
assert response.status_code == 200
|
|
assert "No sources configured yet" in response.text
|
|
assert 'href="/admin/sources/new"' in response.text
|
|
# The grouped information architecture is rendered for every screen.
|
|
for label in ("Knowledge", "Access", "Quality", "Operations", "Sources", "Documents"):
|
|
assert label in response.text
|
|
assert 'aria-current="page"' in response.text
|
|
|
|
|
|
def test_sources_list_renders_registry_rows_with_counts_and_filters() -> None:
|
|
service = FakeSourceManagement()
|
|
source = _folder_source()
|
|
service.sources[source.id] = source
|
|
service.documents[source.id] = [
|
|
DocumentRecord(
|
|
uuid4(), WORKSPACE_ID, SPACE_ID, str(source.id), "a.md", "A", source_id=source.id
|
|
)
|
|
]
|
|
client = _fixture(service)
|
|
response = client.get("/admin/sources?q=infra&kind=folder")
|
|
assert response.status_code == 200
|
|
assert "Infra runbooks" in response.text
|
|
assert "Runbooks space" in response.text
|
|
assert "2026-08-20 10:00" in response.text
|
|
assert f'href="/admin/sources/{source.id}"' in response.text
|
|
assert f'href="/admin/documents?source_id={source.id}"' in response.text
|
|
assert 'role="search"' in response.text
|
|
assert 'data-operations-journey="source-to-search"' in response.text
|
|
assert 'data-mobile-records="sources"' in response.text
|
|
|
|
|
|
def test_invalid_source_filter_is_reported_without_dropping_valid_filters() -> None:
|
|
service = FakeSourceManagement()
|
|
matching = _folder_source("Infra runbooks")
|
|
unrelated = _folder_source("Employee handbook")
|
|
service.sources[matching.id] = matching
|
|
service.sources[unrelated.id] = unrelated
|
|
|
|
response = _fixture(service).get("/admin/sources?q=infra&kind=not-a-kind")
|
|
|
|
assert response.status_code == 200
|
|
assert "Infra runbooks" in response.text
|
|
assert "Employee handbook" not in response.text
|
|
assert "Unrecognised type filter ignored" in response.text
|
|
assert 'value="infra"' in response.text
|
|
assert 'value="not-a-kind"' not in response.text
|
|
|
|
|
|
def test_filtered_source_empty_state_does_not_claim_the_registry_is_empty() -> None:
|
|
service = FakeSourceManagement()
|
|
source = _folder_source()
|
|
service.sources[source.id] = source
|
|
|
|
response = _fixture(service).get("/admin/sources?q=no-such-source")
|
|
|
|
assert response.status_code == 200
|
|
assert "No sources match these filters" in response.text
|
|
assert "No sources configured yet" not in response.text
|
|
assert 'href="/admin/sources"' in response.text
|
|
|
|
|
|
def test_source_detail_shows_configuration_documents_and_sync_history() -> None:
|
|
service = FakeSourceManagement()
|
|
source = _folder_source()
|
|
service.sources[source.id] = source
|
|
service.runs[source.id] = [
|
|
SourceSyncRunRecord(
|
|
id=uuid4(),
|
|
source_id=source.id,
|
|
started_at=datetime(2026, 8, 20, 10, 0, tzinfo=UTC),
|
|
finished_at=datetime(2026, 8, 20, 10, 1, tzinfo=UTC),
|
|
outcome=SourceSyncOutcome.COMPLETED,
|
|
discovered=12,
|
|
new=3,
|
|
changed=1,
|
|
unchanged=8,
|
|
removed=0,
|
|
)
|
|
]
|
|
service.documents[source.id] = [
|
|
DocumentRecord(
|
|
uuid4(),
|
|
WORKSPACE_ID,
|
|
SPACE_ID,
|
|
str(source.id),
|
|
"runbooks/backup.md",
|
|
"Backup",
|
|
source_id=source.id,
|
|
)
|
|
]
|
|
client = _fixture(service)
|
|
response = client.get(f"/admin/sources/{source.id}")
|
|
assert response.status_code == 200
|
|
body = response.text
|
|
assert "/mnt/runbooks" in body
|
|
assert "12 discovered" in body and "3 new" in body
|
|
assert "runbooks/backup.md" in body
|
|
assert "Sync now" in body
|
|
assert "data-confirm" in body # destructive actions confirm
|
|
assert 'action="/admin/sources/' + str(source.id) + '/delete"' in body
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("outcome", "failed", "expected"),
|
|
[
|
|
(SourceSyncOutcome.RUNNING, 0, "Syncing"),
|
|
(SourceSyncOutcome.AWAITING_INGEST, 0, "Indexing child jobs"),
|
|
(SourceSyncOutcome.COMPLETED_WITH_FAILURES, 1, "Completed with failures"),
|
|
],
|
|
)
|
|
def test_source_sync_history_names_unfinished_and_incomplete_outcomes_truthfully(
|
|
outcome: SourceSyncOutcome, failed: int, expected: str
|
|
) -> None:
|
|
service = FakeSourceManagement()
|
|
source = _folder_source()
|
|
service.sources[source.id] = source
|
|
service.runs[source.id] = [
|
|
SourceSyncRunRecord(
|
|
id=uuid4(),
|
|
source_id=source.id,
|
|
started_at=datetime(2026, 8, 20, 10, 0, tzinfo=UTC),
|
|
finished_at=datetime(2026, 8, 20, 10, 1, tzinfo=UTC),
|
|
outcome=outcome,
|
|
discovered=3,
|
|
new=1,
|
|
changed=0,
|
|
unchanged=2,
|
|
removed=0,
|
|
failed=failed,
|
|
)
|
|
]
|
|
|
|
response = _fixture(service).get(f"/admin/sources/{source.id}")
|
|
|
|
assert response.status_code == 200
|
|
assert expected in response.text
|
|
|
|
|
|
def test_unknown_source_is_a_404_with_reference() -> None:
|
|
client = _fixture(FakeSourceManagement())
|
|
response = client.get(f"/admin/sources/{uuid4()}")
|
|
assert response.status_code == 404
|
|
assert "Source not found" in response.text
|
|
assert "Reference:" in response.text
|
|
|
|
|
|
def test_create_source_validates_and_redirects_to_detail() -> None:
|
|
service = FakeSourceManagement()
|
|
client = _fixture(service)
|
|
invalid = client.post(
|
|
"/admin/sources",
|
|
data={"name": "Bad", "kind": "folder", "space_id": str(SPACE_ID), "root": "relative/path"},
|
|
)
|
|
assert invalid.status_code == 400
|
|
assert "absolute" in invalid.text
|
|
assert 'role="alert"' in invalid.text
|
|
|
|
created = client.post(
|
|
"/admin/sources",
|
|
data={
|
|
"name": "Infra runbooks",
|
|
"kind": "folder",
|
|
"space_id": str(SPACE_ID),
|
|
"root": "/mnt/runbooks",
|
|
"include": "*.md",
|
|
"trust_level": "trusted_internal",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert created.status_code == 303
|
|
assert created.headers["location"].startswith("/admin/sources/")
|
|
assert service.created[0]["name"] == "Infra runbooks"
|
|
detail = client.get(created.headers["location"])
|
|
assert detail.status_code == 200
|
|
assert "Source created." in detail.text
|
|
|
|
|
|
def test_actions_call_the_service_and_upload_sources_cannot_sync() -> None:
|
|
service = FakeSourceManagement()
|
|
folder = _folder_source()
|
|
upload = SourceRecord(
|
|
id=uuid4(),
|
|
workspace_id=WORKSPACE_ID,
|
|
space_id=SPACE_ID,
|
|
name="Uploads",
|
|
kind=ConnectorKind.UPLOAD,
|
|
)
|
|
service.sources[folder.id] = folder
|
|
service.sources[upload.id] = upload
|
|
client = _fixture(service)
|
|
sync = client.post(f"/admin/sources/{folder.id}/sync", follow_redirects=False)
|
|
assert sync.status_code == 303 and "notice=sync_requested" in sync.headers["location"]
|
|
pause = client.post(f"/admin/sources/{folder.id}/pause", follow_redirects=False)
|
|
assert pause.status_code == 303
|
|
delete = client.post(f"/admin/sources/{folder.id}/delete", follow_redirects=False)
|
|
assert (
|
|
delete.status_code == 303 and delete.headers["location"] == "/admin/sources?notice=deleted"
|
|
)
|
|
assert [action for action, _ in service.actions] == ["sync", "paused", "deleted"]
|
|
refused = client.post(f"/admin/sources/{upload.id}/sync")
|
|
assert refused.status_code == 409
|
|
assert "fed by its callers" in refused.text
|
|
# Upload sources show no Sync button and no delete.
|
|
page = client.get(f"/admin/sources/{upload.id}")
|
|
assert "Sync now" not in page.text
|
|
|
|
|
|
def test_viewer_cannot_see_the_form_or_mutate_and_leaks_nothing() -> None:
|
|
service = FakeSourceManagement(manage=False)
|
|
source = _folder_source("Secret source")
|
|
service.sources[source.id] = source
|
|
client = _fixture(service, _principal(Role.VIEWER))
|
|
form = client.get("/admin/sources/new")
|
|
assert form.status_code == 403
|
|
assert "Secret source" not in form.text
|
|
mutate = client.post(f"/admin/sources/{source.id}/sync")
|
|
assert mutate.status_code == 403
|
|
listing = client.get("/admin/sources")
|
|
assert listing.status_code == 200
|
|
assert 'href="/admin/sources/new"' not in listing.text
|
|
|
|
|
|
def test_authorization_errors_render_forbidden_state() -> None:
|
|
class Denying(FakeSourceManagement):
|
|
async def list_sources(self, principal, filters):
|
|
raise AuthorizationError("denied")
|
|
|
|
client = _fixture(Denying())
|
|
response = client.get("/admin/sources")
|
|
assert response.status_code == 403
|
|
assert "Access denied" in response.text
|
|
|
|
|
|
# -- the Control Plane must not render "nothing to see" for unfinished work --
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("state", "expected"),
|
|
[
|
|
(SourceSyncState.AWAITING_INGEST, "Indexing"),
|
|
(SourceSyncState.COMPLETED_WITH_FAILURES, "Last sync incomplete"),
|
|
(SourceSyncState.COMPLETED, "Last sync ok"),
|
|
(SourceSyncState.FAILED, "Last sync failed"),
|
|
(SourceSyncState.SCANNING, "Syncing"),
|
|
],
|
|
)
|
|
def test_every_sync_state_renders_a_badge_on_both_screens(
|
|
state: SourceSyncState, expected: str
|
|
) -> None:
|
|
"""A state with no badge shows as an unqualified "Active" source, which an
|
|
operator reads as finished -- exactly the lie convergence removes."""
|
|
|
|
service = FakeSourceManagement()
|
|
source = replace(_folder_source(), sync_state=state)
|
|
service.sources[source.id] = source
|
|
|
|
client = _fixture(service)
|
|
|
|
assert expected in client.get("/admin/sources").text
|
|
assert expected in client.get(f"/admin/sources/{source.id}").text
|
|
|
|
|
|
def test_an_unconverged_item_says_what_to_retry_instead_of_hiding() -> None:
|
|
"""A terminally failed item whose content has not changed is deliberately
|
|
never re-enqueued by the next sync, so the screen has to name the recovery
|
|
path. Reporting only discovery counts would leave it reading as
|
|
"unchanged" forever while nothing was ever indexed."""
|
|
|
|
service = FakeSourceManagement()
|
|
source = _folder_source()
|
|
service.sources[source.id] = source
|
|
failed_job_id = uuid4()
|
|
service.failed_items[source.id] = [
|
|
SourceItemRecord(
|
|
source_id=source.id,
|
|
external_id="runbooks/broken.pdf",
|
|
fingerprint="a" * 64,
|
|
locator="/mnt/rag-sources/runbooks/broken.pdf",
|
|
last_seen_at=datetime(2026, 8, 20, 10, 0, tzinfo=UTC),
|
|
last_failure_code="embedding_failed",
|
|
last_failure_retryable=True,
|
|
last_failure_job_id=failed_job_id,
|
|
)
|
|
]
|
|
client = _fixture(service)
|
|
|
|
response = client.get(f"/admin/sources/{source.id}")
|
|
|
|
assert response.status_code == 200
|
|
body = response.text
|
|
assert "Ingest failed" in body
|
|
assert "runbooks/broken.pdf" in body
|
|
assert "embedding_failed" in body
|
|
assert "retry the failed ingest job" in body
|
|
assert f'href="/admin/jobs/{failed_job_id}"' in body
|
|
|
|
|
|
def test_failed_source_item_does_not_link_an_unreachable_recorded_job() -> None:
|
|
service = FakeSourceManagement()
|
|
source = _folder_source()
|
|
failed_job_id = uuid4()
|
|
service.sources[source.id] = source
|
|
service.failed_items[source.id] = [
|
|
SourceItemRecord(
|
|
source_id=source.id,
|
|
external_id="runbooks/missing.pdf",
|
|
fingerprint="b" * 64,
|
|
locator="/mnt/rag-sources/runbooks/missing.pdf",
|
|
last_seen_at=datetime(2026, 8, 20, 10, 0, tzinfo=UTC),
|
|
last_failure_code="parse_failed",
|
|
last_failure_retryable=True,
|
|
last_failure_job_id=failed_job_id,
|
|
)
|
|
]
|
|
|
|
response = _fixture(service, resolve_failed_jobs=False).get(
|
|
f"/admin/sources/{source.id}"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert f'href="/admin/jobs/{failed_job_id}"' not in response.text
|
|
assert "recorded job is unavailable" in response.text
|
|
|
|
|
|
def test_a_sync_interval_can_be_set_on_the_form_and_is_bounded() -> None:
|
|
"""Automatic syncing is opt-in per source, and the form is where an
|
|
operator opts in. Leaving it blank has to keep meaning "only when I ask"
|
|
-- that is the behaviour every source had before there was a scheduler."""
|
|
|
|
service = FakeSourceManagement()
|
|
client = _fixture(service)
|
|
|
|
without = client.post(
|
|
"/admin/sources",
|
|
data={
|
|
"name": "Manual only",
|
|
"kind": "folder",
|
|
"space_id": str(SPACE_ID),
|
|
"root": "/mnt/runbooks",
|
|
"sync_interval_minutes": "",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert without.status_code == 303
|
|
assert service.created[-1]["sync_interval_seconds"] is None
|
|
|
|
scheduled = client.post(
|
|
"/admin/sources",
|
|
data={
|
|
"name": "Hourly",
|
|
"kind": "folder",
|
|
"space_id": str(SPACE_ID),
|
|
"root": "/mnt/runbooks",
|
|
"sync_interval_minutes": "60",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert scheduled.status_code == 303
|
|
assert service.created[-1]["sync_interval_seconds"] == 3600
|
|
|
|
# One minute would spend the worker rediscovering unchanged content.
|
|
too_often = client.post(
|
|
"/admin/sources",
|
|
data={
|
|
"name": "Too often",
|
|
"kind": "folder",
|
|
"space_id": str(SPACE_ID),
|
|
"root": "/mnt/runbooks",
|
|
"sync_interval_minutes": "1",
|
|
},
|
|
)
|
|
assert too_often.status_code == 400
|
|
assert 'role="alert"' in too_often.text
|