stack/tests/zot/test_merge.py

"""Merge-based dedupe: fidelity facets union into the keeper."""

import json
from pathlib import Path

import pytest

from zot.db import TYPE_MAP, Db
from zot.schema import create_db


@pytest.fixture
def dup_db(tmp_path: Path) -> Path:
    p = tmp_path / "zotero.sqlite"
    create_db(str(p)).close()
    with Db(str(p)) as db:
        keeper = db.create_item(TYPE_MAP["document"], key="AAAAAAAA")
        loser = db.create_item(TYPE_MAP["document"], key="BBBBBBBB")
        for iid in (keeper, loser):
            db.set_field(iid, "url", "https://example.org/x")
        db.set_field(keeper, "title", "X")
        db.set_field(loser, "title", "X (stale)")
        db.sync_tags(keeper, ["shared", "keeper-only"])
        db.sync_tags(loser, ["shared", "loser-only"])
        # loser-only child note
        nid = db.create_item(TYPE_MAP["note"], key="CCCCCCCC")
        db.con.execute(
            "INSERT INTO itemNotes (itemID, parentItemID, note, title) VALUES (?,?,?,?)",
            (nid, loser, "<p>loser note</p>", ""),
        )
        # identical note on both — must NOT be doubled
        for parent, key in ((keeper, "DDDDDDDD"), (loser, "EEEEEEEE")):
            dn = db.create_item(TYPE_MAP["note"], key=key)
            db.con.execute(
                "INSERT INTO itemNotes (itemID, parentItemID, note, title) VALUES (?,?,?,?)",
                (dn, parent, "<p>same on both</p>", ""),
            )
        # loser-only child attachment
        att_loser_only = db.create_item(TYPE_MAP["attachment"], key="FFFFFFFF")
        db.con.execute(
            "INSERT INTO itemAttachments"
            " (itemID, parentItemID, linkMode, contentType, path)"
            " VALUES (?,?,?,?,?)",
            (att_loser_only, loser, 0, "application/pdf", "loser-only.pdf"),
        )
        # identical attachment path on both — must NOT be doubled
        for parent, key in ((keeper, "GGGGGGGG"), (loser, "HHHHHHHH")):
            att = db.create_item(TYPE_MAP["attachment"], key=key)
            db.con.execute(
                "INSERT INTO itemAttachments"
                " (itemID, parentItemID, linkMode, contentType, path)"
                " VALUES (?,?,?,?,?)",
                (att, parent, 0, "application/pdf", "same-on-both.pdf"),
            )
        # collections: both share one, loser is also in a second one
        coll_shared = db.ensure_collection("Shared Collection")
        coll_loser_only = db.ensure_collection("Loser Only Collection")
        db.add_to_collection(keeper, collection_key=coll_shared)
        db.add_to_collection(loser, collection_key=coll_shared)
        db.add_to_collection(loser, collection_key=coll_loser_only)
        # itemRelations: loser has a relation the keeper lacks
        db.con.execute(
            "INSERT INTO relationPredicates (predicate) VALUES (?)",
            ("dc:relation",),
        )
        pred_id = db.con.execute(
            "SELECT predicateID FROM relationPredicates WHERE predicate=?",
            ("dc:relation",),
        ).fetchone()[0]
        db.con.execute(
            "INSERT INTO itemRelations (itemID, predicateID, object) VALUES (?,?,?)",
            (loser, pred_id, "http://zotero.org/users/local/items/OLDKEY01"),
        )
        db.commit()
    return p


def _tags(db: Db, item_id: int) -> set[str]:
    return {
        r[0]
        for r in db.con.execute(
            "SELECT t.name FROM itemTags it JOIN tags t ON t.tagID=it.tagID"
            " WHERE it.itemID=?",
            (item_id,),
        )
    }


def _notes(db: Db, item_id: int) -> list[str]:
    return [
        r[0]
        for r in db.con.execute(
            "SELECT note FROM itemNotes WHERE parentItemID=?", (item_id,)
        )
    ]


def _attachment_paths(db: Db, item_id: int) -> list[str]:
    return [
        r[0]
        for r in db.con.execute(
            "SELECT path FROM itemAttachments WHERE parentItemID=?", (item_id,)
        )
    ]


def _collection_names(db: Db, item_id: int) -> set[str]:
    return {
        r[0]
        for r in db.con.execute(
            "SELECT c.collectionName FROM collectionItems ci"
            " JOIN collections c ON c.collectionID=ci.collectionID"
            " WHERE ci.itemID=?",
            (item_id,),
        )
    }


