"""Tests for FloorPlanRenderStep — Phase 7 Step 4 step entry."""
from __future__ import annotations

from types import SimpleNamespace
from typing import Any, Dict, List
from unittest.mock import MagicMock, patch

from app.core.steps.floor_plan_render_step import FloorPlanRenderStep


def test_step_disabled_when_mode_off():
    """background_mode='off' 이면 applicable_count=0 (step skip)."""
    with patch("app.core.config.settings.background_mode", "off"):
        step = FloorPlanRenderStep.__new__(FloorPlanRenderStep)
        step.project_id = "p"
        step.episode_id = "e"
        result = step._execute()
        assert result["applicable_count"] == 0
        assert result["completed_count"] == 0
        assert result["failed_count"] == 0
        assert result["data"] == {}


# ─────────────────────────────────────────────
# B4 — ImageAsset UPSERT 단위 테스트
# ─────────────────────────────────────────────

def _make_step_with_db(monkeypatch, tmp_path, *, canons=None, existing=None):
    """ImageAsset UPSERT 테스트용 step + mock db.

    canons: List of mock EntityCanon (short_id+id). 기본 [L01].
    existing: query(ImageAsset).filter_by(...).first() 반환값. 기본 None (insert path).
    """
    if canons is None:
        canon = MagicMock()
        canon.short_id = "L01"
        canon.id = "canon-l01"
        canons = [canon]

    monkeypatch.setattr(
        "app.core.config.settings.projects_dir", str(tmp_path / "p")
    )
    (tmp_path / "p").mkdir(parents=True, exist_ok=True)

    step = FloorPlanRenderStep.__new__(FloorPlanRenderStep)
    step.project_id = "proj-test"
    step.episode_id = "ep1"
    step.db = MagicMock()

    canon_query = MagicMock()
    canon_query.filter.return_value.all.return_value = canons
    asset_query = MagicMock()
    asset_query.filter_by.return_value.first.return_value = existing

    def _query_router(model):
        from app.models.project import EntityCanon as _EC, ImageAsset as _IA
        if model is _EC:
            return canon_query
        if model is _IA:
            return asset_query
        return MagicMock()

    step.db.query.side_effect = _query_router
    return step


def test_b4_register_floor_plan_image_asset_basic_insert(tmp_path, monkeypatch):
    """ok 1개 fp → ImageAsset insert 1회 (asset_type/entity/variant 검증)."""
    step = _make_step_with_db(monkeypatch, tmp_path)

    png = tmp_path / "p" / "img.png"
    png.write_bytes(b"PNG")

    results = {
        "fp_living": {
            "status": "ok",
            "png_path": str(png),
            "t2i_prompt": "a top-down floor plan",
        },
    }
    items = {
        "fp_living": {
            "parent_id": "",
            "spec": {"fp_id": "fp_living", "sub_location": "living"},
            "group_id": "g1",
            "primary_loc_id": "L01",
        },
    }

    added: List[Any] = []
    step.db.add = lambda obj: added.append(obj)

    step._register_image_assets(results, items)

    assert len(added) == 1
    asset = added[0]
    assert asset.asset_type == "floor_plan"
    assert asset.entity_id == "canon-l01"
    assert asset.episode_id == "ep1"
    assert asset.project_id == "proj-test"
    assert asset.variant_index == 0
    assert asset.variant_label == "v00"
    assert asset.variant_type == "fp_living"
    assert asset.is_primary == 1
    assert asset.t2i_guide is None
    assert asset.prompt_used == "a top-down floor plan"
    assert asset.generation_model == "gpt-image-2.5-sunburst"
    assert asset.status == "generated"
    step.db.commit.assert_called_once()


def test_b4_register_floor_plan_skips_failed(tmp_path, monkeypatch):
    """status=failed 그룹은 ImageAsset 생성 X."""
    step = _make_step_with_db(monkeypatch, tmp_path)

    results = {
        "fp_failed": {
            "status": "failed",
            "png_path": "",
            "t2i_prompt": "x",
        },
    }
    items = {
        "fp_failed": {
            "parent_id": "",
            "spec": {"fp_id": "fp_failed"},
            "group_id": "g1",
            "primary_loc_id": "L01",
        },
    }

    added: List[Any] = []
    step.db.add = lambda obj: added.append(obj)

    step._register_image_assets(results, items)
    assert added == []
    # commit은 registered=0이므로 호출되지 않음
    step.db.commit.assert_not_called()


