stack/tests/bib/test_sync.py

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

from __future__ import annotations

import sqlite3

from bib.item import Download, Manual, Regulation, Rule, Source
from bib.sync import (
    _ensure_tag,
    _ensure_value,
    _item_to_zotero_fields,
    _now_iso,
    _set_field,
    _sync_tags,
    _zotero_key,
    ensure_collection,
    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)
);
"""


def _make_zotero_db(path: str = ":memory:") -> sqlite3.Connection:
    con = sqlite3.connect(path)
    con.row_factory = sqlite3.Row
    con.executescript(ZOTERO_SCHEMA)
    return con


# ── _zotero_key ─────────────────────────────────────────────────────


class TestZoteroKey:
    def test_length(self) -> None:
        key = _zotero_key()
        assert len(key) == 8

    def test_alphanumeric(self) -> None:
        key = _zotero_key()
        assert key.isalnum()

    def test_uppercase(self) -> None:
        key = _zotero_key()
        assert key == key.upper()


# ── _now_iso ────────────────────────────────────────────────────────


class TestNowIso:
    def test_format(self) -> None:
        result = _now_iso()
        assert len(result) == 19  # "YYYY-MM-DD HH:MM:SS"
        assert " " in result


# ── _ensure_value ───────────────────────────────────────────────────


class TestEnsureValue:
    def test_creates_new(self) -> None:
        con = _make_zotero_db()
        vid = _ensure_value(con, "test value")
        assert isinstance(vid, int)
        assert vid > 0

    def test_finds_existing(self) -> None:
        con = _make_zotero_db()
        v1 = _ensure_value(con, "same")
        v2 = _ensure_value(con, "same")
        assert v1 == v2

    def test_different_values(self) -> None:
        con = _make_zotero_db()
        v1 = _ensure_value(con, "a")
        v2 = _ensure_value(con, "b")
        assert v1 != v2


# ── _ensure_tag ─────────────────────────────────────────────────────


class TestEnsureTag:
    def test_creates_new(self) -> None:
        con = _make_zotero_db()
        tid = _ensure_tag(con, "module:pfs")
        assert isinstance(tid, int)
        assert tid > 0

    def test_finds_existing(self) -> None:
        con = _make_zotero_db()
        t1 = _ensure_tag(con, "module:pfs")
        t2 = _ensure_tag(con, "module:pfs")
        assert t1 == t2


# ── _set_field ──────────────────────────────────────────────────────


class TestSetField:
    def test_sets_known_field(self) -> None:
        con = _make_zotero_db()
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (36, 'TESTKEY1')")
        item_id = con.execute(
            "SELECT itemID FROM items WHERE key = 'TESTKEY1'"
        ).fetchone()[0]

        _set_field(con, item_id, "title", "Test Title")

        row = con.execute(
            "SELECT idv.value FROM itemData id "
            "JOIN itemDataValues idv ON id.valueID = idv.valueID "
            "WHERE id.itemID = ? AND id.fieldID = 1",
            (item_id,),
        ).fetchone()
        assert row[0] == "Test Title"

    def test_skips_empty_value(self) -> None:
        con = _make_zotero_db()
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (36, 'TESTKEY1')")
        item_id = con.execute(
            "SELECT itemID FROM items WHERE key = 'TESTKEY1'"
        ).fetchone()[0]

        _set_field(con, item_id, "title", "")

        row = con.execute(
            "SELECT count(*) FROM itemData WHERE itemID = ?",
            (item_id,),
        ).fetchone()
        assert row[0] == 0

    def test_skips_unknown_field(self) -> None:
        con = _make_zotero_db()
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (36, 'TESTKEY1')")
        item_id = con.execute(
            "SELECT itemID FROM items WHERE key = 'TESTKEY1'"
        ).fetchone()[0]

        _set_field(con, item_id, "nonexistent_field", "value")

        row = con.execute(
            "SELECT count(*) FROM itemData WHERE itemID = ?",
            (item_id,),
        ).fetchone()
        assert row[0] == 0


# ── _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 fields["seriesNumber"] == "Chapter 12"
        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 fields["seriesNumber"] == ""

    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 fields["type"] == "guidance"
        assert fields["publisher"] == "CMS"


# ── _sync_tags ──────────────────────────────────────────────────────


class TestSyncTags:
    def test_adds_tags(self) -> None:
        con = _make_zotero_db()
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (36, 'TESTKEY1')")
        item_id = con.execute(
            "SELECT itemID FROM items WHERE key = 'TESTKEY1'"
        ).fetchone()[0]

        _sync_tags(con, item_id, ["module:pfs", "year:2026"])

        rows = con.execute(
            "SELECT t.name FROM itemTags it "
            "JOIN tags t ON it.tagID = t.tagID "
            "WHERE it.itemID = ?",
            (item_id,),
        ).fetchall()
        names = {r[0] for r in rows}
        assert "module:pfs" in names
        assert "year:2026" in names

    def test_skips_empty_tags(self) -> None:
        con = _make_zotero_db()
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (36, 'TESTKEY1')")
        item_id = con.execute(
            "SELECT itemID FROM items WHERE key = 'TESTKEY1'"
        ).fetchone()[0]

        _sync_tags(con, item_id, ["module:pfs", "", "year:2026"])

        rows = con.execute(
            "SELECT count(*) FROM itemTags WHERE itemID = ?",
            (item_id,),
        ).fetchone()
        assert rows[0] == 2


# ── ensure_collection ───────────────────────────────────────────────


class TestEnsureCollection:
    def test_creates_new(self) -> None:
        con = _make_zotero_db()
        key = ensure_collection(con, "Test Collection")
        assert isinstance(key, str)
        assert len(key) == 8

        row = con.execute(
            "SELECT collectionName FROM collections WHERE key = ?",
            (key,),
        ).fetchone()
        assert row[0] == "Test Collection"

    def test_finds_existing(self) -> None:
        con = _make_zotero_db()
        k1 = ensure_collection(con, "Test")
        k2 = ensure_collection(con, "Test")
        assert k1 == k2

    def test_with_parent(self) -> None:
        con = _make_zotero_db()
        parent_key = ensure_collection(con, "Parent")
        child_key = ensure_collection(con, "Child", parent_key=parent_key)
        assert parent_key != child_key

        # Creating child again should find existing
        child_key2 = ensure_collection(con, "Child", parent_key=parent_key)
        assert child_key == child_key2

    def test_parent_key_not_found(self) -> None:
        con = _make_zotero_db()
        # Parent key doesn't exist — parent_id will be None
        key = ensure_collection(con, "Orphan", parent_key="NOEXIST1")
        assert isinstance(key, str)


# ── 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="RULEKEY1",
                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)
        # Create a collection
        con.execute(
            "INSERT INTO collections (collectionName, libraryID, key, version, synced) "
            "VALUES ('Test', 1, 'COLLKEY1', 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="COLLKEY1")
        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, 'ACOKEY12', 0, 0)"
        )
        con.commit()
        con.close()

        items = [
            Rule(
                title="R1",
                url="https://ex.com/r1",
                collections=["ACOKEY12"],
            )
        ]
        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)
        # Pre-insert an item with the same key
        con.execute("INSERT INTO items (itemTypeID, key) VALUES (36, 'RULEKEY1')")
        con.commit()
        con.close()

        items = [Rule(key="RULEKEY1", 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="NOEXIST1")
        # Collection not found, so no collection assignment
        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=["NOEXIST1"])]
        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, 'MAINKEY1', 0, 0)"
        )
        con.commit()
        con.close()

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