267 lines
8.9 KiB
Python
267 lines
8.9 KiB
Python
"""M1 爬虫模块测试"""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from crawler.crawler import (
|
|
ARTICLE_DELAY_SEC,
|
|
MAX_MEMORY_MB,
|
|
SOURCE_TIMEOUT_SEC,
|
|
compute_url_hash,
|
|
crawl_source,
|
|
)
|
|
from crawler.loader import get_source_by_id, load_sources
|
|
from crawler.models import ArticleItem, CrawlResult, SourceConfig
|
|
from crawler.storage import load_index, write_index_jsonl
|
|
|
|
# ════════════════════════════════════════════════
|
|
# URL Hash
|
|
# ════════════════════════════════════════════════
|
|
|
|
def test_compute_url_hash_consistency():
|
|
"""同一 URL 多次计算 hash 一致"""
|
|
h1 = compute_url_hash("https://example.com/article/123")
|
|
h2 = compute_url_hash("https://example.com/article/123")
|
|
assert h1 == h2
|
|
assert len(h1) == 16
|
|
|
|
|
|
def test_compute_url_hash_different():
|
|
"""不同 URL 产生不同 hash"""
|
|
h1 = compute_url_hash("https://example.com/a")
|
|
h2 = compute_url_hash("https://example.com/b")
|
|
assert h1 != h2
|
|
|
|
|
|
# ════════════════════════════════════════════════
|
|
# 配置加载
|
|
# ════════════════════════════════════════════════
|
|
|
|
def test_load_sources_from_real_config():
|
|
"""从项目真实配置文件加载"""
|
|
sources, settings = load_sources()
|
|
assert len(sources) == 12
|
|
assert settings["concurrency"] == 5
|
|
|
|
reuters = sources[0]
|
|
assert reuters.id == "reuters"
|
|
assert reuters.name == "Reuters"
|
|
assert reuters.enabled is True
|
|
|
|
|
|
def test_get_source_by_id():
|
|
"""按 ID 查找源"""
|
|
sources, _ = load_sources()
|
|
s = get_source_by_id("cnbc", sources)
|
|
assert s is not None
|
|
assert s.name == "CNBC"
|
|
|
|
s = get_source_by_id("nonexistent", sources)
|
|
assert s is None
|
|
|
|
|
|
def test_load_sources_missing_file():
|
|
"""配置文件不存在时抛出异常"""
|
|
with pytest.raises(FileNotFoundError):
|
|
load_sources(Path("/nonexistent/path.yaml"))
|
|
|
|
|
|
def test_source_config_output_dir():
|
|
"""SourceConfig.output_dir 属性"""
|
|
s = SourceConfig(
|
|
id="reuters",
|
|
name="Reuters",
|
|
homepage="https://example.com",
|
|
article_url_pattern="/article/",
|
|
)
|
|
today = __import__("datetime").datetime.now().strftime("%Y%m%d")
|
|
assert str(s.output_dir) == f"data/raw/reuters/{today}"
|
|
|
|
|
|
# ════════════════════════════════════════════════
|
|
# 存储
|
|
# ════════════════════════════════════════════════
|
|
|
|
def test_write_and_load_index_jsonl(tmp_path: Path, monkeypatch):
|
|
"""写入 index.jsonl 后再读取，数据一致"""
|
|
# 临时替换 data/raw 路径
|
|
import crawler.storage as storage_mod
|
|
|
|
articles = [
|
|
ArticleItem(
|
|
source_id="test_source",
|
|
source_name="Test Source",
|
|
url=f"https://example.com/article/{i}",
|
|
url_hash=f"hash{i:04d}",
|
|
title=f"Test Article {i}",
|
|
crawl_time="2026-06-21T00:00:00",
|
|
html_path=f"data/raw/test_source/20260621/hash{i:04d}.html",
|
|
status="success",
|
|
)
|
|
for i in range(3)
|
|
]
|
|
|
|
result = CrawlResult(
|
|
source_id="test_source",
|
|
source_name="Test Source",
|
|
total_found=3,
|
|
total_success=3,
|
|
articles=articles,
|
|
)
|
|
|
|
# Patch Path to use tmp_path
|
|
orig_path = Path
|
|
|
|
def mock_path(p: str) -> Path:
|
|
p_str = str(p)
|
|
if p_str.startswith("data/raw/"):
|
|
return orig_path(tmp_path) / p_str
|
|
return orig_path(p_str)
|
|
|
|
monkeypatch.setattr(storage_mod, "Path", mock_path)
|
|
|
|
index_path = write_index_jsonl(result)
|
|
assert index_path.exists()
|
|
|
|
# 读取
|
|
loaded = load_index("test_source", "20260621")
|
|
assert len(loaded) == 3
|
|
assert loaded[0].source_id == "test_source"
|
|
assert loaded[0].title == "Test Article 0"
|
|
|
|
|
|
def test_load_index_missing_file():
|
|
"""不存在的 index 返回空列表"""
|
|
articles = load_index("nonexistent", "20990101")
|
|
assert articles == []
|
|
|
|
|
|
def test_write_index_jsonl_dedup(tmp_path: Path, monkeypatch):
|
|
"""重复 url_hash 不重复写入"""
|
|
import crawler.storage as storage_mod
|
|
|
|
article = ArticleItem(
|
|
source_id="dedup_test",
|
|
source_name="Dedup Test",
|
|
url="https://example.com/same",
|
|
url_hash="same_hash_0001",
|
|
title="Same Article",
|
|
crawl_time="2026-06-21T00:00:00",
|
|
html_path="data/raw/dedup_test/20260621/same_hash_0001.html",
|
|
status="success",
|
|
)
|
|
|
|
result1 = CrawlResult(source_id="dedup_test", source_name="Dedup Test",
|
|
total_success=1, articles=[article])
|
|
result2 = CrawlResult(source_id="dedup_test", source_name="Dedup Test",
|
|
total_success=1, articles=[article])
|
|
|
|
def mock_path(p: str) -> Path:
|
|
p_str = str(p)
|
|
if p_str.startswith("data/raw/"):
|
|
return Path(tmp_path) / p_str
|
|
return Path(p_str)
|
|
|
|
monkeypatch.setattr(storage_mod, "Path", mock_path)
|
|
|
|
write_index_jsonl(result1)
|
|
write_index_jsonl(result2) # 重复写入
|
|
|
|
loaded = load_index("dedup_test", "20260621")
|
|
assert len(loaded) == 1 # 去重
|
|
|
|
|
|
# ════════════════════════════════════════════════
|
|
# 爬虫引擎 (Mock)
|
|
# ════════════════════════════════════════════════
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crawl_source_with_mock():
|
|
"""Mock Crawl4AI，测试 crawl_source 流程"""
|
|
source = SourceConfig(
|
|
id="mock_source",
|
|
name="Mock Source",
|
|
homepage="https://mock.example.com/",
|
|
article_url_pattern="/news/",
|
|
js_render=False,
|
|
max_articles_per_run=5,
|
|
)
|
|
|
|
# Mock Crawl4AI 的返回
|
|
mock_html = '<a href="/news/article1">Article 1</a><a href="/news/article2">Article 2</a>'
|
|
mock_result = MagicMock()
|
|
mock_result.success = True
|
|
mock_result.html = mock_html
|
|
mock_result.markdown = "# Test Article\n\nContent here."
|
|
mock_result.metadata = {"title": "Test Article"}
|
|
mock_result.error_message = ""
|
|
|
|
with patch("crawler.crawler.AsyncWebCrawler") as mock_crawler_cls:
|
|
mock_crawler = MagicMock()
|
|
mock_crawler.arun = AsyncMock(return_value=mock_result)
|
|
mock_crawler_cls.return_value.__aenter__ = AsyncMock(return_value=mock_crawler)
|
|
mock_crawler_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
result = await crawl_source(source)
|
|
|
|
assert result.source_id == "mock_source"
|
|
assert result.total_found == 2
|
|
assert result.total_success == 2
|
|
assert result.total_failed == 0
|
|
assert len(result.articles) == 2
|
|
|
|
# 每条 article 都有正确的 source_id
|
|
for article in result.articles:
|
|
assert article.source_id == "mock_source"
|
|
assert article.status == "success"
|
|
assert article.html_path
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_crawl_source_homepage_failure():
|
|
"""首页抓取失败时优雅降级"""
|
|
source = SourceConfig(
|
|
id="fail_source",
|
|
name="Fail Source",
|
|
homepage="https://fail.example.com/",
|
|
article_url_pattern="/news/",
|
|
js_render=False,
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.success = False
|
|
mock_result.html = ""
|
|
mock_result.error_message = "Connection timeout"
|
|
|
|
with patch("crawler.crawler.AsyncWebCrawler") as mock_crawler_cls:
|
|
mock_crawler = MagicMock()
|
|
mock_crawler.arun = AsyncMock(return_value=mock_result)
|
|
mock_crawler_cls.return_value.__aenter__ = AsyncMock(return_value=mock_crawler)
|
|
mock_crawler_cls.return_value.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
result = await crawl_source(source)
|
|
|
|
assert result.total_found == 0
|
|
assert result.total_success == 0
|
|
|
|
|
|
# ════════════════════════════════════════════════
|
|
# 资源限制常量
|
|
# ════════════════════════════════════════════════
|
|
|
|
def test_max_memory_mb():
|
|
"""内存限制 < 2000 MB"""
|
|
assert 0 < MAX_MEMORY_MB < 2000
|
|
|
|
|
|
def test_source_timeout_sec():
|
|
"""单源超时在合理范围"""
|
|
assert SOURCE_TIMEOUT_SEC >= 3600 # 至少 1 小时
|
|
|
|
|
|
def test_article_delay_sec():
|
|
"""文章间隔 ≥ 1 秒"""
|
|
assert ARTICLE_DELAY_SEC >= 1.0
|