def test_b4_register_floor_plan_idempotent_upsert(tmp_path, monkeypatch):
    """existing row 있을 때 UPDATE만 (db.add 호출 0)."""
    existing = MagicMock()
    existing.file_path = "old"
    step = _make_step_with_db(monkeypatch, tmp_path, existing=existing)

    png = tmp_path / "p" / "img.png"
    png.write_bytes(b"PNG")

    results = {
        "fp_living": {
            "status": "ok",
            "png_path": str(png),
            "t2i_prompt": "new prompt",
        },
    }
    items = {
        "fp_living": {
            "parent_id": "",
            "spec": {"fp_id": "fp_living"},
            "group_id": "g1",
            "primary_loc_id": "L01",
        },
    }

    added: List[Any] = []
    step.db.add = lambda obj: added.append(obj)

    step._register_image_assets(results, items)

    assert added == []  # UPDATE path — no db.add
    assert existing.prompt_used == "new prompt"
    assert existing.variant_label == "v00"
    assert existing.variant_index == 0
    assert existing.is_primary == 1
    assert existing.t2i_guide is None
    assert existing.status == "generated"
    step.db.commit.assert_called_once()


def test_b4_register_floor_plan_multi_fp_per_location(tmp_path, monkeypatch):
    """동일 primary_loc_id 의 multiple fp → variant_index 0/1 (sorted fp_id)."""
    step = _make_step_with_db(monkeypatch, tmp_path)

    png_a = tmp_path / "p" / "a.png"
    png_a.write_bytes(b"A")
    png_b = tmp_path / "p" / "b.png"
    png_b.write_bytes(b"B")

    # 입력 dict 순서가 sort 결과와 다르도록 의도적으로 거꾸로
    results = {
        "fp_kitchen": {"status": "ok", "png_path": str(png_b), "t2i_prompt": "K"},
        "fp_bedroom": {"status": "ok", "png_path": str(png_a), "t2i_prompt": "B"},
    }
    items = {
        "fp_kitchen": {
            "parent_id": "", "spec": {"fp_id": "fp_kitchen"}, "group_id": "g1",
            "primary_loc_id": "L01",
        },
        "fp_bedroom": {
            "parent_id": "", "spec": {"fp_id": "fp_bedroom"}, "group_id": "g1",
            "primary_loc_id": "L01",
        },
    }

    added: List[Any] = []
    step.db.add = lambda obj: added.append(obj)

    step._register_image_assets(results, items)

    assert len(added) == 2
    by_fp = {a.variant_type: a for a in added}
    # sorted: fp_bedroom(0/v00/primary), fp_kitchen(1/v01/non-primary)
    assert by_fp["fp_bedroom"].variant_index == 0
    assert by_fp["fp_bedroom"].variant_label == "v00"
    assert by_fp["fp_bedroom"].is_primary == 1
    assert by_fp["fp_kitchen"].variant_index == 1
    assert by_fp["fp_kitchen"].variant_label == "v01"
    assert by_fp["fp_kitchen"].is_primary == 0


def test_b4_register_floor_plan_no_canon_skips(tmp_path, monkeypatch):
    """EntityCanon 없으면 graceful skip — db.add/commit 호출 X."""
    step = _make_step_with_db(monkeypatch, tmp_path, canons=[])
    png = tmp_path / "p" / "img.png"
    png.write_bytes(b"PNG")

    results = {
        "fp_living": {"status": "ok", "png_path": str(png), "t2i_prompt": "x"},
    }
    items = {
        "fp_living": {
            "parent_id": "", "spec": {"fp_id": "fp_living"}, "group_id": "g1",
            "primary_loc_id": "L01",
        },
    }
    added: List[Any] = []
    step.db.add = lambda obj: added.append(obj)

    step._register_image_assets(results, items)
    assert added == []
    step.db.commit.assert_not_called()


