Jyotisha/tests/test_qizheng_chart_engine.py

"""Adapter tests for the vendored seven-governors engine."""

from __future__ import annotations

import json
import shutil
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest

from scripts import qizheng_chart_engine as engine
from scripts.qizheng_chart_engine import calculate_qizheng_chart

ROOT = Path(__file__).resolve().parents[1]
GOLDEN = ROOT / "tests" / "golden" / "qizheng_stem_branch_19900409.json"
SAMPLE = {
    "year": 1990,
    "month": 4,
    "day": 9,
    "hour": 13,
    "minute": 24,
    "lat": 31.19,
    "lon": 121.44,
    "tz": 8,
}


def _require_node() -> None:
    if shutil.which("node") is None:
        pytest.skip("node runtime is not available")


def test_golden_fixture_is_real_engine_output() -> None:
    chart = json.loads(GOLDEN.read_text(encoding="utf-8"))
    assert set(chart["bodies"]) == set(engine.REQUIRED_BODIES)
    assert len(chart["palaces"]) == 12
    assert chart["ascendant"]["palace"] == "午宮"
    assert chart["ketuMode"] == "apogee"
    sun = chart["bodies"]["sun"]
    assert sun["mansion"] == "奎"
    assert round(sun["siderealLon"], 2) == 177.94
    assert round(sun["mansionDegree"], 2) == 7.44
    assert chart["dignities"]["sun"] == "陷"
    assert chart["starSpirits"] == []
    assert len(chart["aspects"]) == 27


def test_live_chart_matches_golden_and_product_contract() -> None:
    _require_node()
    golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
    result = calculate_qizheng_chart(SAMPLE)
    assert result["coordinate_system"] == "qizheng_mansion_degrees_from_jiao"
    assert set(result["bodies"]) == set(engine.REQUIRED_BODIES)
    assert len(result["palaces"]) == 12
    assert result["ascendant"]["palace"] == golden["ascendant"]["palace"]
    assert result["calculation"]["ketu_mode"] == "apogee"
    assert result["calculation"]["sidereal_mode"] == {"type": "modern"}
    assert result["dignities"]["status"] == "unclosed"
    assert result["dignities"]["may_enter_conclusions"] is False
    assert result["dignities"]["runtime_promotable_count"] == 0
    assert "神煞" not in result["boundary"]
    assert "未闭合" in result["boundary"] or "未闭合" in result["dignities"]["status"] or result["dignities"]["status"] == "unclosed"
    assert round(result["bodies"]["sun"]["siderealLon"], 2) == round(golden["bodies"]["sun"]["siderealLon"], 2)
    raw = json.dumps(result["raw_engine_output"], ensure_ascii=False)
    assert "vendor" not in raw
    assert "stem-branch" not in raw or result["raw_engine_output"].get("ketuMode") == "apogee"
    assert "\\" not in raw or "[redacted-path]" in raw


def test_ketu_mode_request_is_echoed_and_engine_mode_is_visible() -> None:
    _require_node()
    result = calculate_qizheng_chart({**SAMPLE, "ketu_mode": "descending-node"})
    assert result["calculation"]["ketu_mode"] == "descending-node"
    # a911c890 CLI has no ketuMode flag; actual school is the engine default.
    assert result["calculation"]["engine_ketu_mode"] == "apogee"


def test_missing_lat_is_structured_input_error() -> None:
    with pytest.raises(engine.QizhengChartError, match="lat is required") as caught:
        calculate_qizheng_chart({k: v for k, v in SAMPLE.items() if k != "lat"})
    assert caught.value.error_code == "ERR_QIZHENG_INPUT"


def test_node_missing_error(monkeypatch) -> None:
    monkeypatch.setattr(engine.shutil, "which", lambda name: None)
    with pytest.raises(engine.QizhengChartError, match="unavailable") as caught:
        calculate_qizheng_chart(SAMPLE)
    assert caught.value.error_code == "ERR_QIZHENG_NODE_MISSING"


def test_cli_missing_error(monkeypatch, tmp_path: Path) -> None:
    monkeypatch.setattr(engine, "VENDORED_CLI", tmp_path / "cli.cjs")
    monkeypatch.setattr(engine.shutil, "which", lambda name: "node")
    with pytest.raises(engine.QizhengChartError, match="unavailable") as caught:
        calculate_qizheng_chart(SAMPLE)
    assert caught.value.error_code == "ERR_QIZHENG_CLI_MISSING"


def test_timeout_error(monkeypatch) -> None:
    monkeypatch.setattr(engine.shutil, "which", lambda name: "node")

    def _boom(*args, **kwargs):  # noqa: ANN001
        raise subprocess.TimeoutExpired(cmd="node", timeout=20)

    monkeypatch.setattr(engine.subprocess, "run", _boom)
    with pytest.raises(engine.QizhengChartError, match="超时") as caught:
        calculate_qizheng_chart(SAMPLE)
    assert caught.value.error_code == "ERR_QIZHENG_TIMEOUT"


def test_nonzero_exit_does_not_leak_stderr(monkeypatch) -> None:
    monkeypatch.setattr(engine.shutil, "which", lambda name: "node")
    monkeypatch.setattr(
        engine.subprocess,
        "run",
        lambda *args, **kwargs: SimpleNamespace(
            returncode=1, stdout="", stderr="C:\\\\secret\\\\stem-branch\\\\cli.cjs exploded"
        ),
    )
    with pytest.raises(engine.QizhengChartError, match="exited with an error") as caught:
        calculate_qizheng_chart(SAMPLE)
    assert caught.value.error_code == "ERR_QIZHENG_ENGINE_EXIT"
    assert "secret" not in str(caught.value)
    assert "cli.cjs" not in str(caught.value)


def test_non_json_stdout_error(monkeypatch) -> None:
    monkeypatch.setattr(engine.shutil, "which", lambda name: "node")
    monkeypatch.setattr(
        engine.subprocess,
        "run",
        lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="not-json", stderr=""),
    )
    with pytest.raises(engine.QizhengChartError, match="JSON") as caught:
        calculate_qizheng_chart(SAMPLE)
    assert caught.value.error_code == "ERR_QIZHENG_BAD_JSON"


def test_source_never_invokes_forbidden_cli_flags() -> None:
    source = Path(engine.__file__).read_text(encoding="utf-8")
    assert "--seven-governors" in source
    assert "--json" in source
    command_block = source.split("command = [", 1)[1].split("]", 1)[0]
    for flag in engine.FORBIDDEN_CLI_FLAGS:
        assert flag not in command_block