class TestMergeInto:
    def test_unions_tags_and_adopts_missing_notes(self, dup_db):
        from zot.merge import merge_into

        with Db(str(dup_db)) as db:
            keeper = db.find_item_by_key("AAAAAAAA")
            loser = db.find_item_by_key("BBBBBBBB")
            merge_into(db, keeper, loser)
            db.commit()
            assert _tags(db, keeper) >= {"shared", "keeper-only", "loser-only"}
            notes = _notes(db, keeper)
            assert "<p>loser note</p>" in notes
            assert notes.count("<p>same on both</p>") == 1
            assert db.find_item_by_key("BBBBBBBB") is None

    def test_keeper_fields_win(self, dup_db):
        from zot.merge import merge_into

        with Db(str(dup_db)) as db:
            keeper = db.find_item_by_key("AAAAAAAA")
            loser = db.find_item_by_key("BBBBBBBB")
            merge_into(db, keeper, loser)
            row = db.con.execute(
                "SELECT idv.value FROM itemData id"
                " JOIN fields f ON f.fieldID=id.fieldID"
                " JOIN itemDataValues idv ON idv.valueID=id.valueID"
                " WHERE id.itemID=? AND f.fieldName='title'",
                (keeper,),
            ).fetchone()
            assert row[0] == "X"

    def test_adopts_missing_attachments_and_dedups_by_path(self, dup_db):
        from zot.merge import merge_into

        with Db(str(dup_db)) as db:
            keeper = db.find_item_by_key("AAAAAAAA")
            loser = db.find_item_by_key("BBBBBBBB")
            counts = merge_into(db, keeper, loser)
            db.commit()
            paths = _attachment_paths(db, keeper)
            assert "loser-only.pdf" in paths
            assert paths.count("same-on-both.pdf") == 1
            assert counts["attachments_adopted"] == 1

    def test_unions_collection_memberships(self, dup_db):
        from zot.merge import merge_into

        with Db(str(dup_db)) as db:
            keeper = db.find_item_by_key("AAAAAAAA")
            loser = db.find_item_by_key("BBBBBBBB")
            counts = merge_into(db, keeper, loser)
            db.commit()
            assert _collection_names(db, keeper) == {
                "Shared Collection",
                "Loser Only Collection",
            }
            # "Shared Collection" already had keeper as a member — only
            # the new one should count as added.
            assert counts["collections_added"] == 1

    def test_moves_relations_to_keeper(self, dup_db):
        from zot.merge import merge_into

        with Db(str(dup_db)) as db:
            keeper = db.find_item_by_key("AAAAAAAA")
            loser = db.find_item_by_key("BBBBBBBB")
            counts = merge_into(db, keeper, loser)
            db.commit()
            objects = {
                r[0]
                for r in db.con.execute(
                    "SELECT object FROM itemRelations WHERE itemID=?", (keeper,)
                )
            }
            assert "http://zotero.org/users/local/items/OLDKEY01" in objects
            assert counts["relations_moved"] == 1


class TestDriver:
    def test_dry_run_reports_without_deleting(self, dup_db, tmp_path):
        from zot.merge import merge_url_duplicates

        stats = merge_url_duplicates(dup_db, dry_run=True, log_dir=tmp_path)
        assert stats["clusters"] == 1
        with Db(str(dup_db)) as db:
            assert db.find_item_by_key("BBBBBBBB") is not None
        logs = list(tmp_path.glob("merge-log-*.json"))
        assert logs and json.loads(logs[0].read_text())[0]["keeper_key"] == "AAAAAAAA"

    def test_live_run_merges(self, dup_db, tmp_path):
        from zot.merge import merge_url_duplicates

        stats = merge_url_duplicates(dup_db, dry_run=False, log_dir=tmp_path)
        assert stats["merged"] == 1
        with Db(str(dup_db)) as db:
            assert db.find_item_by_key("BBBBBBBB") is None

    def test_type_mismatch_skipped(self, tmp_path):
        from zot.merge import merge_url_duplicates

        p = tmp_path / "z.sqlite"
        create_db(str(p)).close()
        with Db(str(p)) as db:
            a = db.create_item(TYPE_MAP["document"], key="AAAAAAAA")
            b = db.create_item(TYPE_MAP["journalArticle"], key="BBBBBBBB")
            for iid in (a, b):
                db.set_field(iid, "url", "https://example.org/y")
            db.commit()
        stats = merge_url_duplicates(p, dry_run=False, log_dir=tmp_path)
        assert stats["skipped_type_mismatch"] == 1
        with Db(str(p)) as db:
            assert db.find_item_by_key("BBBBBBBB") is not None

    def test_junk_urls_ignored(self, tmp_path):
        from zot.merge import merge_url_duplicates

        p = tmp_path / "z.sqlite"
        create_db(str(p)).close()
        with Db(str(p)) as db:
            a = db.create_item(TYPE_MAP["journalArticle"], key="AAAAAAAA")
            b = db.create_item(TYPE_MAP["journalArticle"], key="BBBBBBBB")
            for iid in (a, b):
                db.set_field(iid, "url", "997-998")
            db.commit()
        stats = merge_url_duplicates(p, dry_run=False, log_dir=tmp_path)
        assert stats["clusters"] == 0