def test_w20e6c_register_floor_plan_no_primary_loc_id_skips_with_info_log(
    tmp_path, monkeypatch, caplog,
):
    """W20E6-C: fp with empty primary_loc_id (no master_plan bg refs it)
    is skipped from ImageAsset registration. The skip surfaces as an
    INFO-level log carrying ``no_primary_loc_id`` for auditability,
    distinguishing it from the existing ``missing canon`` warning path.
    No db.add / commit calls are issued for the skipped fp.
    """
    import logging

    step = _make_step_with_db(monkeypatch, tmp_path)
    png = tmp_path / "p" / "img.png"
    png.write_bytes(b"PNG")

    results = {
        "fp_orphan": {
            "status": "ok", "png_path": str(png), "t2i_prompt": "x",
        },
    }
    items = {
        "fp_orphan": {
            "parent_id": "",
            "spec": {"fp_id": "fp_orphan"},
            "group_id": "g1",
            "primary_loc_id": "",  # ← no bg depends_on_fp this fp
        },
    }
    added: List[Any] = []
    step.db.add = lambda obj: added.append(obj)

    with caplog.at_level(
        logging.INFO,
        logger="app.core.steps.floor_plan_render_step",
    ):
        step._register_image_assets(results, items)

    assert added == []
    step.db.commit.assert_not_called()
    assert any(
        "no_primary_loc_id" in rec.message and "fp_orphan" in rec.message
        for rec in caplog.records
    ), caplog.records


# ─────────────────────────────────────────────
# Carry B — _compute_fp_primary_loc_ids helper (역참조 단위 테스트)
# ─────────────────────────────────────────────

def test_compute_fp_primary_loc_ids_basic_reverse_ref():
    """ref background 있는 fp → 그 loc_id, ref background 0 인 fp → 빈값."""
    from app.core.steps.floor_plan_render_step import _compute_fp_primary_loc_ids

    plans_map = {
        "g1": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_a"}, {"fp_id": "fp_b"}],
            "backgrounds": [
                {"bg_id": "bg1", "loc_id": "L01", "depends_on_fp": ["fp_a"]},
            ],
        }},
    }
    assert _compute_fp_primary_loc_ids(plans_map) == {"fp_a": "L01", "fp_b": ""}


def test_compute_fp_primary_loc_ids_multi_loc_sorted():
    """여러 loc 의 background 가 같은 fp 를 ref → 정렬 첫 값 (deterministic)."""
    from app.core.steps.floor_plan_render_step import _compute_fp_primary_loc_ids

    plans_map = {
        "g1": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_x"}],
            "backgrounds": [
                {"bg_id": "b1", "loc_id": "L05", "depends_on_fp": ["fp_x"]},
                {"bg_id": "b2", "loc_id": "L02", "depends_on_fp": ["fp_x"]},
            ],
        }},
    }
    assert _compute_fp_primary_loc_ids(plans_map) == {"fp_x": "L02"}


def test_compute_fp_primary_loc_ids_first_wins_on_dup_fp_id():
    """동일 fp_id 가 여러 group 에 등장 → 먼저 등장한 group 의 loc (first-wins)."""
    from app.core.steps.floor_plan_render_step import _compute_fp_primary_loc_ids

    plans_map = {
        "g1": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_d"}],
            "backgrounds": [
                {"bg_id": "b1", "loc_id": "L01", "depends_on_fp": ["fp_d"]},
            ],
        }},
        "g2": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_d"}],
            "backgrounds": [
                {"bg_id": "b2", "loc_id": "L09", "depends_on_fp": ["fp_d"]},
            ],
        }},
    }
    assert _compute_fp_primary_loc_ids(plans_map) == {"fp_d": "L01"}


def test_compute_fp_primary_loc_ids_scopes_backgrounds_per_group():
    """fp 의 primary_loc 는 그 fp 가 속한 group 의 background 만으로 결정.

    다른 group 의 background 가 같은 fp_id 를 depends_on_fp 로 참조해도
    무시된다 (group 경계 안에서만 역참조).
    """
    from app.core.steps.floor_plan_render_step import _compute_fp_primary_loc_ids

    plans_map = {
        "g1": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_y"}],
            "backgrounds": [],  # g1 엔 fp_y 를 ref 하는 bg 없음
        }},
        "g2": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_other"}],
            # g2 의 bg 가 fp_y 를 ref 하지만 fp_y 는 g1 소속 → 무시
            "backgrounds": [
                {"bg_id": "b1", "loc_id": "L99", "depends_on_fp": ["fp_y"]},
            ],
        }},
    }
    result = _compute_fp_primary_loc_ids(plans_map)
    assert result["fp_y"] == ""


def test_compute_fp_primary_loc_ids_skips_non_ok_group():
    """status != 'ok' 인 group 의 floor_plans 는 결과에서 제외."""
    from app.core.steps.floor_plan_render_step import _compute_fp_primary_loc_ids

    plans_map = {
        "g1": {"status": "failed", "plan": {
            "floor_plans": [{"fp_id": "fp_skip"}],
            "backgrounds": [
                {"bg_id": "b1", "loc_id": "L01", "depends_on_fp": ["fp_skip"]},
            ],
        }},
        "g2": {"status": "ok", "plan": {
            "floor_plans": [{"fp_id": "fp_keep"}],
            "backgrounds": [
                {"bg_id": "b2", "loc_id": "L02", "depends_on_fp": ["fp_keep"]},
            ],
        }},
    }
    assert _compute_fp_primary_loc_ids(plans_map) == {"fp_keep": "L02"}


