kicad-mcp-server/tests/test_jlcpcb_downloader.py

"""
Regression tests for the JLCPCB prebuilt-catalog downloader (issue #199).

These tests do NOT hit the network. They build a small SQLite source that
mimics CDFER's schema (incl. the v_components view and bare-integer lcsc) and
verify that convert_source_sqlite() produces a database in the schema that
JLCPCBPartsManager consumes, plus that download_database() falls through
sources gracefully when none are usable.
"""

import json
import sqlite3
import sys
from pathlib import Path

import pytest

# Match the import root used elsewhere in the test suite (python/ on sys.path).
PYTHON_DIR = Path(__file__).resolve().parent.parent / "python"
if str(PYTHON_DIR) not in sys.path:
    sys.path.insert(0, str(PYTHON_DIR))

from commands import jlcpcb_downloader  # noqa: E402


def _make_cdfer_like_source(path: Path) -> None:
    """Create a tiny SQLite mirroring CDFER: components + categories + manufacturers + view."""
    con = sqlite3.connect(str(path))
    con.executescript("""
        CREATE TABLE manufacturers (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE categories (id INTEGER PRIMARY KEY, category TEXT, subcategory TEXT);
        CREATE TABLE components (
            lcsc INTEGER PRIMARY KEY,
            category_id INTEGER,
            mfr TEXT,
            package TEXT,
            joints INTEGER,
            manufacturer_id INTEGER,
            basic INTEGER,
            preferred INTEGER,
            description TEXT,
            datasheet TEXT,
            stock INTEGER,
            price TEXT
        );
        CREATE VIEW v_components AS
            SELECT c.lcsc, c.category_id, cat.category, cat.subcategory, c.mfr,
                   c.package, c.joints, m.name AS manufacturer, c.basic, c.preferred,
                   c.description, c.datasheet, c.stock, c.price
            FROM components c
            JOIN categories cat ON cat.id = c.category_id
            JOIN manufacturers m ON m.id = c.manufacturer_id;
        """)
    con.execute("INSERT INTO manufacturers VALUES (1, 'Yageo')")
    con.execute("INSERT INTO categories VALUES (1, 'Resistors', 'Chip Resistor')")
    # basic part, with CDFER-style price JSON (qFrom/qTo/price)
    con.execute(
        "INSERT INTO components VALUES (25804, 1, 'RC0603FR-071KL', '0603', 2, 1, 1, 0, "
        "'1k 1% 0603 resistor', 'http://ds/25804', 5000, ?)",
        (
            json.dumps(
                [
                    {"qFrom": 1, "qTo": 99, "price": 0.0042},
                    {"qFrom": 100, "qTo": None, "price": 0.0021},
                ]
            ),
        ),
    )
    # extended part, empty price
    con.execute(
        "INSERT INTO components VALUES (11111, 1, 'EXT-PART', '0402', 2, 1, 0, 0, "
        "'extended resistor', '', 12, '[]')"
    )
    con.commit()
    con.close()


def test_convert_source_sqlite_produces_manager_schema(tmp_path):
    source = tmp_path / "cdfer.sqlite3"
    target = tmp_path / "jlcpcb_parts.db"
    _make_cdfer_like_source(source)

    stats = jlcpcb_downloader.convert_source_sqlite(source, target)

    assert stats["total"] == 2
    assert stats["basic"] == 1
    assert stats["extended"] == 1

    con = sqlite3.connect(str(target))
    con.row_factory = sqlite3.Row
    rows = {r["lcsc"]: dict(r) for r in con.execute("SELECT * FROM components")}

    # bare-int lcsc must become C-prefixed
    assert "C25804" in rows and "C11111" in rows

    basic = rows["C25804"]
    assert basic["library_type"] == "Basic"
    assert basic["mfr_part"] == "RC0603FR-071KL"  # mfr -> mfr_part
    assert basic["manufacturer"] == "Yageo"  # via v_components view
    assert basic["category"] == "Resistors"
    assert basic["subcategory"] == "Chip Resistor"
    assert basic["stock"] == 5000

    # CDFER price JSON normalized to [{"qty","price"}]
    breaks = json.loads(basic["price_json"])
    assert breaks[0] == {"qty": 1, "price": 0.0042}
    assert breaks[1] == {"qty": 100, "price": 0.0021}

    assert rows["C11111"]["library_type"] == "Extended"
    assert json.loads(rows["C11111"]["price_json"]) == []

    # FTS index exists and is queryable
    fts_hit = con.execute(
        "SELECT lcsc FROM components_fts WHERE components_fts MATCH '1k*'"
    ).fetchall()
    assert any(r[0] == "C25804" for r in fts_hit)
    con.close()


def _make_yaqwsx_like_source(path: Path) -> None:
    """Create a SQLite mirroring yaqwsx's FULL-catalog schema: normalized, NO view.

    category/manufacturer are IDs in sibling tables (no v_components view), which
    is what the real ~10GB cache.sqlite3 looks like.
    """
    con = sqlite3.connect(str(path))
    con.executescript("""
        CREATE TABLE manufacturers (id INTEGER PRIMARY KEY, name TEXT);
        CREATE TABLE categories (id INTEGER PRIMARY KEY, category TEXT, subcategory TEXT);
        CREATE TABLE components (
            lcsc INTEGER PRIMARY KEY,
            category_id INTEGER,
            mfr TEXT,
            package TEXT,
            joints INTEGER,
            manufacturer_id INTEGER,
            basic INTEGER,
            preferred INTEGER,
            description TEXT,
            datasheet TEXT,
            stock INTEGER,
            price TEXT
        );
        """)
    con.execute("INSERT INTO manufacturers VALUES (7, 'Texas Instruments')")
    con.execute("INSERT INTO categories VALUES (3, 'ICs', 'LDO Regulators')")
    con.execute(
        "INSERT INTO components VALUES (12345, 3, 'TLV70033', 'SOT-23-5', 5, 7, 0, 0, "
        "'3.3V LDO', 'http://ds/12345', 0, ?)",
        (json.dumps([{"qFrom": 1, "qTo": None, "price": 0.12}]),),
    )
    con.commit()
    con.close()


