stack/tests/bib/test_sync.py

"""Tests for bib.sync — push bib items into Zotero SQLite.

ORM primitives (key gen, ensure_value, ensure_tag, set_field, sync_tags,
ensure_collection) are tested in ``tests/zot/test_db.py``.  This file
tests the bib→Zotero adapter: field mapping and push_to_zotero.
"""

from __future__ import annotations

import sqlite3

from bib.item import Download, Manual, Regulation, Rule, Source
from bib.sync import _item_to_zotero_fields, push_to_zotero

# ── Zotero schema for tests ─────────────────────────────────────────

ZOTERO_SCHEMA = """
CREATE TABLE IF NOT EXISTS items (
    itemID INTEGER PRIMARY KEY AUTOINCREMENT,
    itemTypeID INTEGER NOT NULL,
    dateAdded TEXT,
    dateModified TEXT,
    clientDateModified TEXT,
    libraryID INTEGER DEFAULT 1,
    key TEXT NOT NULL UNIQUE,
    version INTEGER DEFAULT 0,
    synced INTEGER DEFAULT 0
);

CREATE TABLE IF NOT EXISTS itemDataValues (
    valueID INTEGER PRIMARY KEY AUTOINCREMENT,
    value TEXT NOT NULL UNIQUE
);

CREATE TABLE IF NOT EXISTS itemData (
    itemID INTEGER NOT NULL,
    fieldID INTEGER NOT NULL,
    valueID INTEGER NOT NULL,
    PRIMARY KEY (itemID, fieldID)
);

CREATE TABLE IF NOT EXISTS tags (
    tagID INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL UNIQUE
);

CREATE TABLE IF NOT EXISTS itemTags (
    itemID INTEGER NOT NULL,
    tagID INTEGER NOT NULL,
    type INTEGER DEFAULT 0,
    PRIMARY KEY (itemID, tagID)
);

CREATE TABLE IF NOT EXISTS collections (
    collectionID INTEGER PRIMARY KEY AUTOINCREMENT,
    collectionName TEXT NOT NULL,
    parentCollectionID INTEGER,
    clientDateModified TEXT,
    libraryID INTEGER DEFAULT 1,
    key TEXT NOT NULL UNIQUE,
    version INTEGER DEFAULT 0,
    synced INTEGER DEFAULT 0
);

CREATE TABLE IF NOT EXISTS collectionItems (
    collectionID INTEGER NOT NULL,
    itemID INTEGER NOT NULL,
    orderIndex INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (collectionID, itemID)
);

CREATE TABLE IF NOT EXISTS creators (
    creatorID INTEGER PRIMARY KEY AUTOINCREMENT,
    firstName TEXT,
    lastName TEXT,
    fieldMode INT,
    UNIQUE (lastName, firstName, fieldMode)
);

CREATE TABLE IF NOT EXISTS itemCreators (
    itemID INT NOT NULL,
    creatorID INT NOT NULL,
    creatorTypeID INT NOT NULL DEFAULT 1,
    orderIndex INT NOT NULL DEFAULT 0,
    PRIMARY KEY (itemID, creatorID, creatorTypeID, orderIndex),
    UNIQUE (itemID, orderIndex)
);

CREATE TABLE IF NOT EXISTS itemTypes (
    itemTypeID INTEGER PRIMARY KEY,
    typeName TEXT
);

CREATE TABLE IF NOT EXISTS fields (
    fieldID INTEGER PRIMARY KEY,
    fieldName TEXT
);

CREATE TABLE IF NOT EXISTS creatorTypes (
    creatorTypeID INTEGER PRIMARY KEY,
    creatorType TEXT
);
"""


