534 lines
20 KiB
Python
534 lines
20 KiB
Python
"""M3 三层去重模块单元测试。"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from dedup import (
|
|
DEFAULT_HAMMING_THRESHOLD,
|
|
Deduper,
|
|
DedupLayer,
|
|
DedupResult,
|
|
Fingerprint,
|
|
FingerprintStore,
|
|
article_to_fingerprint,
|
|
content_hash,
|
|
hamming,
|
|
normalize_content,
|
|
simhash64,
|
|
)
|
|
from extractor.models import ProcessedArticle
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 辅助工厂函数
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _make_article(
|
|
*,
|
|
url: str = "https://www.reuters.com/business/1",
|
|
url_hash: str = "abc1234567890000",
|
|
source_id: str = "reuters",
|
|
source_name: str = "Reuters",
|
|
title: str = "Fed Holds Rates Steady as Markets Rally",
|
|
content: str = (
|
|
"The Federal Reserve held interest rates steady on Wednesday, "
|
|
"citing solid economic growth and a strong labor market. "
|
|
"Markets rallied in response, with the S&P 500 gaining 1.2 percent."
|
|
),
|
|
publish_time: str = "2026-06-16T10:00:00",
|
|
word_count: int = 0,
|
|
) -> ProcessedArticle:
|
|
return ProcessedArticle(
|
|
source_id=source_id,
|
|
source_name=source_name,
|
|
url=url,
|
|
url_hash=url_hash,
|
|
title=title,
|
|
content=content,
|
|
publish_time=publish_time,
|
|
word_count=word_count or len(content.split()),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_db(tmp_path: Path) -> Path:
|
|
return tmp_path / "fp.sqlite3"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# hasher 测试
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestNormalizeContent:
|
|
"""normalize_content 函数测试。"""
|
|
|
|
def test_strips_punctuation_and_whitespace(self):
|
|
norm = normalize_content("Hello, world!\nThis is text.")
|
|
assert norm == "HelloworldThisistext"
|
|
|
|
def test_handles_empty(self):
|
|
assert normalize_content("") == ""
|
|
assert normalize_content(" \n\t ") == ""
|
|
|
|
def test_preserves_letters_and_digits(self):
|
|
# $ 是货币符号（Sc 类别），会被保留；% 和 . 是标点（Po）会被移除
|
|
norm = normalize_content("AAPL up 5.2% to $150.00!")
|
|
assert norm == "AAPLup52to$15000"
|
|
|
|
def test_handles_chinese_characters(self):
|
|
norm = normalize_content("宁德时代 发布 新一代 麒麟电池！")
|
|
assert norm == "宁德时代发布新一代麒麟电池"
|
|
|
|
|
|
class TestContentHash:
|
|
"""content_hash 函数测试。"""
|
|
|
|
def test_deterministic(self):
|
|
a = "The Fed raised rates today."
|
|
b = "The Fed raised rates today."
|
|
assert content_hash(a) == content_hash(b)
|
|
|
|
def test_punctuation_invariant(self):
|
|
a = "Apple reports record earnings."
|
|
b = "Apple, reports... record earnings!!!"
|
|
assert content_hash(a) == content_hash(b)
|
|
|
|
def test_differs_for_different_text(self):
|
|
assert content_hash("Fed raises rates") != content_hash("Fed cuts rates")
|
|
|
|
|
|
class TestSimhash64:
|
|
"""simhash64 函数测试。"""
|
|
|
|
def test_identical_text_same_value(self):
|
|
text = "The Federal Reserve held interest rates steady on Wednesday."
|
|
assert simhash64(text) == simhash64(text)
|
|
|
|
def test_minor_changes_close_distance(self):
|
|
"""轻量改写，长文本汉明距离应在阈值内。"""
|
|
base = (
|
|
"The Federal Reserve held interest rates steady on Wednesday, "
|
|
"citing solid economic growth and a strong labor market. "
|
|
"Markets rallied in response, with the S&P 500 gaining 1.2 percent. "
|
|
"Analysts expect rates to remain unchanged through the summer."
|
|
) * 2
|
|
rewritten = "WASHINGTON (Reuters) - " + base + " (Reporting by John Smith)"
|
|
d = hamming(simhash64(base), simhash64(rewritten))
|
|
assert d <= DEFAULT_HAMMING_THRESHOLD, (
|
|
f"长文本前后加来源标识汉明距离 {d} 不应超过阈值"
|
|
)
|
|
|
|
def test_unrelated_text_far_distance(self):
|
|
"""完全不相关的两段长文本汉明距离应远大于阈值。"""
|
|
a = "The Federal Reserve held interest rates steady on Wednesday." * 5
|
|
b = "Apple announced a new iPhone model with revolutionary features." * 5
|
|
d = hamming(simhash64(a), simhash64(b))
|
|
assert d > DEFAULT_HAMMING_THRESHOLD * 2
|
|
|
|
def test_empty_returns_zero(self):
|
|
assert simhash64("") == 0
|
|
assert simhash64(" ") == 0
|
|
|
|
|
|
class TestHamming:
|
|
"""hamming 距离函数测试。"""
|
|
|
|
def test_same_value_zero(self):
|
|
assert hamming(0, 0) == 0
|
|
assert hamming(0xDEADBEEF, 0xDEADBEEF) == 0
|
|
|
|
def test_basic(self):
|
|
assert hamming(0xFF, 0x00) == 8
|
|
assert hamming(0xFF00FF00, 0x00FF00FF) == 32
|
|
|
|
def test_single_bit(self):
|
|
assert hamming(1, 0) == 1
|
|
assert hamming(1 << 63, 0) == 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# FingerprintStore 测试
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestFingerprintStore:
|
|
"""FingerprintStore 功能测试。"""
|
|
|
|
def test_upsert_and_get(self, tmp_db):
|
|
fp = Fingerprint(
|
|
url_hash="hash1",
|
|
content_hash="ch1",
|
|
simhash=0xDEADBEEFCAFEBABE,
|
|
source_id="reuters",
|
|
url="https://x/1",
|
|
title="Test Title",
|
|
publish_date="2026-06-16",
|
|
)
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(fp)
|
|
got = store.get_by_url_hash("hash1")
|
|
assert got is not None
|
|
assert got.content_hash == "ch1"
|
|
assert got.simhash == 0xDEADBEEFCAFEBABE
|
|
assert got.publish_date == "2026-06-16"
|
|
|
|
def test_upsert_replaces_existing(self, tmp_db):
|
|
base = Fingerprint(
|
|
url_hash="h",
|
|
content_hash="ch1",
|
|
simhash=1,
|
|
source_id="reuters",
|
|
url="u",
|
|
title="t",
|
|
)
|
|
updated = base.model_copy(update={"content_hash": "ch2", "simhash": 999})
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(base)
|
|
store.upsert(updated)
|
|
got = store.get_by_url_hash("h")
|
|
assert got is not None
|
|
assert got.content_hash == "ch2"
|
|
assert got.simhash == 999
|
|
assert store.count() == 1
|
|
|
|
def test_find_by_content_hash(self, tmp_db):
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(Fingerprint(
|
|
url_hash="h1", content_hash="ch", simhash=0,
|
|
source_id="reuters", url="u1", title="t1",
|
|
))
|
|
assert store.find_by_content_hash("ch") is not None
|
|
assert store.find_by_content_hash("nope") is None
|
|
|
|
def test_candidates_within_window(self, tmp_db):
|
|
with FingerprintStore(tmp_db) as store:
|
|
for d, h in [
|
|
("2026-05-01", "old"),
|
|
("2026-06-15", "near"),
|
|
("2026-07-30", "far"),
|
|
]:
|
|
store.upsert(Fingerprint(
|
|
url_hash=h, content_hash=h, simhash=0,
|
|
source_id="reuters", url=f"u/{h}", title=h, publish_date=d,
|
|
))
|
|
cands = store.candidates_for_simhash("2026-06-16", window_days=7)
|
|
url_hashes = sorted(c.url_hash for c in cands)
|
|
assert url_hashes == ["near"]
|
|
|
|
def test_candidates_no_date_returns_all(self, tmp_db):
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(Fingerprint(
|
|
url_hash="h1", content_hash="c1", simhash=0,
|
|
source_id="reuters", url="u", title="t", publish_date=None,
|
|
))
|
|
cands = store.candidates_for_simhash(None, 30)
|
|
assert len(cands) == 1
|
|
|
|
def test_simhash_high_bit_hex(self, tmp_db):
|
|
"""64 位 SimHash 高位为 1 时，hex 存取应保持无符号。"""
|
|
high = (1 << 63) | 0x1234
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(Fingerprint(
|
|
url_hash="h", content_hash="c", simhash=high,
|
|
source_id="reuters", url="u", title="t",
|
|
))
|
|
got = store.get_by_url_hash("h")
|
|
assert got is not None
|
|
assert got.simhash == high
|
|
|
|
def test_count_by_source(self, tmp_db):
|
|
with FingerprintStore(tmp_db) as store:
|
|
for i, src in enumerate(["reuters", "reuters", "cnbc"]):
|
|
store.upsert(Fingerprint(
|
|
url_hash=f"h{i}", content_hash=f"c{i}", simhash=i,
|
|
source_id=src, url=f"u{i}", title=f"t{i}",
|
|
))
|
|
counts = store.count_by_source()
|
|
assert counts == {"reuters": 2, "cnbc": 1}
|
|
|
|
def test_date_range(self, tmp_db):
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(Fingerprint(
|
|
url_hash="h1", content_hash="c1", simhash=0,
|
|
source_id="reuters", url="u1", title="t1",
|
|
publish_date="2026-06-10",
|
|
))
|
|
store.upsert(Fingerprint(
|
|
url_hash="h2", content_hash="c2", simhash=0,
|
|
source_id="cnbc", url="u2", title="t2",
|
|
publish_date="2026-06-20",
|
|
))
|
|
lo, hi = store.date_range()
|
|
assert lo == "2026-06-10"
|
|
assert hi == "2026-06-20"
|
|
|
|
def test_delete(self, tmp_db):
|
|
with FingerprintStore(tmp_db) as store:
|
|
store.upsert(Fingerprint(
|
|
url_hash="h1", content_hash="c1", simhash=0,
|
|
source_id="reuters", url="u1", title="t1",
|
|
))
|
|
store.delete("h1")
|
|
assert store.get_by_url_hash("h1") is None
|
|
assert store.count() == 0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# article_to_fingerprint 测试
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestArticleToFingerprint:
|
|
"""article_to_fingerprint 转换测试。"""
|
|
|
|
def test_fields(self):
|
|
art = _make_article()
|
|
fp = article_to_fingerprint(art)
|
|
assert fp.url_hash == art.url_hash
|
|
assert fp.simhash == simhash64(art.content)
|
|
assert fp.content_hash == content_hash(art.content)
|
|
assert fp.publish_date == "2026-06-16"
|
|
assert fp.source_id == "reuters"
|
|
|
|
def test_handles_empty_publish_time(self):
|
|
art = _make_article(publish_time="")
|
|
fp = article_to_fingerprint(art)
|
|
assert fp.publish_date is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Deduper 三层去重测试
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestDeduper:
|
|
"""Deduper 三层去重功能测试。"""
|
|
|
|
def test_first_article_is_unique(self, tmp_db):
|
|
art = _make_article()
|
|
with Deduper(db_path=tmp_db) as d:
|
|
result = d.ingest(art)
|
|
assert not result.is_duplicate
|
|
assert result.matched_layer is None
|
|
assert d.stats().total == 1
|
|
|
|
def test_layer1_url_hash(self, tmp_db):
|
|
"""同一 url_hash 直接命中 L1。"""
|
|
a1 = _make_article()
|
|
a2 = _make_article() # 同 url_hash 同 url
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert result.is_duplicate
|
|
assert result.matched_layer == DedupLayer.URL
|
|
assert d.stats().total == 1, "L1 命中应不写入新指纹"
|
|
|
|
def test_layer2_content_hash(self, tmp_db):
|
|
"""url 不同但 content 完全一致 → L2。"""
|
|
a1 = _make_article(url="https://a.com/1", url_hash="hash1aaaaaaaaaaa")
|
|
a2 = _make_article(url="https://b.com/2", url_hash="hash2bbbbbbbbbbb")
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert result.is_duplicate
|
|
assert result.matched_layer == DedupLayer.CONTENT
|
|
assert result.matched_url_hash == "hash1aaaaaaaaaaa"
|
|
|
|
def test_layer2_punctuation_difference_caught(self, tmp_db):
|
|
"""标点/空白差异不应阻止 L2 命中（normalize_content 应剥离）。"""
|
|
a1 = _make_article(
|
|
url="https://a/1", url_hash="aaaa",
|
|
content="The Fed raised rates today. Markets rallied strongly!"
|
|
)
|
|
a2 = _make_article(
|
|
url="https://b/2", url_hash="bbbb",
|
|
content="The Fed, raised... rates today!!! Markets -- rallied -- strongly."
|
|
)
|
|
assert content_hash(a1.content) == content_hash(a2.content)
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert result.matched_layer == DedupLayer.CONTENT
|
|
|
|
def test_layer3_simhash_minor_rewrite(self, tmp_db):
|
|
"""长文本 + 转载前后缀，落入 SimHash 层（贴近真实跨源转载场景）。"""
|
|
long_body = (
|
|
"The Federal Reserve held interest rates steady on Wednesday, "
|
|
"citing solid economic growth and a strong labor market. "
|
|
"Markets rallied in response, with the S&P 500 gaining 1.2 percent. "
|
|
"Treasury yields fell as investors welcomed the decision. "
|
|
"Analysts expect the central bank to remain on hold through September."
|
|
) * 2
|
|
rewritten = "By Reuters Staff - " + long_body + " (Additional reporting by Jane Doe)"
|
|
a1 = _make_article(url="https://a/1", url_hash="aaaaa", content=long_body)
|
|
a2 = _make_article(url="https://b/2", url_hash="bbbbb", content=rewritten)
|
|
# 必要前提：content_hash 不同（否则会被 L2 截胡）
|
|
assert content_hash(a1.content) != content_hash(a2.content)
|
|
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert result.is_duplicate
|
|
assert result.matched_layer == DedupLayer.SIMHASH
|
|
assert result.hamming_distance is not None
|
|
assert result.hamming_distance <= DEFAULT_HAMMING_THRESHOLD
|
|
|
|
def test_layer3_unrelated_articles_kept(self, tmp_db):
|
|
"""完全不相关文章不去重。"""
|
|
a1 = _make_article(
|
|
url="https://a/1", url_hash="aaaaa",
|
|
content="The Federal Reserve held interest rates steady on Wednesday." * 5,
|
|
)
|
|
a2 = _make_article(
|
|
url="https://b/2", url_hash="bbbbb",
|
|
content="Apple announced a new iPhone model with revolutionary features." * 5,
|
|
title="Apple Unveils New iPhone",
|
|
)
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert not result.is_duplicate
|
|
assert d.stats().total == 2
|
|
|
|
def test_layer3_outside_time_window_kept(self, tmp_db):
|
|
"""SimHash 相近，但 publish_date 距离过远（> 30 天）不去重。"""
|
|
body = (
|
|
"The Federal Reserve held interest rates steady on Wednesday, "
|
|
"citing solid economic growth and a strong labor market."
|
|
) * 3
|
|
a1 = _make_article(
|
|
url="https://a/1", url_hash="aaaa1", content=body,
|
|
publish_time="2026-01-01T09:00:00",
|
|
)
|
|
a2 = _make_article(
|
|
url="https://b/2", url_hash="bbbb2", content=body[:50] + body,
|
|
publish_time="2026-06-16T09:00:00",
|
|
)
|
|
assert content_hash(a1.content) != content_hash(a2.content)
|
|
with Deduper(db_path=tmp_db, time_window_days=30) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert not result.is_duplicate, "时间窗口外不应命中 SimHash"
|
|
|
|
def test_threshold_zero_only_exact_simhash(self, tmp_db):
|
|
"""阈值 0 → 仅当 SimHash 完全相同才视为重复（且会先被 L2 拦截）。"""
|
|
a1 = _make_article(
|
|
url="https://a/1", url_hash="aaaa1",
|
|
content="The Federal Reserve held rates steady on Wednesday."
|
|
)
|
|
a2 = _make_article(
|
|
url="https://b/2", url_hash="bbbb2",
|
|
content="The Federal Reserve held rates unchanged on Wednesday."
|
|
)
|
|
with Deduper(db_path=tmp_db, simhash_threshold=0) as d:
|
|
d.ingest(a1)
|
|
result = d.ingest(a2)
|
|
assert not result.is_duplicate
|
|
|
|
def test_ingest_same_source_mixed(self, tmp_db):
|
|
"""混合重复/不重复文章的同源摄入。"""
|
|
a1 = _make_article(url="https://a/1", url_hash="h1", content="Story A " * 10)
|
|
a2 = _make_article(url="https://a/2", url_hash="h2", content="Story B " * 10)
|
|
a3 = _make_article(url="https://a/3", url_hash="h3", content="Story B " * 10) # 同 a2
|
|
a4 = _make_article(url="https://a/4", url_hash="h4", content="Story C " * 10)
|
|
|
|
with Deduper(db_path=tmp_db) as d:
|
|
r1 = d.ingest(a1)
|
|
r2 = d.ingest(a2)
|
|
r3 = d.ingest(a3)
|
|
r4 = d.ingest(a4)
|
|
|
|
assert not r1.is_duplicate
|
|
assert not r2.is_duplicate
|
|
assert r3.is_duplicate # L2 命中
|
|
assert not r4.is_duplicate
|
|
assert d.stats().total == 3
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Deduper - check 不写入
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestDeduperCheck:
|
|
"""Deduper.check() 只读测试。"""
|
|
|
|
def test_check_does_not_write(self, tmp_db):
|
|
art = _make_article()
|
|
with Deduper(db_path=tmp_db) as d:
|
|
result = d.check(art)
|
|
assert not result.is_duplicate
|
|
assert d.stats().total == 0 # check 不应入库
|
|
|
|
def test_check_detects_duplicate_after_ingest(self, tmp_db):
|
|
a1 = _make_article()
|
|
a2 = _make_article()
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(a1)
|
|
result = d.check(a2)
|
|
assert result.is_duplicate
|
|
assert result.matched_layer == DedupLayer.URL
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Deduper - stats
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestDeduperStats:
|
|
"""统计信息测试。"""
|
|
|
|
def test_stats_aggregates_by_source(self, tmp_db):
|
|
with Deduper(db_path=tmp_db) as d:
|
|
d.ingest(_make_article(
|
|
source_id="reuters", url="https://a/1", url_hash="r000000000000001",
|
|
))
|
|
d.ingest(_make_article(
|
|
source_id="reuters", url="https://a/2", url_hash="r000000000000002",
|
|
content="Another completely different article about markets." * 10,
|
|
))
|
|
d.ingest(_make_article(
|
|
source_id="cnbc", url="https://b/1", url_hash="c000000000000001",
|
|
content="CNBC exclusive report on technology stocks." * 10,
|
|
))
|
|
stats = d.stats()
|
|
assert stats.total == 3
|
|
assert stats.by_source == {"reuters": 2, "cnbc": 1}
|
|
assert stats.earliest is not None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# DedupResult 测试
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestDedupResult:
|
|
"""DedupResult 模型测试。"""
|
|
|
|
def test_short_summary_unique(self):
|
|
r = DedupResult(url_hash="abc", is_duplicate=False)
|
|
assert r.short_summary() == "[UNIQUE] abc"
|
|
|
|
def test_short_summary_duplicate_url(self):
|
|
r = DedupResult(
|
|
url_hash="abc",
|
|
is_duplicate=True,
|
|
matched_layer=DedupLayer.URL,
|
|
matched_url_hash="xyz",
|
|
)
|
|
assert "[DUP/url]" in r.short_summary()
|
|
|
|
def test_short_summary_duplicate_simhash(self):
|
|
r = DedupResult(
|
|
url_hash="abc",
|
|
is_duplicate=True,
|
|
matched_layer=DedupLayer.SIMHASH,
|
|
matched_url_hash="xyz",
|
|
hamming_distance=2,
|
|
)
|
|
summary = r.short_summary()
|
|
assert "[DUP/simhash]" in summary
|
|
assert "hd=2" in summary
|