"""Step manifest model routing 통합 (problems.md #5).

이전: ``app.modules.llm.llm_client.PIPELINE_STEPS`` 가 별도 dict 로 정의되어
``app.core.step_manifest.STEP_MANIFEST`` 의 ``default_model`` 필드와 drift 가능
(``planning_doc_analysis`` 의 alias 불일치 등). UI / runtime / admin 어느 쪽이
truth 인지 모호.

본 PR: STEP_MANIFEST 가 single source. ``_build_pipeline_steps`` 가 manifest +
extension table 을 합쳐 ``PIPELINE_STEPS`` view 를 build. ``_resolve_model`` 은
이 view 만 lookup. ``llm_router.py`` (dead code) 삭제.

검증:
  - manifest 의 모든 step 이 PIPELINE_STEPS 에 동일 model 로 등장
  - extension table (sub_steps + v2 legacy) 도 PIPELINE_STEPS 에 등장
  - manifest 와 extension 충돌 시 manifest 우선
  - _resolve_model 의 분기 + project_config override
  - llm_router.py 삭제 후 의존 없음
"""
from __future__ import annotations

import pytest


# ---------------------------------------------------------------------------
# 1. manifest single-source 일관성
# ---------------------------------------------------------------------------


def test_pipeline_steps_includes_every_manifest_entry():
    """STEP_MANIFEST 의 모든 step 이 PIPELINE_STEPS 에 존재."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import PIPELINE_STEPS

    missing = [sid for sid in STEP_MANIFEST if sid not in PIPELINE_STEPS]
    assert not missing, f"manifest step 누락: {missing}"


def test_pipeline_steps_default_matches_manifest_default_model():
    """PIPELINE_STEPS 의 default 가 manifest.default_model 와 동일 (mismatch 0)."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import PIPELINE_STEPS

    mismatches = []
    for sid, info in STEP_MANIFEST.items():
        expected = info.get("default_model")
        actual = PIPELINE_STEPS[sid].get("default")
        if expected != actual:
            mismatches.append((sid, expected, actual))
    assert not mismatches, f"manifest <-> PIPELINE_STEPS drift: {mismatches}"


def test_pipeline_steps_label_matches_manifest_label():
    """label 도 manifest 우선."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import PIPELINE_STEPS

    mismatches = [
        (sid, STEP_MANIFEST[sid]["label"], PIPELINE_STEPS[sid]["label"])
        for sid in STEP_MANIFEST
        if STEP_MANIFEST[sid].get("label") != PIPELINE_STEPS[sid].get("label")
    ]
    assert not mismatches, f"label drift: {mismatches}"


def test_pipeline_steps_category_matches_manifest_category():
    """category 도 manifest 우선."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import PIPELINE_STEPS

    mismatches = [
        (sid, STEP_MANIFEST[sid]["category"], PIPELINE_STEPS[sid]["category"])
        for sid in STEP_MANIFEST
        if STEP_MANIFEST[sid].get("category") != PIPELINE_STEPS[sid].get("category")
    ]
    assert not mismatches, f"category drift: {mismatches}"


def test_planning_doc_analysis_alias_is_unified():
    """problems.md #5 핵심 mismatch — planning_doc_analysis alias 통일."""
    from app.core.step_manifest import STEP_MANIFEST

    assert STEP_MANIFEST["planning_doc_analysis"]["default_model"] == "gemini-lite"


# ---------------------------------------------------------------------------
# 2. extension table — manifest 미등록 step
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "step,expected_default,expected_category",
    [
        ("prompt_translation", "gpt-mini", "image_sub"),
        ("scene_t2i_gen", "gemini-image", "image_sub"),
        ("scene_t2i_validation", "gpt", "image_sub"),
        ("prompt_sanitize", "gpt", "image_sub"),
        ("angle_recommend", "gpt", "image_sub"),
        ("fal_angle_apply", "fal-ai", "image_sub"),
        ("final_select", "gpt", "image_sub"),
        ("t2i_translation", "gpt-mini", "image_sub"),
        ("entity_extract", "gpt", "analysis"),
        ("entity_style", "gpt", "analysis"),
        ("entity_detail_batch", "gpt", "analysis"),
        ("webbook_gen", "gpt", "auxiliary"),
        ("style_rules", "gpt", "analysis"),
        ("outlook_merge", "gpt", "analysis"),
        ("location_consistency", "gemini-pro", "analysis"),
    ],
)
def test_extension_steps_present(step, expected_default, expected_category):
    """extension table 의 sub_step / v2 legacy 가 PIPELINE_STEPS 에 등록."""
    from app.modules.llm.llm_client import PIPELINE_STEPS

    info = PIPELINE_STEPS.get(step)
    assert info is not None, f"extension step '{step}' 누락"
    assert info["default"] == expected_default
    assert info["category"] == expected_category