def _make_zotero_db(path: str = ":memory:") -> sqlite3.Connection:
    from zot.db import CREATOR_TYPES, FIELD_MAP, TYPE_MAP

    con = sqlite3.connect(path)
    con.row_factory = sqlite3.Row
    con.executescript(ZOTERO_SCHEMA)
    for name, tid in TYPE_MAP.items():
        con.execute("INSERT OR IGNORE INTO itemTypes VALUES (?, ?)", (tid, name))
    for name, fid in FIELD_MAP.items():
        con.execute("INSERT OR IGNORE INTO fields VALUES (?, ?)", (fid, name))
    for name, cid in CREATOR_TYPES.items():
        con.execute("INSERT OR IGNORE INTO creatorTypes VALUES (?, ?)", (cid, name))
    con.commit()
    return con


# ── _item_to_zotero_fields ─────────────────────────────────────────


class TestItemToZoteroFields:
    def test_rule(self) -> None:
        item = Rule(
            title="PFS Final Rule",
            fr_volume="90",
            fr_page="98452",
            document_number="2025-19787",
            cms_id="CMS-1832-F",
            rule_type="final",
            date_published="2025-11-01",
            effective_date="2026-01-01",
            url="https://example.com/rule",
            abstract="Rule abstract",
        )
        fields = _item_to_zotero_fields(item)
        assert fields["nameOfAct"] == "PFS Final Rule"
        assert fields["code"] == "FR"
        assert fields["codeNumber"] == "90"
        assert fields["pages"] == "98452"
        assert fields["session"] == "CMS-1832-F"
        assert "Document: 2025-19787" in fields["history"]
        assert "Type: final" in fields["history"]
        assert "Effective: 2026-01-01" in fields["history"]

    def test_regulation(self) -> None:
        item = Regulation(
            title="42 CFR Part 414",
            cfr_title="42",
            cfr_part="414",
            cfr_section="414.22",
            authority="42 USC 1395w-4",
            effective_date="2025-01-01",
            url="https://ecfr.gov/414",
        )
        fields = _item_to_zotero_fields(item)
        assert fields["code"] == "C.F.R."
        assert fields["codeNumber"] == "42"
        assert fields["section"] == "414.22"
        assert "Part 414" in fields["history"]
        assert "Authority: 42 USC 1395w-4" in fields["history"]

    def test_regulation_no_authority(self) -> None:
        item = Regulation(cfr_title="42", cfr_part="414")
        fields = _item_to_zotero_fields(item)
        assert fields["history"] == "Part 414"
        assert "Authority" not in fields["history"]

    def test_manual(self) -> None:
        item = Manual(
            title="Chapter 12",
            manual_name="Claims Processing Manual",
            pub_number="100-04",
            chapter="12",
            transmittal="R100",
            institution="CMS",
            date_published="2025-01-01",
            url="https://cms.gov/manual",
        )
        fields = _item_to_zotero_fields(item)
        assert fields["reportType"] == "Internet-Only Manual"
        assert fields["reportNumber"] == "100-04"
        assert fields["seriesTitle"] == "Claims Processing Manual"
        assert "Chapter 12" in fields["extra"]
        assert "Transmittal: R100" in fields["extra"]
        assert fields["place"] == "Baltimore, MD"

    def test_manual_no_transmittal(self) -> None:
        item = Manual(title="Test", chapter="5")
        fields = _item_to_zotero_fields(item)
        assert "Transmittal" not in fields["extra"]

    def test_manual_no_chapter(self) -> None:
        item = Manual(title="Test")
        fields = _item_to_zotero_fields(item)
        assert "seriesNumber" not in fields

    def test_download(self) -> None:
        item = Download(
            title="RVU26A",
            file_urls=["https://cms.gov/rvu.zip"],
            date_published="2026-01-01",
            url="https://cms.gov/rvu26a",
        )
        fields = _item_to_zotero_fields(item)
        assert fields["title"] == "RVU26A"
        assert fields["websiteType"] == "Government Data Portal"
        assert "Files: https://cms.gov/rvu.zip" in fields["extra"]

    def test_download_no_files(self) -> None:
        item = Download(title="Test", url="https://cms.gov/test")
        fields = _item_to_zotero_fields(item)
        assert "Files:" not in fields.get("extra", "")

    def test_source(self) -> None:
        item = Source(
            title="Test Doc",
            doc_type="guidance",
            institution="CMS",
            date_published="2025-01-01",
            url="https://cms.gov/doc",
        )
        fields = _item_to_zotero_fields(item)
        assert fields["title"] == "Test Doc"
        assert "Type: guidance" in fields["extra"]
        assert fields["publisher"] == "CMS"


