path: root/.ai/scripts/tests/test_apkg_to_orgdrill.py

"""Tests for apkg-to-orgdrill.py — the inverse of flashcard-to-anki.py.

The converter reads an Anki .apkg (a zip holding collection.anki2 / .anki21
sqlite) and emits an org-drill .org in the house canonical shape. It is
stdlib-only (zipfile + sqlite3), so it imports directly — no genanki stub.

The apkg schema these tests build by hand mirrors what genanki actually
writes, confirmed against a real apkg generated from flashcard-to-anki.py:
  - col.decks  : JSON {did: {"name": ...}}, always including id-1 "Default"
  - col.models : JSON {mid: {"name": ..., "flds": [{"name": "Front"}, ...]}}
  - notes.flds : fields joined by \x1f; tags space-padded (" tag ")
  - cards      : nid -> did (the Default deck carries no cards)

The round-trip test closes the loop through flashcard-to-anki.py's own
parse(): original org -> forward parse tuples -> apkg fixture -> converter
-> recovered org -> forward parse -> assert the (front, back, tag) tuples
match. Only the apkg materialization is hand-built (the genanki boundary);
everything else is the real code on both sides.
"""
from __future__ import annotations

import importlib.util
import json
import sqlite3
import sys
import types
import zipfile
from pathlib import Path

import pytest

SCRIPTS = Path(__file__).resolve().parents[1]
CONVERTER = SCRIPTS / "apkg-to-orgdrill.py"
FORWARD = SCRIPTS / "flashcard-to-anki.py"


def _load(path: Path, name: str, stub_genanki: bool = False):
    if stub_genanki:
        sys.modules.setdefault("genanki", types.ModuleType("genanki"))
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    # Register before exec: @dataclass resolves cls.__module__ via sys.modules
    # (Python 3.14), which is None for an unregistered importlib module.
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


@pytest.fixture(scope="module")
def conv():
    return _load(CONVERTER, "apkg_to_orgdrill")


@pytest.fixture(scope="module")
def forward():
    return _load(FORWARD, "flashcard_to_anki", stub_genanki=True)


# --- fixture builder: write a genanki-shaped apkg by hand ------------------

def _make_apkg(
    path: Path,
    decks: dict[int, str],
    models: dict[int, list[str]],
    notes: list[tuple[int, int, list[str], str]],  # (nid, mid, fields, tag)
    cards: list[tuple[int, int]],  # (nid, did)
    *,
    media: str = "{}",
) -> None:
    """Materialize a minimal apkg matching genanki's collection.anki2 shape."""
    col_dir = path.parent / f"{path.stem}-build"
    col_dir.mkdir(parents=True, exist_ok=True)
    db = col_dir / "collection.anki2"
    if db.exists():
        db.unlink()
    con = sqlite3.connect(db)
    con.execute("CREATE TABLE col (id INTEGER, decks TEXT, models TEXT)")
    decks_json = {"1": {"name": "Default"}}
    decks_json.update({str(did): {"name": name} for did, name in decks.items()})
    models_json = {
        str(mid): {"name": f"{decks.get(list(decks)[0], 'M')} model",
                   "flds": [{"name": n, "ord": i} for i, n in enumerate(flds)]}
        for mid, flds in models.items()
    }
    con.execute("INSERT INTO col (id, decks, models) VALUES (1, ?, ?)",
                (json.dumps(decks_json), json.dumps(models_json)))
    con.execute("CREATE TABLE notes (id INTEGER, mid INTEGER, flds TEXT, tags TEXT)")
    for nid, mid, fields, tag in notes:
        con.execute("INSERT INTO notes (id, mid, flds, tags) VALUES (?, ?, ?, ?)",
                    (nid, mid, "\x1f".join(fields), f" {tag} " if tag else "  "))
    con.execute("CREATE TABLE cards (id INTEGER, nid INTEGER, did INTEGER)")
    for i, (nid, did) in enumerate(cards):
        con.execute("INSERT INTO cards (id, nid, did) VALUES (?, ?, ?)", (1000 + i, nid, did))
    con.commit()
    con.close()
    with zipfile.ZipFile(path, "w") as z:
        z.write(db, "collection.anki2")
        z.writestr("media", media)