def test_extension_does_not_override_manifest():
    """동일 step 이 manifest + extension 양쪽에 있으면 manifest 우선."""
    # _PIPELINE_STEP_EXTENSIONS 의 step 들과 manifest 의 step 들이 disjoint 확인.
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import _PIPELINE_STEP_EXTENSIONS

    overlap = set(STEP_MANIFEST) & set(_PIPELINE_STEP_EXTENSIONS)
    assert not overlap, (
        f"manifest 와 extension 이 겹침 — manifest 우선이지만 extension 정의가 dead: {overlap}"
    )


# ---------------------------------------------------------------------------
# 3. _resolve_model 분기
# ---------------------------------------------------------------------------


def test_resolve_model_uses_manifest_default_for_known_step():
    """manifest source step 들의 default 가 _resolve_model 결과에 반영."""
    from app.modules.llm.llm_client import _resolve_model

    # 2026-07-11 Gemini 원복(사용자 goal): 구 Gemini 담당 스텝은 이관 前 값.
    # text_cleanup/planning_doc_analysis 는 PDF base64 image_url multimodal
    # (Gemini 전용 경로 — OpenAI 는 Invalid MIME type 거부, 2회차 E2E 실측)라
    # gemini-lite 유지.
    assert _resolve_model("text_cleanup") == "gemini-lite"
    assert _resolve_model("scene_segmentation") == "gemini-flash"
    assert _resolve_model("planning_doc_analysis") == "gemini-lite"


def test_resolve_model_uses_extension_for_sub_step():
    from app.modules.llm.llm_client import _resolve_model

    assert _resolve_model("prompt_translation") == "gpt-mini"
    assert _resolve_model("scene_t2i_gen") == "gemini-image"
    assert _resolve_model("fal_angle_apply") == "fal-ai"


def test_resolve_model_falls_back_to_gemini_pro_for_unknown_step():
    """미등록 step → 'gemini-pro' default (기존 동작 보존)."""
    from app.modules.llm.llm_client import _resolve_model

    assert _resolve_model("nonexistent_step_xyz") == "gemini-pro"


def test_resolve_model_warns_once_for_unknown_step(caplog, monkeypatch):
    """미등록 step 시 process 당 1회 logger.warning emit (review I2)."""
    import logging
    from app.modules.llm import llm_client

    # process-global cache 초기화 — 다른 테스트 영향 격리.
    monkeypatch.setattr(llm_client, "_UNKNOWN_STEP_WARNED", set())

    with caplog.at_level(logging.WARNING, logger="app.modules.llm.llm_client"):
        llm_client._resolve_model("ghost_step_aaa")
        llm_client._resolve_model("ghost_step_aaa")  # 재호출
        llm_client._resolve_model("ghost_step_aaa")

    matching = [
        r for r in caplog.records
        if "unknown step 'ghost_step_aaa'" in r.message
    ]
    assert len(matching) == 1, f"기대 1회, 실제 {len(matching)}회"


def test_build_pipeline_steps_warns_on_overlap(caplog, monkeypatch):
    """extension <-> manifest overlap 시 logger.warning + manifest 우선 (review M1)."""
    import logging
    from app.modules.llm import llm_client
    from app.core.step_manifest import STEP_MANIFEST

    # text_cleanup 는 manifest 에 있음. extension 에 임시 추가하면 overlap 발생.
    fake_ext = {"text_cleanup": {"label": "Z", "default": "fake", "category": "z"}}
    monkeypatch.setattr(llm_client, "_PIPELINE_STEP_EXTENSIONS", fake_ext)

    with caplog.at_level(logging.WARNING, logger="app.modules.llm.llm_client"):
        view = llm_client._build_pipeline_steps()

    # manifest 가 우선 — text_cleanup 의 default 가 manifest 값 유지.
    assert view["text_cleanup"]["default"] == STEP_MANIFEST["text_cleanup"]["default_model"]
    assert any("overlap" in r.message for r in caplog.records)


def test_resolve_model_project_config_overrides_default():
    from app.modules.llm.llm_client import _resolve_model

    cfg = {"text_cleanup": {"model": "gpt"}}
    assert _resolve_model("text_cleanup", cfg) == "gpt"


def test_resolve_model_project_config_without_model_falls_back_to_default():
    """override dict 에 model key 없으면 default."""
    from app.modules.llm.llm_client import _resolve_model

    cfg = {"text_cleanup": {}}  # model 키 없음
    assert _resolve_model("text_cleanup", cfg) == "gemini-lite"


