backend/tests/test_cli.py

"""Tests for the Phase 3a CLI ingestion scaffold (backend/cli.py).

Covers parser routing and that each subcommand forwards the right args to the
existing ingestion functions — WITHOUT running ML (the underlying modules are
faked in sys.modules). A subprocess test enforces the key invariant: importing
cli + building the parser pulls in no torch (the lazy-import boundary the
eventual web/ingestion split depends on).
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
from types import SimpleNamespace

BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BACKEND_DIR)

import cli


def test_parser_routes_subcommands_to_handlers():
    parser = cli.build_parser()

    ns = parser.parse_args(["ingest", "video", "--natural-key", "K", "--label", "480p", "--keep-source"])
    assert ns.func is cli._cmd_ingest_video
    assert ns.natural_key == "K" and ns.label == "480p" and ns.keep_source is True and ns.all is False

    ns = parser.parse_args(["ingest", "vod", "--language", "S", "--redownload"])
    assert ns.func is cli._cmd_ingest_vod and ns.language == "S" and ns.redownload is True

    ns = parser.parse_args(["ingest", "images", "--source", "web"])
    assert ns.func is cli._cmd_ingest_images and ns.source == "web"

    ns = parser.parse_args(["process", "subtitles", "--language", "F", "--keep-html"])
    assert ns.func is cli._cmd_process_subtitles and ns.language == "F" and ns.keep_html is True

    ns = parser.parse_args(["reindex", "embeddings", "--batch-size", "10"])
    assert ns.func is cli._cmd_reindex_embeddings and ns.batch_size == 10


def test_ingest_vod_forwards_args(monkeypatch):
    calls = {}
    fake = SimpleNamespace(
        download_vod=lambda lang, redownload=False: (calls.update(dl=(lang, redownload)) or {"ok": 1}),
        combine_media_info=lambda lang: (calls.update(combine=lang) or {"a": {}, "b": {}}),
    )
    monkeypatch.setitem(sys.modules, "jw_api", fake)

    rc = cli._cmd_ingest_vod(SimpleNamespace(language="E", redownload=True))

    assert rc == 0
    assert calls["dl"] == ("E", True)
    assert calls["combine"] == "E"


def test_ingest_vod_reports_error_when_download_fails(monkeypatch):
    fake = SimpleNamespace(download_vod=lambda lang, redownload=False: None, combine_media_info=lambda lang: {})
    monkeypatch.setitem(sys.modules, "jw_api", fake)
    assert cli._cmd_ingest_vod(SimpleNamespace(language="E", redownload=False)) == 1


def test_ingest_video_single_forwards_args(monkeypatch):
    calls = {}
    fake = SimpleNamespace(
        process_video=lambda nk, label, delete_video_after=None: (
            calls.update(pv=(nk, label, delete_video_after)) or {"status": "success"}
        ),
        process_all_local_videos=lambda **k: {"status": "success"},
    )
    monkeypatch.setitem(sys.modules, "process_video", fake)

    rc = cli._cmd_ingest_video(
        SimpleNamespace(natural_key="vid-1", label="720p", all=False, reprocess=False, keep_source=False)
    )

    assert rc == 0
    assert calls["pv"] == ("vid-1", "720p", None)  # keep_source False -> delete_video_after None (sentinel)


def test_ingest_video_keep_source_forces_no_delete(monkeypatch):
    calls = {}
    fake = SimpleNamespace(
        process_video=lambda nk, label, delete_video_after=None: (
            calls.update(d=delete_video_after) or {"status": "success"}
        ),
        process_all_local_videos=lambda **k: {"status": "success"},
    )
    monkeypatch.setitem(sys.modules, "process_video", fake)

    cli._cmd_ingest_video(
        SimpleNamespace(natural_key="v", label="720p", all=False, reprocess=False, keep_source=True)
    )

    assert calls["d"] is False


def test_ingest_video_all_forwards_args(monkeypatch):
    calls = {}
    fake = SimpleNamespace(
        process_video=lambda *a, **k: {"status": "success"},
        process_all_local_videos=lambda label, reprocess, delete_video_after: (
            calls.update(batch=(label, reprocess, delete_video_after)) or {"status": "success"}
        ),
    )
    monkeypatch.setitem(sys.modules, "process_video", fake)

    cli._cmd_ingest_video(
        SimpleNamespace(natural_key=None, label="720p", all=True, reprocess=True, keep_source=False)
    )

    assert calls["batch"] == ("720p", True, None)


def test_ingest_video_requires_key_or_all_without_importing_ml(monkeypatch):
    # No process_video fake injected: the validation must fire BEFORE the import,
    # so this must not raise ImportError and must return an error code.
    monkeypatch.delitem(sys.modules, "process_video", raising=False)
    rc = cli._cmd_ingest_video(
        SimpleNamespace(natural_key=None, label="720p", all=False, reprocess=False, keep_source=False)
    )
    assert rc == 1


def test_emit_exit_codes():
    assert cli._emit({"status": "error", "message": "x"}) == 1
    assert cli._emit({"status": "success"}) == 0
    assert cli._emit({"indexed": 3}) == 0


def test_ingest_subtitles_forwards_args(monkeypatch):
    calls = {}
    monkeypatch.setitem(sys.modules, "jw_api", SimpleNamespace(combine_media_info=lambda lang: {"a": {}, "b": {}}))
    monkeypatch.setitem(
        sys.modules,
        "subtitles_download",
        SimpleNamespace(
            process_all_media_for_subtitles=lambda media, lang: (
                calls.update(c=(len(media), lang)) or {"downloaded": 2}
            )
        ),
    )
    assert cli._cmd_ingest_subtitles(SimpleNamespace(language="E")) == 0
    assert calls["c"] == (2, "E")


def test_ingest_subtitles_errors_without_catalog(monkeypatch):
    monkeypatch.setitem(sys.modules, "jw_api", SimpleNamespace(combine_media_info=lambda lang: {}))
    monkeypatch.setitem(
        sys.modules, "subtitles_download",
        SimpleNamespace(process_all_media_for_subtitles=lambda *a: {}),
    )
    assert cli._cmd_ingest_subtitles(SimpleNamespace(language="E")) == 1


def test_ingest_images_publications_branch(monkeypatch):
    calls = {}
    monkeypatch.setitem(
        sys.modules, "search_publication_images",
        SimpleNamespace(index_all_publication_images=lambda: (calls.update(pub=True) or {"indexed": 5})),
    )
    assert cli._cmd_ingest_images(SimpleNamespace(source="publications", language="E")) == 0
    assert calls.get("pub") is True


def test_ingest_images_web_branch_forwards_language(monkeypatch):
    calls = {}
    monkeypatch.setitem(
        sys.modules, "web_images_processor",
        SimpleNamespace(crawl_web_images=lambda language="E": (calls.update(lang=language) or {"indexed": 3})),
    )
    assert cli._cmd_ingest_images(SimpleNamespace(source="web", language="S")) == 0
    assert calls["lang"] == "S"


def test_process_subtitles_forwards_remove_html(monkeypatch):
    calls = {}
    monkeypatch.setitem(sys.modules, "search", SimpleNamespace(get_search_instance=lambda: object()))
    monkeypatch.setitem(
        sys.modules, "processing_route_helpers",
        SimpleNamespace(
            process_subtitles_sync=lambda runtime, *, language, remove_html: (
                calls.update(r=(language, remove_html)) or {"processed": 1}
            )
        ),
    )
    assert cli._cmd_process_subtitles(SimpleNamespace(language="E", keep_html=False)) == 0
    assert calls["r"] == ("E", True)  # keep_html False -> remove_html True


def test_process_subtitles_translates_error_to_exit_code(monkeypatch):
    monkeypatch.setitem(sys.modules, "search", SimpleNamespace(get_search_instance=lambda: object()))

    def boom(runtime, *, language, remove_html):
        raise RuntimeError("no subtitles dir")

    monkeypatch.setitem(sys.modules, "processing_route_helpers", SimpleNamespace(process_subtitles_sync=boom))
    assert cli._cmd_process_subtitles(SimpleNamespace(language="E", keep_html=False)) == 1


def test_reindex_embeddings_forwards_args(monkeypatch):
    calls = {}
    inst = SimpleNamespace(
        rebuild_subtitle_embeddings=lambda language, batch_size: (
            calls.update(r=(language, batch_size)) or {"status": "success"}
        )
    )
    monkeypatch.setitem(sys.modules, "search", SimpleNamespace(get_search_instance=lambda: inst))
    assert cli._cmd_reindex_embeddings(SimpleNamespace(language="E", batch_size=10)) == 0
    assert calls["r"] == ("E", 10)


def test_reindex_subtitles_forwards_args(monkeypatch):
    calls = {}
    inst = SimpleNamespace(rebuild_index=lambda language: (calls.update(r=language) or {"status": "success"}))
    monkeypatch.setitem(sys.modules, "search", SimpleNamespace(get_search_instance=lambda: inst))
    assert cli._cmd_reindex_subtitles(SimpleNamespace(language=None)) == 0
    assert calls["r"] is None


def test_result_json_writes_result_dict_to_file(tmp_path, monkeypatch):
    """--result-json writes the emitted result dict to a file (clean IPC for callers).

    Uses the ML-free `ingest video` validation error so no torch is imported.
    """
    monkeypatch.setattr(cli, "ensure_runtime_dirs", lambda *_a, **_k: None)
    monkeypatch.setattr(cli, "get_app_config", lambda: None)
    out = tmp_path / "result.json"

    # No --natural-key and no --all -> handler emits an error dict (ML-free path).
    rc = cli.main(["--result-json", str(out), "ingest", "video"])

    assert rc == 1  # error status -> exit code 1
    written = json.loads(out.read_text())
    assert written["status"] == "error"
    assert "natural-key" in written["message"]


def test_result_json_absent_does_not_write_file(tmp_path, monkeypatch):
    """Without --result-json, _emit only prints; it must not create a file."""
    monkeypatch.setattr(cli, "ensure_runtime_dirs", lambda *_a, **_k: None)
    monkeypatch.setattr(cli, "get_app_config", lambda: None)
    # Reset any module-level result-file state a prior test may have set.
    cli._set_result_file(None)

    rc = cli.main(["ingest", "video"])

    assert rc == 1
    assert not (tmp_path / "should_not_exist.json").exists()


def test_cli_import_and_parser_are_ml_free():
    """Importing cli + building the parser must not pull in torch/process_video."""
    code = (
        f"import sys; sys.path.insert(0, {BACKEND_DIR!r});"
        "import cli; cli.build_parser();"
        "assert 'torch' not in sys.modules, 'cli import pulled torch';"
        "assert 'process_video' not in sys.modules, 'cli import pulled process_video';"
        "print('ML-FREE-OK')"
    )
    result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True)
    assert result.returncode == 0, result.stderr
    assert "ML-FREE-OK" in result.stdout