# --- html_to_org_body ------------------------------------------------------

def test_html_to_org_splits_br_into_lines(conv):
    assert conv.html_to_org_body("one<br>two<br>three") == ["one", "two", "three"]


def test_html_to_org_handles_br_variants(conv):
    assert conv.html_to_org_body("a<br/>b<br />c<BR>d") == ["a", "b", "c", "d"]


def test_html_to_org_unescapes_entities_amp_last(conv):
    # Inverts escape_html (which escapes & first): &lt; &gt; &amp; -> < > &.
    assert conv.html_to_org_body("x &lt;tag&gt; &amp; y") == ["x <tag> & y"]


def test_html_to_org_preserves_a_literal_escaped_entity(conv):
    # Forward-escaping the literal "&lt;" yields "&amp;lt;"; the inverse must
    # recover "&lt;", not "<".
    assert conv.html_to_org_body("&amp;lt;") == ["&lt;"]


def test_html_to_org_strips_answer_hr(conv):
    assert conv.html_to_org_body('front<hr id="answer">back') == ["front", "back"]


def test_html_to_org_empty_back_is_empty(conv):
    assert conv.html_to_org_body("") == []


# --- read_apkg -------------------------------------------------------------

def test_read_apkg_single_deck_recovers_front_back_tag_deck(conv, tmp_path):
    apkg = tmp_path / "d.apkg"
    _make_apkg(
        apkg,
        decks={20: "My Deck"},
        models={9: ["Front", "Back"]},
        notes=[(100, 9, ["Q1?", "A1.<br>line2"], "sec-one")],
        cards=[(100, 20)],
    )
    recovered = conv.read_apkg(apkg)
    assert len(recovered) == 1
    note = recovered[0]
    assert note.deck == "My Deck"
    assert note.front == "Q1?"
    assert note.back_html == "A1.<br>line2"
    assert note.tag == "sec-one"


def test_read_apkg_multiple_decks_grouped(conv, tmp_path):
    apkg = tmp_path / "multi.apkg"
    _make_apkg(
        apkg,
        decks={20: "Deck A", 21: "Deck B"},
        models={9: ["Front", "Back"]},
        notes=[(100, 9, ["QA?", "AA"], "ta"), (101, 9, ["QB?", "AB"], "tb")],
        cards=[(100, 20), (101, 21)],
    )
    decks = {n.deck for n in conv.read_apkg(apkg)}
    assert decks == {"Deck A", "Deck B"}


def test_read_apkg_skips_default_deck_without_cards(conv, tmp_path):
    apkg = tmp_path / "def.apkg"
    _make_apkg(
        apkg,
        decks={20: "Real Deck"},
        models={9: ["Front", "Back"]},
        notes=[(100, 9, ["Q?", "A"], "t")],
        cards=[(100, 20)],
    )
    assert {n.deck for n in conv.read_apkg(apkg)} == {"Real Deck"}


def test_read_apkg_warns_and_skips_non_basic_model(conv, tmp_path, capsys):
    apkg = tmp_path / "cloze.apkg"
    _make_apkg(
        apkg,
        decks={20: "Cloze Deck"},
        models={9: ["Text", "Extra"]},  # not Front/Back
        notes=[(100, 9, ["some {{c1::text}}", "extra"], "t")],
        cards=[(100, 20)],
    )
    recovered = conv.read_apkg(apkg)
    assert recovered == []
    assert "skip" in capsys.readouterr().err.lower()


