42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.crawler import CrawlError, detect_rules
|
|
|
|
|
|
CACHE_DIR = Path(__file__).resolve().parents[1] / "fixtures" / "source_cache"
|
|
INDEX_PATH = CACHE_DIR / "index.json"
|
|
|
|
|
|
|
|
def cached_sources():
|
|
if not INDEX_PATH.exists():
|
|
return []
|
|
index = json.loads(INDEX_PATH.read_text())
|
|
return [record for record in index.get("sources", {}).values() if record.get("cached")]
|
|
|
|
|
|
@pytest.mark.cached_sources
|
|
@pytest.mark.parametrize("record", cached_sources(), ids=lambda record: record["name"])
|
|
def test_automatic_extraction_from_cached_source(record, record_extraction_result):
|
|
document = CACHE_DIR / record["filename"]
|
|
result = {"name": record["name"], "url": record["url"], "mode": record["mode"], "document": record["filename"], "events": 0, "status": "failed", "message": ""}
|
|
try:
|
|
rules, events, message = detect_rules(document.read_text(encoding="utf-8", errors="replace"), record["url"])
|
|
result.update({"events": len(events), "status": "passed" if events else "failed", "message": message, "item_selector": rules.get("item_selector", "")})
|
|
assert events, "Automatic extraction returned no events"
|
|
except CrawlError as exc:
|
|
result["message"] = str(exc)
|
|
pytest.fail(str(exc))
|
|
finally:
|
|
record_extraction_result(result)
|
|
|
|
|
|
@pytest.mark.cached_sources
|
|
def test_source_cache_exists():
|
|
assert INDEX_PATH.exists(), "No source cache exists. Run: python scripts/cache_source_documents.py"
|
|
assert cached_sources(), "The source cache index contains no successful HTML documents."
|
|
|