claudia/tests/test_pipeline.py

"""End-to-end pipeline test for Claudia runner and components."""

from __future__ import annotations

from claudia.db import MemoryDB
from claudia.reporter import MemoryReporter
from claudia.scheduler import Runner
from claudia.spiders import ExampleSpider


def test_pipeline_end_to_end():
    """Run an example spider through the runner and assert outcomes."""
    spider = ExampleSpider()
    db = MemoryDB()
    reporter = MemoryReporter()
    runner = Runner(db=db, reporters=[reporter])

    urls = [
        "https://example.com/a",
        "https://example.com/b",
    ]
    summary = runner.run({spider: urls})

    # 插入应等于解析产出（每个 URL 一个文档）
    assert summary.total_docs == 2
    assert summary.per_spider.get(spider.name) == 2
    assert not summary.errors

    # reporter 至少包含 start/success/summary 事件
    assert any(e.startswith("start:") for e in reporter.events)
    assert any(e.startswith("success:") for e in reporter.events)
    assert any(e.startswith("summary:") for e in reporter.events)