def test_convert_yaqwsx_schema_populates_category_and_manufacturer(tmp_path):
    """Full-catalog (yaqwsx) source has no v_components view; the join must be built."""
    source = tmp_path / "cache.sqlite3"
    target = tmp_path / "jlcpcb_parts.db"
    _make_yaqwsx_like_source(source)

    stats = jlcpcb_downloader.convert_source_sqlite(source, target)
    assert stats["total"] == 1

    con = sqlite3.connect(str(target))
    con.row_factory = sqlite3.Row
    row = dict(con.execute("SELECT * FROM components WHERE lcsc='C12345'").fetchone())
    con.close()

    # These would be blank without the built v_components join:
    assert row["category"] == "ICs"
    assert row["subcategory"] == "LDO Regulators"
    assert row["manufacturer"] == "Texas Instruments"
    assert row["mfr_part"] == "TLV70033"
    assert row["library_type"] == "Extended"
    assert json.loads(row["price_json"]) == [{"qty": 1, "price": 0.12}]


def test_normalize_price_json_handles_scalar_and_array_and_empty():
    assert json.loads(jlcpcb_downloader.normalize_price_json(None)) == []
    assert json.loads(jlcpcb_downloader.normalize_price_json("")) == []
    assert json.loads(jlcpcb_downloader.normalize_price_json("[]")) == []
    assert json.loads(jlcpcb_downloader.normalize_price_json(0.05)) == [{"qty": 1, "price": 0.05}]
    arr = json.loads(jlcpcb_downloader.normalize_price_json('[{"qFrom":10,"qTo":50,"price":1.5}]'))
    assert arr == [{"qty": 10, "price": 1.5}]


def test_download_database_falls_through_when_no_source_available(tmp_path, monkeypatch):
    """With CDFER download failing, no 7z present, and no API creds -> clean failure dict."""
    monkeypatch.setattr(
        jlcpcb_downloader.PlatformHelper, "get_data_dir", staticmethod(lambda: tmp_path)
    )

    def _boom(*a, **k):
        raise RuntimeError("network down")

    monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _boom)
    monkeypatch.setattr(jlcpcb_downloader, "_find_7z", lambda: None)
    # Ensure no official creds leak in from the environment.
    for var in ("JLCPCB_APP_ID", "JLCPCB_API_KEY", "JLCPCB_API_SECRET"):
        monkeypatch.delenv(var, raising=False)

    result = jlcpcb_downloader.download_database()

    assert result["success"] is False
    assert "errors" in result and len(result["errors"]) >= 1
    assert any("cdfer" in e for e in result["errors"])


def test_download_database_prefer_cdfer_converts(tmp_path, monkeypatch):
    """prefer_source='cdfer' downloads then converts via the real conversion path."""
    monkeypatch.setattr(
        jlcpcb_downloader.PlatformHelper, "get_data_dir", staticmethod(lambda: tmp_path)
    )

    def _fake_cdfer(cache_dir, progress=None):
        cache_dir.mkdir(parents=True, exist_ok=True)
        src = cache_dir / "cdfer.sqlite3"
        _make_cdfer_like_source(src)
        return src, "Thu, 02 Apr 2026 06:22:46 GMT"

    monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _fake_cdfer)

    result = jlcpcb_downloader.download_database(prefer_source="cdfer")

    assert result["success"] is True
    assert result["source"] == "cdfer"
    assert result["total_parts"] == 2
    assert result["basic_parts"] == 1
    assert result["catalog_last_modified"].startswith("Thu, 02 Apr 2026")
    assert (tmp_path / "jlcpcb_parts.db").exists()


def _http_date(days_ago: int) -> str:
    from datetime import datetime, timedelta, timezone
    from email.utils import format_datetime

    return format_datetime(datetime.now(timezone.utc) - timedelta(days=days_ago))


def test_result_flags_stale_catalog_and_leaves_fresh_alone(tmp_path, monkeypatch):
    """A catalog older than STALE_AFTER_DAYS must be flagged; a fresh one must not."""
    monkeypatch.setattr(
        jlcpcb_downloader.PlatformHelper, "get_data_dir", staticmethod(lambda: tmp_path)
    )

    def _fake_cdfer_dated(http_date):
        def _impl(cache_dir, progress=None):
            cache_dir.mkdir(parents=True, exist_ok=True)
            src = cache_dir / "cdfer.sqlite3"
            _make_cdfer_like_source(src)
            return src, http_date

        return _impl

    # Stale: 60 days old
    monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _fake_cdfer_dated(_http_date(60)))
    stale = jlcpcb_downloader.download_database(prefer_source="cdfer")
    assert stale.get("stale") is True
    assert "warning" in stale
    assert stale["catalog_age_days"] >= jlcpcb_downloader.STALE_AFTER_DAYS

    # Fresh: 1 day old
    monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _fake_cdfer_dated(_http_date(1)))
    fresh = jlcpcb_downloader.download_database(prefer_source="cdfer")
    assert fresh.get("stale") is not True
    assert "warning" not in fresh
    assert fresh["catalog_age_days"] <= 1