v6ole
7455d7e426
130 issues auto-fixed (import ordering, UP045/UP006 type annotations), 33 issues manually fixed (E712/E501/E402/E722 + N818 rename + per-file wechat ignore for N8xx naming conventions). All 33 tests pass.
79 lines
2.1 KiB
Python
79 lines
2.1 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig, PipelineResult
|
|
|
|
|
|
class FakeSpider(BaseSpider):
|
|
name = "test_spider"
|
|
source_code = "test_source"
|
|
source_name = "测试来源"
|
|
|
|
async def crawl(self) -> CrawlResult:
|
|
return CrawlResult(
|
|
source_code=self.source_code,
|
|
source_name=self.source_name,
|
|
total_count=5,
|
|
new_count=3,
|
|
crawled_at=datetime.now(),
|
|
)
|
|
|
|
def get_pipeline_config(self) -> PipelineConfig:
|
|
return PipelineConfig(
|
|
filter_enabled=True,
|
|
keywords=["测试"],
|
|
dedup_enabled=True,
|
|
notify_mode="filtered",
|
|
mark_sent=False,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_base_spider_crawl():
|
|
spider = FakeSpider()
|
|
result = await spider.crawl()
|
|
assert result.source_code == "test_source"
|
|
assert result.total_count == 5
|
|
assert result.new_count == 3
|
|
assert result.success is True
|
|
|
|
|
|
def test_pipeline_config_defaults():
|
|
config = PipelineConfig()
|
|
assert config.filter_enabled is True
|
|
assert config.keywords == []
|
|
assert config.dedup_enabled is True
|
|
assert config.notify_mode == "filtered"
|
|
assert config.mark_sent is False
|
|
|
|
|
|
def test_pipeline_result_defaults():
|
|
result = PipelineResult()
|
|
assert result.stored == 0
|
|
assert result.filtered == 0
|
|
assert result.notified == 0
|
|
assert result.markdown_generated is False
|
|
|
|
|
|
def test_generate_content_hash():
|
|
h = BaseSpider.generate_content_hash(
|
|
"title", "2026-05-09", "unit", "https://x.com", "source"
|
|
)
|
|
assert len(h) == 64
|
|
assert all(c in "0123456789abcdef" for c in h)
|
|
|
|
|
|
def test_generate_content_hash_deterministic():
|
|
h1 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s")
|
|
h2 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s")
|
|
assert h1 == h2
|
|
|
|
|
|
def test_crawl_result_failure():
|
|
result = CrawlResult(
|
|
source_code="s", source_name="n",
|
|
error_message="连接超时"
|
|
)
|
|
assert result.success is False
|