tests/test_cli.py

from __future__ import annotations

import io
import json
import sqlite3

import deberta_ime.cli as cli_module
from deberta_ime import RerankRequest
from deberta_ime.cli import run
from deberta_ime.mozc import build_mozc_index


class StaticScorer:
    def score_candidates(self, request: RerankRequest) -> list[float]:
        return [-3.0, -0.5]


def test_jsonl_cli_returns_ranked_candidates_and_decision_metadata() -> None:
    stdin = io.StringIO(
        json.dumps(
            {
                "reading": "はし",
                "left_context": ["川", "に", "架かる"],
                "right_context": ["を", "渡る"],
                "candidates": [
                    {"surface": "箸", "prior_score": 0.0},
                    {"surface": "橋", "prior_score": 0.0},
                ],
            },
            ensure_ascii=False,
        )
    )
    stdout = io.StringIO()

    exit_code = run(
        ["rerank", "--prior-weight", "0", "--min-margin", "0.5"],
        stdin=stdin,
        stdout=stdout,
        scorer=StaticScorer(),
    )

    output = json.loads(stdout.getvalue())
    assert exit_code == 0
    assert output["ok"] is True
    assert output["decision"] == {
        "changed": True,
        "reason": "accepted",
        "margin": 2.5,
    }
    assert output["ranked"][0] == {
        "surface": "橋",
        "original_rank": 1,
        "prior_score": 0.0,
        "model_score": -0.5,
        "combined_score": -0.5,
    }


