56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
"""Link source = curated URL list (one page per line, no crawl following)."""
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
import blocks
|
|
from pipeline import GenContext
|
|
from routes import _validate_source
|
|
|
|
|
|
# --- _validate_source: per-line validation ---------------------------------
|
|
|
|
def test_validate_source_link_multiline_ok():
|
|
_validate_source("link", "https://a.com/x\nhttp://b.com/y") # no raise
|
|
|
|
|
|
def test_validate_source_link_rejects_bad_line():
|
|
with pytest.raises(HTTPException):
|
|
_validate_source("link", "https://a.com/x\nftp://bad")
|
|
|
|
|
|
def test_validate_source_link_rejects_empty():
|
|
with pytest.raises(HTTPException):
|
|
_validate_source("link", " \n ")
|
|
|
|
|
|
# --- _prepare_source: load every line, all pages as content, no triage -----
|
|
|
|
async def test_prepare_source_link_loads_all_urls_as_content(testdb, tmp_path, monkeypatch):
|
|
topic = "T"
|
|
folder = tmp_path / "source"
|
|
folder.mkdir()
|
|
|
|
captured = {}
|
|
|
|
def fake_load(urls, target, *, cancelled=None):
|
|
captured["urls"] = list(urls)
|
|
for i, u in enumerate(urls):
|
|
(Path(target) / f"p{i}.txt").write_text(f"QUELLE: {u}\n\nInhalt {i}", encoding="utf-8")
|
|
return len(urls)
|
|
|
|
monkeypatch.setattr(blocks, "load_pages", fake_load)
|
|
monkeypatch.setattr(blocks, "_convert_pdfs", lambda f: None)
|
|
monkeypatch.setattr(blocks, "_crawl_done", lambda t: False)
|
|
monkeypatch.setattr(blocks, "_step_idx", lambda t, n: 0) # step list needs source.json — irrelevant here
|
|
|
|
ctx = GenContext(topic=topic, provider="claude", is_cancelled=lambda: False)
|
|
q = {"type": "link", "location": " https://a.com/x \n\nhttps://b.com/y\n", "spec": ""}
|
|
ok = await blocks._prepare_source(ctx, lambda *a, **k: None, {"arbeit": tmp_path}, q, folder, "")
|
|
|
|
assert ok is True
|
|
assert captured["urls"] == ["https://a.com/x", "https://b.com/y"] # split + strip + blanks dropped
|
|
content = await testdb.list_content(topic)
|
|
assert len(content) == 2 # every page kept, no relevance triage
|