def test_read_apkg_reads_anki21_collection_name(conv, tmp_path):
    # A .anki21 collection filename must be read the same as .anki2.
    apkg = tmp_path / "new.apkg"
    _make_apkg(
        apkg,
        decks={20: "Deck"},
        models={9: ["Front", "Back"]},
        notes=[(100, 9, ["Q?", "A"], "t")],
        cards=[(100, 20)],
    )
    # Rewrite the zip renaming the collection member to .anki21.
    with zipfile.ZipFile(apkg) as z:
        data = z.read("collection.anki2")
        media = z.read("media")
    with zipfile.ZipFile(apkg, "w") as z:
        z.writestr("collection.anki21", data)
        z.writestr("media", media)
    assert conv.read_apkg(apkg)[0].front == "Q?"


def test_read_apkg_flags_media_reference(conv, tmp_path, capsys):
    apkg = tmp_path / "media.apkg"
    _make_apkg(
        apkg,
        decks={20: "Deck"},
        models={9: ["Front", "Back"]},
        notes=[(100, 9, ["Q?", 'see <img src="x.png">'], "t")],
        cards=[(100, 20)],
    )
    conv.read_apkg(apkg)
    assert "media" in capsys.readouterr().err.lower()


# --- notes_to_org ----------------------------------------------------------

def test_notes_to_org_emits_canonical_shape(conv):
    Note = conv.Note
    notes = [
        Note(deck="My Deck", front="Q1?", back_html="A1.", tag="alpha"),
        Note(deck="My Deck", front="Q2?", back_html="A2.", tag="alpha"),
    ]
    ids = iter(["id-1", "id-2"])
    org = conv.notes_to_org(notes, "My Deck", new_id=lambda: next(ids))
    assert "#+TITLE: My Deck" in org
    assert "* alpha" in org
    assert "** Q1? :drill:" in org
    assert ":ID: id-1" in org
    assert ":ID: id-2" in org
    assert org.count("* alpha") == 1  # both cards share one section


def test_notes_to_org_distinct_tags_get_distinct_sections(conv):
    Note = conv.Note
    notes = [
        Note(deck="D", front="Qa?", back_html="a", tag="alpha"),
        Note(deck="D", front="Qb?", back_html="b", tag="beta"),
    ]
    org = conv.notes_to_org(notes, "D", new_id=lambda: "x")
    assert "* alpha" in org and "* beta" in org


# --- round-trip through the real forward parse() ---------------------------

def test_round_trip_matches_forward_parse_tuples(conv, forward, tmp_path):
    original = (
        "#+TITLE: RT Deck\n"
        "\n"
        "* First Section\n"
        "** What is 2+2? :drill:\n"
        ":PROPERTIES:\n:ID: aaaa\n:END:\n"
        "Four.\n"
        "Second line with <angle> & amp.\n"
        "\n"
        "* Second Section\n"
        "** Capital of France? :drill:\n"
        "Paris.\n"
    )
    tuples = forward.parse(original)  # [(front, back_html, anki_tags), ...]
    assert len(tuples) == 2

    apkg = tmp_path / "rt.apkg"
    _make_apkg(
        apkg,
        decks={20: "RT Deck"},
        models={9: ["Front", "Back"]},
        # anki_tags is a list; the apkg tags field is space-joined.
        notes=[(100 + i, 9, [f, b], " ".join(tags))
               for i, (f, b, tags) in enumerate(tuples)],
        cards=[(100 + i, 20) for i in range(len(tuples))],
    )

    by_deck = conv.convert(apkg)
    assert set(by_deck) == {"RT Deck"}
    recovered_tuples = forward.parse(by_deck["RT Deck"])
    assert recovered_tuples == tuples


# --- errors ----------------------------------------------------------------

def test_read_apkg_missing_collection_errors(conv, tmp_path):
    bad = tmp_path / "bad.apkg"
    with zipfile.ZipFile(bad, "w") as z:
        z.writestr("media", "{}")
    with pytest.raises(Exception):
        conv.read_apkg(bad)


def test_read_apkg_not_a_zip_errors(conv, tmp_path):
    notzip = tmp_path / "plain.apkg"
    notzip.write_text("not a zip")
    with pytest.raises(Exception):
        conv.read_apkg(notzip)