def test_cli_reads_and_writes_utf8_files_without_console_encoding(tmp_path) -> None:
    input_path = tmp_path / "request.jsonl"
    output_path = tmp_path / "response.jsonl"
    input_path.write_text(
        json.dumps(
            {
                "reading": "はし",
                "candidates": ["箸", "橋"],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )

    exit_code = run(
        [
            "rerank",
            "--input",
            str(input_path),
            "--output",
            str(output_path),
            "--prior-weight",
            "0",
        ],
        scorer=StaticScorer(),
    )

    output = json.loads(output_path.read_text(encoding="utf-8"))
    assert exit_code == 0
    assert output["reading"] == "はし"
    assert output["ranked"][0]["surface"] == "橋"


def test_cli_rejects_same_input_and_output_without_truncating(tmp_path) -> None:
    path = tmp_path / "request.jsonl"
    original = json.dumps({"reading": "はし", "candidates": ["箸", "橋"]})
    path.write_text(original, encoding="utf-8")
    stderr = io.StringIO()

    exit_code = run(
        ["rerank", "--input", str(path), "--output", str(path)],
        stderr=stderr,
        scorer=StaticScorer(),
    )

    assert exit_code == 2
    assert "must be different" in stderr.getvalue()
    assert path.read_text(encoding="utf-8") == original


def test_mozc_index_cli_builds_index_and_prints_source_manifest(tmp_path) -> None:
    dictionary_dir = tmp_path / "dictionary_oss"
    dictionary_dir.mkdir()
    (dictionary_dir / "dictionary00.txt").write_text(
        "はし\t1\t1\t3500\t橋\n",
        encoding="utf-8",
    )
    index_path = tmp_path / "mozc.sqlite3"
    stdout = io.StringIO()

    exit_code = run(
        [
            "mozc-index",
            "--dictionary-dir",
            str(dictionary_dir),
            "--output",
            str(index_path),
            "--source-revision",
            "fixture-revision",
        ],
        stdout=stdout,
    )

    payload = json.loads(stdout.getvalue())
    assert exit_code == 0
    assert index_path.is_file()
    assert payload["ok"] is True
    assert payload["manifest"]["source_revision"] == "fixture-revision"
    assert payload["manifest"]["indexed_entries"] == 1


def test_mozc_index_cli_reports_invalid_source_without_traceback(tmp_path) -> None:
    stdout = io.StringIO()
    stderr = io.StringIO()

    exit_code = run(
        [
            "mozc-index",
            "--dictionary-dir",
            str(tmp_path / "missing"),
            "--output",
            str(tmp_path / "mozc.sqlite3"),
        ],
        stdout=stdout,
        stderr=stderr,
    )

    assert exit_code == 2
    assert stdout.getvalue() == ""
    assert stderr.getvalue() == "error: no dictionary??.txt files found\n"


def test_mozc_index_cli_reports_sqlite_failure_without_traceback(tmp_path, monkeypatch) -> None:
    def fail_build(*args: object, **kwargs: object) -> None:
        del args, kwargs
        raise sqlite3.OperationalError("fixture database failure")

    monkeypatch.setattr(cli_module, "build_mozc_index", fail_build)
    stdout = io.StringIO()
    stderr = io.StringIO()

    exit_code = run(
        [
            "mozc-index",
            "--dictionary-dir",
            str(tmp_path / "dictionary"),
            "--output",
            str(tmp_path / "mozc.sqlite3"),
        ],
        stdout=stdout,
        stderr=stderr,
    )

    assert exit_code == 2
    assert stdout.getvalue() == ""
    assert stderr.getvalue() == "error: fixture database failure\n"


def test_mozc_rerank_cli_generates_finite_candidates_before_reranking(tmp_path) -> None:
    dictionary_dir = tmp_path / "dictionary_oss"
    dictionary_dir.mkdir()
    (dictionary_dir / "dictionary00.txt").write_text(
        "はし\t1\t1\t3500\t箸\nはし\t1\t1\t3800\t橋\n",
        encoding="utf-8",
    )
    index_path = tmp_path / "mozc.sqlite3"
    build_mozc_index(dictionary_dir, index_path, source_revision="fixture")
    stdin = io.StringIO(
        json.dumps(
            {
                "reading": "ハシ",
                "left_context": ["川", "に", "架かる"],
                "right_context": ["を", "渡る"],
            },
            ensure_ascii=False,
        )
    )
    stdout = io.StringIO()

    class StrongContextScorer:
        def score_candidates(self, request: RerankRequest) -> list[float]:
            return [-10.0, 0.0]

    exit_code = run(
        [
            "mozc-rerank",
            "--index",
            str(index_path),
            "--profile",
            "bidirectional",
        ],
        stdin=stdin,
        stdout=stdout,
        scorer=StrongContextScorer(),
    )

    payload = json.loads(stdout.getvalue())
    assert exit_code == 0
    assert payload["candidate_source"] == {
        "kind": "mozc_oss_dictionary_index",
        "source_revision": "fixture",
        "limit": 8,
    }
    assert payload["profile"] == "bidirectional"
    assert payload["profile_config"] == {
        "prior_weight": 3.0,
        "min_margin": 1.5,
        "selected_on_revision": "7bc20119f476b552635e0640644e577b6fd3606b",
    }
    assert [item["surface"] for item in payload["ranked"]] == ["橋", "箸"]
    assert payload["decision"]["changed"] is True


def test_incremental_profile_rejects_future_context_instead_of_leaking_it(tmp_path) -> None:
    dictionary_dir = tmp_path / "dictionary_oss"
    dictionary_dir.mkdir()
    (dictionary_dir / "dictionary00.txt").write_text(
        "はし\t1\t1\t3500\t箸\nはし\t1\t1\t3800\t橋\n",
        encoding="utf-8",
    )
    index_path = tmp_path / "mozc.sqlite3"
    build_mozc_index(dictionary_dir, index_path, source_revision="fixture")
    stdin = io.StringIO(
        json.dumps(
            {"reading": "はし", "left_context": ["川"], "right_context": ["を"]},
            ensure_ascii=False,
        )
    )
    stdout = io.StringIO()

    exit_code = run(
        ["mozc-rerank", "--index", str(index_path), "--profile", "incremental"],
        stdin=stdin,
        stdout=stdout,
        scorer=StaticScorer(),
    )

    payload = json.loads(stdout.getvalue())
    assert exit_code == 2
    assert payload["ok"] is False
    assert payload["error"] == "incremental profile does not accept right_context"


def test_mozc_rerank_cli_rejects_corrupt_index_before_reading_requests(tmp_path) -> None:
    index_path = tmp_path / "corrupt.sqlite3"
    index_path.write_bytes(b"not a sqlite database")
    stderr = io.StringIO()

    exit_code = run(
        ["mozc-rerank", "--index", str(index_path)],
        stdin=io.StringIO(""),
        stdout=io.StringIO(),
        stderr=stderr,
        scorer=StaticScorer(),
    )

    assert exit_code == 2
    assert stderr.getvalue().startswith("error: cannot open Mozc index: ")