Aislo/resources/tester/test_m02_item_inputs.py

"""M02 항목 선택 입력(PLAN 25-1b · 25-2a) — 입력 목록 · 항목 연결 파일 · 알약 · 권한.

① 입력 목록은 시험 안 마스터 견본(`fixtures/m02_master_sample` 사본 · PLAN 26-5a)으로 —
   일위대가 있는 항목 셋에서 로직 수 · 입력 수 · 자동값이 M01 조합 미리 보기가 쓰는 것(로직 줄
   `입력` · `master_auto`)과 같은지 · 로직마다 갈라 합치지 않는지 · 일위대가 없는 항목은 빈 목록.
② 연결 파일 쓰기는 임시 storage(회사 · 개인 층) + 견본 시스템 집계표로만 — 판 409 · 없는 키 422 ·
   사라짐 표시 · 알약 · 층 권한 · 열 지우기 · 항목 복사 · 분류 더하기 복사가 파일을 같이 다룸.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from common_util.common_util_auth import verify_session
from config import config_system
from M01_MasterData import M01_MasterData_Store as m01
from M01_MasterData import M01_MasterData_Store_Combo as combo_store
from M02_MasterTemplete import M02_MasterTemplete_Inputs as inputs_store
from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layers_router
from M02_MasterTemplete import M02_MasterTemplete_Store as store
from M02_MasterTemplete import M02_Template_Layers as layers
from resources.tester.helper_m02_master_sample import copy_master_sample
from resources.tester.test_m02_structure import write_sample_table

TABLE = "table/구조물집계표"
UA = "UA000004"  # 견본 항목 셋(rv · c4 · c1)이 쓰는 조합 — 담은 로직 여럿
OTHER_UA = "UA000007"


def _app(session: dict[str, Any]) -> TestClient:
    app = FastAPI()
    app.include_router(layers_router.router)
    app.include_router(items_router.router)
    app.dependency_overrides[verify_session] = lambda: session
    return TestClient(app)


def _m01_view(ua: str) -> list[tuple[str, list[str], dict]]:
    """M01 조합 미리 보기가 그리는 것 — 담은 로직마다 (키, 입력 이름들, 자동값)."""
    row = next(r for r in m01.read(combo_store.FILE)[0]["줄"] if r["키"] == ua)
    whole = m01.loaded()[1]
    out = []
    for one in row["담은로직"]:
        try:
            logic = whole.logic(one["로직"])
        except combo_store.cm.mf.FormulaError:
            continue  # 화면도 못 읽은 로직은 건너뜀
        auto = inputs_store._num(m01.auto(one["로직"])["값"])
        out.append((one["로직"], [s["이름"] for s in logic.get("입력") or []], auto))
    return out


# ── ① 입력 목록 — 마스터 견본 사본 ─────────────────────


def test_입력_목록은_M01_조합_미리_보기와_같음(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setattr(store, "FOLDER", copy_master_sample(tmp_path / "master"))
    client = _app({"user_id": 1, "company_id": 1, "role": "USER", "is_master": False})
    rows = [r for r in json.loads((store.FOLDER / store.COMBO_FILE).read_bytes())["줄"]]
    with_ua = [r for r in rows if r.get("일위대가")]
    picked, seen = [], set()
    for r in with_ua:  # 서로 다른 조합 셋
        if r["일위대가"] not in seen:
            picked.append(r)
            seen.add(r["일위대가"])
        if len(picked) == 3:
            break
    assert len(picked) == 3
    for r in picked:
        res = client.get(f"/api/m02/layers/system/items/{r['키']}/inputs")
        assert res.status_code == 200, res.text  # 일반 사용자 읽기
        got = res.json()
        want = _m01_view(r["일위대가"])
        assert got["일위대가"] == r["일위대가"] and got["이름"]
        assert [lg["로직"] for lg in got["로직"]] == [k for k, _, _ in want]
        for logic, (_, names, auto) in zip(got["로직"], want):
            assert [i["이름"] for i in logic["입력"]] == names  # 로직마다 · 합치지 않음
            assert {i["이름"]: i["자동값"] for i in logic["입력"]} == auto
            assert all({"이름", "설명", "단위", "자동값"} <= set(i) for i in logic["입력"])
    # 이름이 같은 입력이 두 로직에 있어도 두 번 나옴
    names = [
        i["이름"]
        for r in with_ua
        for lg in client.get(f"/api/m02/layers/system/items/{r['키']}/inputs").json()["로직"]
        for i in lg["입력"]
    ]
    assert len(names) > len(set(names))
    no_ua = next(r for r in rows if not r.get("일위대가"))
    empty = client.get(f"/api/m02/layers/system/items/{no_ua['키']}/inputs").json()
    assert empty == {"일위대가": None, "이름": None, "로직": [], "빠진로직": []}
    assert client.get("/api/m02/layers/system/items/SS999999/inputs").status_code == 404


# ── ② 연결 파일 — 임시 storage ─────────────────────────


@pytest.fixture()
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
    system = tmp_path / "system"
    system.mkdir()
    write_sample_table(system)
    monkeypatch.setattr(store, "FOLDER", system)
    monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
    store.renumber()
    storage = tmp_path / "storage"
    storage.mkdir()
    monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
    session: dict[str, Any] = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
    return {"client": _app(session), "session": session, "storage": storage}


def _as(world: dict, role: str = "USER", user: int = 42) -> TestClient:
    world["session"].update(role=role, user_id=user)
    return world["client"]


def _put_table(client: TestClient, layer: str, edit) -> None:
    url = f"/api/m02/layers/{layer}/templates/{TABLE}"
    got = client.get(url).json()
    edit(got["문서"]["열"])
    res = client.put(url, json={"판": got["판"], "문서": got["문서"]})
    assert res.status_code == 200, res.text


def _col(cid: str, ua: str | None = UA) -> dict[str, Any]:
    return {
        "id": cid,
        "머리": ["sub4 돌쌓기", cid, None, None],
        "단위": "㎡",
        "꼴": "수",
        "일위대가": ua,
    }


def _key(client: TestClient, layer: str, cid: str) -> str:
    items = client.get(f"/api/m02/items?layer={layer}").json()
    return next(r["key"] for r in items if r["열"] == cid)


def test_연결_저장_다시_읽기_판_422_사라짐_알약(world):
    client = _as(world)
    _put_table(client, "personal", lambda cols: cols.append(_col("sub4_a")))
    key = _key(client, "personal", "sub4_a")
    base = f"/api/m02/layers/personal/items/{key}"
    listing = client.get(f"{base}/inputs").json()
    first, second = listing["로직"][0], listing["로직"][1]
    assert client.get(f"{base}/input-links").json() == {"판": None, "줄": []}
    rows = [
        {"로직": first["로직"], "입력": first["입력"][0]["이름"], "자리": "면적"},
        {"로직": second["로직"], "입력": second["입력"][0]["이름"], "자리": "재료"},
        {"로직": first["로직"], "입력": first["입력"][1]["이름"], "자리": "선택"},
    ]
    saved = client.put(f"{base}/input-links", json={"판": None, "줄": rows})
    assert saved.status_code == 200, saved.text
    assert saved.json()["판"] == 1
    again = client.get(f"{base}/input-links").json()
    assert again == saved.json()
    assert again["줄"][0]["알약"] == f"입력|{first['로직']}|{first['입력'][0]['이름']}"
    assert "알약" not in again["줄"][2]  # 선택은 시트 알약이 아님
    file = world["storage"] / "7" / "42" / "templates" / "inputs" / "sub4_a.json"
    assert json.loads(file.read_text("utf-8")) == saved.json()
    # 판 다르면 409 · 없는 키 · 모르는 자리 · 두 번은 422
    assert client.put(f"{base}/input-links", json={"판": None, "줄": rows}).status_code == 409
    for bad in (
        [{"로직": "GF999999", "입력": "높이", "자리": "면적"}],
        [{"로직": first["로직"], "입력": "없는입력", "자리": "면적"}],
        [{"로직": first["로직"], "입력": first["입력"][0]["이름"], "자리": "옆"}],
        [rows[0], rows[0]],
    ):
        res = client.put(f"{base}/input-links", json={"판": 1, "줄": bad})
        assert res.status_code == 422, res.text
    # 알약 — 면적 = 조합 물량 + 면적으로 이은 입력 · 재료 = 재료로 이은 입력만
    area = client.get(f"{base}/pills?place=면적").json()
    assert area["물량"] and [p["키"] for p in area["입력"]] == [again["줄"][0]["알약"]]
    mat = client.get(f"{base}/pills?place=재료").json()
    assert mat["물량"] == [] and [p["입력"] for p in mat["입력"]] == [second["입력"][0]["이름"]]
    assert " · " in mat["입력"][0]["이름"]  # 입력 · 로직 이름(합치지 않음)
    # 일위대가가 바뀌면 옛 로직 연결은 읽을 때 사라짐 · 알약에서 빠짐
    _put_table(client, "personal", lambda cols: cols[-1].update(일위대가=OTHER_UA))
    gone = client.get(f"{base}/input-links").json()["줄"]
    assert all(r.get("사라짐") for r in gone)
    assert client.get(f"{base}/pills?place=면적").json()["입력"] == []


def test_권한_일반_사용자_읽기_쓰기(world):
    admin = _as(world, "SYSTEM_ADMIN")
    _put_table(admin, "system", lambda cols: cols.append(_col("sub4_sys")))
    key = _key(admin, "system", "sub4_sys")
    base = f"/api/m02/layers/system/items/{key}"
    user = _as(world)
    assert user.get(f"{base}/inputs").status_code == 200
    assert user.get(f"{base}/input-links").status_code == 200
    assert user.put(f"{base}/input-links", json={"판": None, "줄": []}).status_code == 403
    assert _as(world, "SYSTEM_ADMIN").put(f"{base}/input-links", json={"줄": []}).status_code == 200
    # 회사 층 — 같은 회사 일반 사용자 읽기 · 쓰기 403 · 회사 관리자 쓰기
    _put_table(_as(world, "ADMIN"), "company", lambda cols: cols.append(_col("sub4_c")))
    ckey = _key(_as(world), "company", "sub4_c")
    cbase = f"/api/m02/layers/company/items/{ckey}"
    assert _as(world).get(f"{cbase}/inputs").status_code == 200
    assert _as(world).put(f"{cbase}/input-links", json={"줄": []}).status_code == 403
    assert _as(world, "ADMIN").put(f"{cbase}/input-links", json={"줄": []}).status_code == 200
    # 남의 개인 층은 못 읽음
    assert _as(world).get(f"{cbase}/inputs?owner=9").status_code == 403


def test_열_지우기_항목_복사_분류_복사가_연결을_같이(world):
    admin = _as(world, "ADMIN")
    _put_table(admin, "company", lambda cols: cols.append(_col("sub4_c")))
    key = _key(admin, "company", "sub4_c")
    base = f"/api/m02/layers/company/items/{key}"
    logic = admin.get(f"{base}/inputs").json()["로직"][0]
    rows = [{"로직": logic["로직"], "입력": logic["입력"][0]["이름"], "자리": "면적"}]
    assert admin.put(f"{base}/input-links", json={"판": None, "줄": rows}).status_code == 200
    # 항목 개인 복사 — 새 열에 연결이 판 1 로
    user = _as(world)
    made = user.post(
        "/api/m02/layers/personal/items/copy", json={"from_layer": "company", "key": key}
    ).json()
    mine = user.get(f"/api/m02/layers/personal/items/{made['key']}/input-links").json()
    assert mine["판"] == 1 and [r["입력"] for r in mine["줄"]] == [rows[0]["입력"]]
    # 분류 더하기(본뜬열) — 새 열이 원본 열 연결을 받음
    admin = _as(world, "ADMIN")
    twin = {**_col("sub4_c2", None), "본뜬열": "sub4_c"}
    _put_table(admin, "company", lambda cols: cols.append(twin))
    folder = world["storage"] / "7" / "templates"
    with store.at(folder):
        assert inputs_store.read_file("sub4_c2")["줄"] == inputs_store.read_file("sub4_c")["줄"]
    # 열 지우기 — 연결 파일도 같이 지움
    _put_table(
        admin, "company", lambda cols: cols.remove(next(c for c in cols if c["id"] == "sub4_c"))
    )
    assert not (folder / "inputs" / "sub4_c.json").is_file()
    assert (folder / "inputs" / "sub4_c2.json").is_file()