def test_resolve_model_unknown_step_with_project_config_uses_provided_model():
    """알 수 없는 step 도 project_config 가 있으면 그 값 사용."""
    from app.modules.llm.llm_client import _resolve_model

    cfg = {"random_step": {"model": "gpt-mini"}}
    assert _resolve_model("random_step", cfg) == "gpt-mini"


# ---------------------------------------------------------------------------
# 4. dead code 제거 (llm_router.py)
# ---------------------------------------------------------------------------


def test_llm_router_module_removed():
    """problems.md #5 — dead code llm_router.py 삭제."""
    import importlib

    with pytest.raises(ImportError):
        importlib.import_module("app.modules.llm.llm_router")


def test_no_residual_imports_to_llm_router():
    """app/ 하위 어떤 파일도 llm_router 를 import 하지 않음 (회귀 가드)."""
    from pathlib import Path

    backend_root = Path(__file__).resolve().parent.parent.parent
    app_root = backend_root / "app"
    offenders = []
    for py in app_root.rglob("*.py"):
        text = py.read_text(encoding="utf-8", errors="ignore")
        if "from app.modules.llm.llm_router" in text or "import app.modules.llm.llm_router" in text:
            offenders.append(str(py))
    assert not offenders, f"llm_router 잔존 import: {offenders}"


# ---------------------------------------------------------------------------
# 5. UI / runtime parity
# ---------------------------------------------------------------------------


def test_ui_and_runtime_share_same_pipeline_steps():
    """UI (api/v1/projects.py) 와 runtime (_resolve_model) 둘 다 같은 PIPELINE_STEPS 사용."""
    # api/v1/projects.py 가 직접 import 하므로 같은 dict object 보장.
    from app.api.v1 import projects as projects_api
    from app.modules.llm import llm_client

    # api 측에서 PIPELINE_STEPS 를 가져왔는지 source 확인 (정적).
    src = (
        __import__("pathlib").Path(projects_api.__file__).read_text(encoding="utf-8")
    )
    assert "from app.modules.llm.llm_client import PIPELINE_STEPS" in src, (
        "UI 가 PIPELINE_STEPS 를 다른 source 에서 import"
    )
    # 같은 dict 인지 sanity (UI 가 _resolve_model 호출 시 동일 view).
    from app.modules.llm.llm_client import PIPELINE_STEPS, _resolve_model

    sample = next(iter(PIPELINE_STEPS))
    assert _resolve_model(sample) == PIPELINE_STEPS[sample]["default"]


def test_pdf_data_url_steps_require_gemini_alias():
    """PDF data-URL capability 계약 (Codex 가드, 2026-07-10 2회차 E2E 실측).

    text_cleanup/planning_doc_analysis 는 PDF 를 data:application/pdf;base64
    image_url 로 LLM 에 직접 전달한다 — OpenAI 는 "Invalid MIME type" 거부라
    이 두 스텝의 resolved alias 는 반드시 Gemini 계열이어야 한다. 값 핀이
    아니라 capability 계약: 다음 일괄 모델 교체 때 이 스텝을 OpenAI 로
    옮기면 여기서 막힌다 (옮기려면 PDF 전달 경로 자체를 바꿔야 함).
    """
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import _GEMINI_ALIASES, _resolve_model

    for sid in ("text_cleanup", "planning_doc_analysis"):
        alias = _resolve_model(sid)
        assert alias in _GEMINI_ALIASES, (
            f"{sid} 는 PDF data-URL multimodal 경로 — Gemini alias 필수, "
            f"got {alias!r}"
        )
        assert STEP_MANIFEST[sid]["provider"] == "gemini", sid


def test_gpt_mini_rows_have_gemini_provider():
    """provider SOT 정합 (Codex NARROW 2026-07-11): gpt-mini alias 의 물리
    provider 는 Gemini flash — manifest provider/nested sub_step/AVAILABLE_MODELS
    가 전부 gemini 로 일치해야 split-brain(router=Gemini, catalog=OpenAI) 방지."""
    from app.core.step_manifest import STEP_MANIFEST
    from app.modules.llm.llm_client import AVAILABLE_MODELS

    for sid, info in STEP_MANIFEST.items():
        if info.get("default_model") == "gpt-mini":
            assert info.get("provider") == "gemini", sid
        for sub in (info.get("sub_steps") or []):
            if isinstance(sub, dict) and sub.get("default_model") == "gpt-mini":
                assert sub.get("provider") == "gemini", (sid, sub.get("id"))
    alias_provider = {m["alias"]: m["provider"] for m in AVAILABLE_MODELS}
    assert alias_provider.get("gpt-mini") == "gemini"