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: ")