# ── push_to_zotero ──────────────────────────────────────────────────


class TestPushToZotero:
    def test_push_rule(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [
            Rule(
                key="RULEKEY2",
                title="PFS Final Rule",
                fr_volume="90",
                fr_page="98452",
                url="https://example.com/rule",
                tags=["module:pfs"],
            )
        ]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 1
        assert stats["skipped"] == 0
        assert stats["tags"] == 1

    def test_push_all_types(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [
            Rule(title="Rule", url="https://ex.com/rule"),
            Manual(title="Manual", url="https://ex.com/manual"),
            Regulation(title="Reg", url="https://ex.com/reg"),
            Download(title="DL", url="https://ex.com/dl"),
            Source(title="Src", url="https://ex.com/src"),
        ]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 5

    def test_skip_existing_url(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [Rule(title="First", url="https://ex.com/rule")]
        push_to_zotero(items, zotero_db=db_path)

        items2 = [Rule(title="Second", url="https://ex.com/rule", tags=["new-tag"])]
        stats = push_to_zotero(items2, zotero_db=db_path)
        assert stats["skipped"] == 1
        assert stats["created"] == 0

    def test_skip_unknown_type(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        from bib.item import Item

        items = [Item(item_type="unknown", title="Unknown")]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["skipped"] == 1

    def test_with_collection(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.execute(
            "INSERT INTO collections (collectionName, libraryID, key, version, synced) "
            "VALUES ('Test', 1, 'CKEY2345', 0, 0)"
        )
        con.commit()
        con.close()

        items = [Rule(title="R1", url="https://ex.com/r1")]
        stats = push_to_zotero(items, zotero_db=db_path, collection_key="CKEY2345")
        assert stats["collections"] == 1

    def test_item_collections(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.execute(
            "INSERT INTO collections (collectionName, libraryID, key, version, synced) "
            "VALUES ('ACO', 1, 'ACKEY234', 0, 0)"
        )
        con.commit()
        con.close()

        items = [
            Rule(
                title="R1",
                url="https://ex.com/r1",
                collections=["ACKEY234"],
            )
        ]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 1

    def test_key_collision(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (20, 'RKEY2345')")
        con.commit()
        con.close()

        items = [Rule(key="RKEY2345", title="Collision")]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 1

    def test_no_url_item(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [Rule(title="No URL")]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 1

    def test_short_key_generates_new(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [Rule(key="SHORT", title="Short Key")]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 1

    def test_collection_key_not_found(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [Rule(title="R1")]
        stats = push_to_zotero(items, zotero_db=db_path, collection_key="ZZZZ2345")
        assert stats["collections"] == 0
        assert stats["created"] == 1

    def test_item_collection_not_found(self, tmp_path) -> None:
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.close()

        items = [Rule(title="R1", collections=["ZZZZ2345"])]
        stats = push_to_zotero(items, zotero_db=db_path)
        assert stats["created"] == 1

    def test_item_collection_same_as_main(self, tmp_path) -> None:
        """If item collection matches collection_key, skip duplicate."""
        db_path = str(tmp_path / "zotero.sqlite")
        con = _make_zotero_db(db_path)
        con.execute(
            "INSERT INTO collections (collectionName, libraryID, key, version, synced) "
            "VALUES ('Main', 1, 'MKEY2345', 0, 0)"
        )
        con.commit()
        con.close()

        items = [
            Rule(
                title="R1",
                url="https://ex.com/r1",
                collections=["MKEY2345"],
            )
        ]
        stats = push_to_zotero(items, zotero_db=db_path, collection_key="MKEY2345")
        assert stats["collections"] == 1