126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""Tests for dev/scripts/dedupe_bib_legacy.py — merging legacy duplicate
|
|
rows inside bib.sqlite: Class C (same-url rows predating URL-dedup in
|
|
Store.upsert, #624/91c3306) and Class B (listserv re-sends stored as
|
|
separate items under distinct email:<Message-ID> urls, refs #665).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from bib.item import Source
|
|
from bib.store import Store
|
|
|
|
# Load the script as a module — it lives outside the src/ tree, mirroring
|
|
# the pattern in tests/dev/test_migrate_fr_txt.py (dev/ isn't an
|
|
# importable package).
|
|
_SCRIPT = (
|
|
Path(__file__).resolve().parents[2] / "dev" / "scripts" / "dedupe_bib_legacy.py"
|
|
)
|
|
_spec = importlib.util.spec_from_file_location("_dedupe_bib_legacy", _SCRIPT)
|
|
assert _spec and _spec.loader
|
|
_dedupe_bib_legacy = importlib.util.module_from_spec(_spec)
|
|
sys.modules["_dedupe_bib_legacy"] = _dedupe_bib_legacy
|
|
_spec.loader.exec_module(_dedupe_bib_legacy)
|
|
dedupe_bib = _dedupe_bib_legacy.dedupe_bib
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
s = Store(str(tmp_path / "bib.sqlite"))
|
|
yield s
|
|
s.close()
|
|
|
|
|
|
def test_url_group_keeps_earliest_and_unions_tags(store):
|
|
a = store.create(
|
|
Source(title="ACO REACH", url="https://x.test/reach"),
|
|
tags=["module:aco", "keep-me"],
|
|
)
|
|
# Second row with the same url — simulate a pre-#624 duplicate insert
|
|
# that predates Store.upsert's URL-based dedup.
|
|
con = store._con()
|
|
con.execute(
|
|
"INSERT INTO items (key, item_type, title, url, extra_json)"
|
|
" VALUES ('ZZZZZZZ2','source','ACO REACH','https://x.test/reach','{}')"
|
|
)
|
|
con.commit()
|
|
store.add_tag("ZZZZZZZ2", "loser-tag")
|
|
|
|
out = dedupe_bib(store, dry_run=False)
|
|
|
|
assert out["url_groups"] == 1
|
|
assert out["removed"] == 1
|
|
kept = store.get(a)
|
|
assert "loser-tag" in kept.tags
|
|
assert "keep-me" in kept.tags
|
|
assert (a, "ZZZZZZZ2") in out["pairs"]
|
|
with pytest.raises(KeyError):
|
|
store.get("ZZZZZZZ2")
|
|
|
|
|
|
def test_email_group_aliases_mids(store):
|
|
s1 = Source(title="iQIES Hold Times", url="email:aaa@x")
|
|
s1.abstract = "Same body."
|
|
s2 = Source(title="iQIES Hold Times", url="email:bbb@x")
|
|
s2.abstract = "Same body."
|
|
k1 = store.create(s1, tags=["source:email"])
|
|
k2 = store.create(s2, tags=["source:email"])
|
|
|
|
out = dedupe_bib(store, dry_run=False)
|
|
|
|
assert out["email_groups"] == 1
|
|
assert out["removed"] == 1
|
|
assert out["aliased"] == 1
|
|
assert (k1, k2) in out["pairs"]
|
|
# Read extra_json via raw SQL, not Item.to_row() — Source.to_row()
|
|
# only serializes its own declared fields (doc_type) and would
|
|
# silently drop alias_mids on a round-trip through the pydantic
|
|
# model (see dedupe_bib_legacy.py's docstring / email_ingest.py:180).
|
|
con = store._con()
|
|
row = con.execute("SELECT extra_json FROM items WHERE key = ?", (k1,)).fetchone()
|
|
ej = json.loads(row["extra_json"])
|
|
assert "bbb@x" in ej["alias_mids"]
|
|
with pytest.raises(KeyError):
|
|
store.get(k2)
|
|
|
|
|
|
def test_email_group_idempotent_second_run_adds_nothing(store):
|
|
s1 = Source(title="iQIES Hold Times", url="email:aaa@x")
|
|
s1.abstract = "Same body."
|
|
s2 = Source(title="iQIES Hold Times", url="email:bbb@x")
|
|
s2.abstract = "Same body."
|
|
k1 = store.create(s1, tags=["source:email"])
|
|
store.create(s2, tags=["source:email"])
|
|
|
|
dedupe_bib(store, dry_run=False)
|
|
out2 = dedupe_bib(store, dry_run=False)
|
|
|
|
assert out2["removed"] == 0
|
|
assert out2["aliased"] == 0
|
|
con = store._con()
|
|
row = con.execute("SELECT extra_json FROM items WHERE key = ?", (k1,)).fetchone()
|
|
ej = json.loads(row["extra_json"])
|
|
assert ej["alias_mids"] == ["bbb@x"]
|
|
|
|
|
|
def test_dry_run_touches_nothing(store):
|
|
s1 = Source(title="T", url="email:a@x")
|
|
s1.abstract = "B"
|
|
s2 = Source(title="T", url="email:b@x")
|
|
s2.abstract = "B"
|
|
store.create(s1, tags=["source:email"])
|
|
store.create(s2, tags=["source:email"])
|
|
|
|
out = dedupe_bib(store, dry_run=True)
|
|
|
|
assert out["removed"] == 0
|
|
assert out["email_groups"] == 1
|
|
assert out["pairs"] == []
|
|
assert len(store.list_items(query="T")) == 2
|