# ─────────────────────────────────────────────
# Carry B — _execute 가 helper 추출 후에도 동일한 primary_loc_id 전달 (불변 가드)
# ─────────────────────────────────────────────

def test_execute_passes_correct_primary_loc_id_to_register(tmp_path, monkeypatch):
    """helper 추출 후 _execute 가 _register_image_assets 에 넘기는 items 의
    primary_loc_id 가 기존과 동일 — fp_a(ref bg L01)='L01', fp_b(ref bg 0)=''."""
    plans_cp = {"data": {"plans": {"g1": {"status": "ok", "plan": {
        "floor_plans": [
            {"fp_id": "fp_a", "depends_on_fp": []},
            {"fp_id": "fp_b", "depends_on_fp": []},
        ],
        "backgrounds": [
            {"bg_id": "bg1", "loc_id": "L01", "depends_on_fp": ["fp_a"]},
        ],
    }}}}}
    prompts_cp = {"data": {"floor_plans": {
        "fp_a": {"status": "ok", "t2i_prompt": "pa"},
        "fp_b": {"status": "ok", "t2i_prompt": "pb"},
    }}}

    monkeypatch.setattr("app.core.config.settings.background_mode", "on")
    monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path))

    step = FloorPlanRenderStep.__new__(FloorPlanRenderStep)
    step.project_id = "proj"
    step.episode_id = "ep"
    step.db = MagicMock()

    cps = {"floor_plan_prompt": prompts_cp, "background_master_plan": plans_cp}
    step._load_prev_checkpoint = lambda step_id: cps.get(step_id)

    captured: Dict[str, Any] = {}
    step._register_image_assets = (
        lambda results, items: captured.update(items=items)
    )

    fake = SimpleNamespace(
        status="ok", png_path=str(tmp_path / "fp.png"),
        attempts=1, ref_used="text_only", error="",
    )
    with patch(
        "app.modules.pipeline.floor_plan_render.render_one_floor_plan",
        return_value=fake,
    ), patch(
        "app.core.steps.floor_plan_render_step._resolve_openai_client",
        return_value=MagicMock(),
    ):
        result = step._execute()

    items = captured["items"]
    assert items["fp_a"]["primary_loc_id"] == "L01"
    assert items["fp_b"]["primary_loc_id"] == ""
    assert result["failed_count"] == 0


def test_resolve_openai_client_passes_api_key_and_timeout(monkeypatch):
    """_resolve_openai_client는 키를 **명시 전달**해야 한다.

    회귀 가드: bare OpenAI()는 os.environ["OPENAI_API_KEY"]만 읽어 .env에서
    로드된 키도, 2슬롯 failover 로 전환된 키도 인식하지 못한다.
    2026-07-30 부터 키는 `app.core.openai_keys` 브로커가 정하고,
    실제 SDK 클라이언트는 첫 호출 시점에 활성 슬롯 키로 만들어진다.
    """
    import openai

    from app.core.steps import floor_plan_render_step

    captured: Dict[str, Any] = {}

    class _FakeOpenAI:
        def __init__(self, **kwargs):
            captured.update(kwargs)
            self.images = SimpleNamespace(
                generate=lambda **kw: {"ok": True})

    monkeypatch.setattr(openai, "OpenAI", _FakeOpenAI)
    from types import SimpleNamespace  # noqa: F401 (아래 _FakeOpenAI)
    from app.core import openai_keys

    openai_keys.reset()
    monkeypatch.setattr("app.core.config.settings.openai_api_key", "sk-floorplan-test")
    monkeypatch.setattr(
        "app.core.config.settings.openai_api_key_secondary", "")
    monkeypatch.setattr("app.core.config.settings.llm_timeout_image_gen", 420)

    client = floor_plan_render_step._resolve_openai_client()
    # 브로커 프록시는 **첫 호출 시** 실제 SDK 클라이언트를 만든다 —
    # 키가 명시 전달되는지는 그 시점에 드러난다.
    client.images.generate(model="gpt-image-2.5-sunburst")

    assert captured["api_key"] == "sk-floorplan-test"
    assert captured["timeout"] == 420.0