Aislo/resources/tester/test_m02_template_layers.py

"""M02 양식 층 — 자리 · 복사 · 초기화 · 회사 적용 · 가져오기 · 권한 (임시 storage)."""

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 import common_util_storage
from common_util.common_util_auth import verify_session
from config import config_system
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as router_module
from M02_MasterTemplete import M02_Template_Layers as layers

P1 = "11111111-1111-1111-1111-111111111111"
P2 = "22222222-2222-2222-2222-222222222222"
P_OTHER = "33333333-3333-3333-3333-333333333333"


@pytest.fixture
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
    storage = tmp_path / "storage"
    storage.mkdir()
    monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
    monkeypatch.setattr(common_util_storage, "STORAGE_BASE_DIR", str(storage))
    monkeypatch.setattr(layers, "SYSTEM_ROOT", tmp_path / "master_template")
    layers.write_template(layers.system_dir(), "table", "구조물집계표", {"판": 1, "열": []})
    layers.write_template(layers.system_dir(), "drawing", "A1_도각", {"format": 6})

    projects = {
        P1: {"id": P1, "name": "첫째", "company_id": 7, "user_id": 42},
        P2: {"id": P2, "name": "둘째", "company_id": 7, "user_id": 43},
        P_OTHER: {"id": P_OTHER, "name": "남의 회사", "company_id": 9, "user_id": 90},
    }
    for row in projects.values():
        row["storage_path"] = f"storage/{row['company_id']}/{row['user_id']}/{row['id']}"
        root = Path(common_util_storage.resolve_stored_project_path(row["storage_path"]))
        layers.seed_project(root)
    users = {42: 7, 43: 7, 90: 9}

    async def project_row(project_id: str) -> dict[str, Any] | None:
        found = projects.get(str(project_id))
        return dict(found) if found else None

    async def user_company(user_id: int) -> int | None:
        return users.get(user_id)

    async def company_users(company_id: int) -> list[dict[str, Any]]:
        return [{"user_id": u, "name": f"u{u}"} for u, c in users.items() if c == company_id]

    async def company_projects(company_id: int) -> list[dict[str, Any]]:
        return [
            {"project_id": p["id"], "name": p["name"], "storage_path": p["storage_path"]}
            for p in projects.values()
            if p["company_id"] == company_id
        ]

    monkeypatch.setattr(router_module, "_project_row", project_row)
    monkeypatch.setattr(router_module, "_user_company", user_company)
    monkeypatch.setattr(router_module, "_company_users", company_users)
    monkeypatch.setattr(router_module, "_company_projects", company_projects)

    session = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
    app = FastAPI()
    app.include_router(router_module.router)
    app.dependency_overrides[verify_session] = lambda: session
    return {"client": TestClient(app), "session": session, "storage": storage}


def _root(world: dict[str, Any], project_id: str, company: int = 7, user: int = 42) -> Path:
    return (world["storage"] / str(company) / str(user) / project_id).resolve()


def _url(project_id: str = P1) -> str:
    return f"/api/m02/layers/project/templates/table/구조물집계표?project_id={project_id}"


def test_프로젝트_만들기_복사는_작업본과_초기본_둘에_manifest(world: dict[str, Any]) -> None:
    root = _root(world, P1)
    assert (root / "templates/table/구조물집계표.json").is_file()
    assert (root / "templates/_initial/drawing/A1_도각.json").is_file()
    entry = layers.read_manifest(root / "templates")["table/구조물집계표"]
    assert entry["층"] == "system" and entry["판"] == layers.version_of(
        layers.system_dir() / "table/구조물집계표.json"
    )


def test_옛_프로젝트_넣기는_더하기만(world: dict[str, Any]) -> None:
    root = _root(world, P1)
    layers.write_template(root / "templates", "table", "구조물집계표", {"고침": True})
    assert layers.seed_project(root, only_missing=True) == {"작업본": [], "초기": []}
    assert layers.read_template(root / "templates", "table", "구조물집계표")["문서"] == {
        "고침": True
    }


def test_작업본_저장_판_다르면_409_초기화는_초기본으로(world: dict[str, Any]) -> None:
    client = world["client"]
    got = client.get(_url()).json()
    assert got["층"] == "project" and got["문서"]["판"] == 1
    saved = client.put(_url(), json={"판": got["판"], "문서": {"판": 2}})
    assert saved.status_code == 200
    stale = client.put(_url(), json={"판": got["판"], "문서": {"판": 3}})
    assert stale.status_code == 409
    reset = client.post(f"/api/m02/projects/{P1}/templates/reset", json={})
    assert "table/구조물집계표" in reset.json()["초기화"]
    assert client.get(_url()).json()["문서"] == {"판": 1, "열": []}
    initial = _root(world, P1) / "templates/_initial/table/구조물집계표.json"
    assert layers.read_template(initial.parent.parent, "table", "구조물집계표")["문서"]["판"] == 1


def test_회사_공식_저장은_관리자만_적용은_작업본만(world: dict[str, Any]) -> None:
    client, session = world["client"], world["session"]
    client.put(_url(), json={"판": client.get(_url()).json()["판"], "문서": {"회사": 1}})
    body = {"to": "company", "종류": "table", "이름": "구조물집계표"}
    assert client.post(f"/api/m02/projects/{P1}/templates/save-as", json=body).status_code == 403
    session["role"] = "ADMIN"
    assert client.post(f"/api/m02/projects/{P1}/templates/save-as", json=body).status_code == 200
    applied = client.post(f"/api/m02/projects/{P2}/templates/apply", json={"from": "company"})
    assert applied.status_code == 200
    root2 = _root(world, P2, user=43)
    assert layers.read_template(root2 / "templates", "table", "구조물집계표")["문서"] == {"회사": 1}
    assert (
        layers.read_template(root2 / "templates/_initial", "table", "구조물집계표")["문서"]["판"]
        == 1
    )
    assert layers.read_manifest(root2 / "templates")["table/구조물집계표"]["층"] == "company"


def test_개인_양식은_본인만_쓰고_같은_회사는_읽어_가져옴(world: dict[str, Any]) -> None:
    client, session = world["client"], world["session"]
    body = {"to": "personal", "종류": "table", "이름": "구조물집계표"}
    assert client.post(f"/api/m02/projects/{P1}/templates/save-as", json=body).status_code == 200
    mine = f"/api/m02/layers/personal/templates/table/구조물집계표?project_id={P1}"
    assert client.get(mine).status_code == 200
    session.update(user_id=43)
    theirs = mine + "&user_id=42"
    assert client.get(theirs).status_code == 200
    put = client.put(
        f"/api/m02/layers/personal/templates/table/구조물집계표?project_id={P1}&user_id=42",
        json={"판": None, "문서": {}},
    )
    assert put.status_code == 200  # user_id 는 쓰기에서 무시 — 본인(43) 자리에 새로 씀
    assert (world["storage"] / "7/43/templates/table/구조물집계표.json").is_file()
    got = client.post(
        f"/api/m02/projects/{P2}/templates/apply", json={"from": "personal", "user_id": 42}
    )
    assert got.status_code == 200
    assert client.get(mine.replace(P1, P_OTHER)).status_code == 403
    far = client.post(
        f"/api/m02/projects/{P2}/templates/apply", json={"from": "personal", "user_id": 90}
    )
    assert far.status_code == 403


def test_가져오기는_같은_회사_프로젝트만_목록도(world: dict[str, Any]) -> None:
    client = world["client"]
    ok = client.post(
        f"/api/m02/projects/{P1}/templates/apply", json={"from": "project", "project_id": P2}
    )
    assert ok.status_code == 200
    bad = client.post(
        f"/api/m02/projects/{P1}/templates/apply", json={"from": "project", "project_id": P_OTHER}
    )
    assert bad.status_code == 403
    sources = client.get(f"/api/m02/projects/{P1}/sources").json()
    assert [row["project_id"] for row in sources["project"]] == [P2]
    assert {row["이름"] for row in sources["system"]} == {"구조물집계표", "A1_도각"}


def test_시스템_층은_읽기만_이름은_막음(world: dict[str, Any]) -> None:
    client = world["client"]
    assert client.get("/api/m02/layers/system/templates").json()["양식"]
    put = client.put(
        "/api/m02/layers/system/templates/table/구조물집계표", json={"판": None, "문서": {}}
    )
    assert put.status_code == 403
    for name in ("..", "_initial", "a.b/c"):
        with pytest.raises(ValueError):
            layers.template_path(layers.system_dir(), "table", name)


def test_채운_표는_작업본을_채워_돌려주고_저장_안_함(
    world: dict[str, Any], monkeypatch: pytest.MonkeyPatch
) -> None:
    async def designs(project_id: str) -> list[dict[str, Any]]:
        return [{"chainage_m": 40.0, "design": {"pipe_length_m": 12}}]

    monkeypatch.setattr(router_module, "_cross_designs", designs)
    monkeypatch.setattr(router_module.fill, "recalc", lambda document: {"계산": {}})
    root = _root(world, P1)
    master = config_system.PROJECT_ROOT / "resources/master_template/table/구조물집계표.json"
    layers.write_template(
        root / "templates", "table", "구조물집계표", json.loads(master.read_text(encoding="utf-8"))
    )
    edits = root / "B04_PreProcess/drainage/edits"
    edits.mkdir(parents=True, exist_ok=True)
    (edits / "pipe_points.json").write_text(
        json.dumps({"points": [{"chainage_m": 40.0, "source": "user"}]}), encoding="utf-8"
    )
    before = layers.version_of(root / "templates/table/구조물집계표.json")
    got = world["client"].get(f"/api/m02/projects/{P1}/tables/구조물집계표/filled").json()
    row = next(r for r in got["문서"]["줄"] if r["id"] == "s40.00")
    assert row["값"]["sta"] == "NO.2" and row["값"]["pp_len|파형강관|1000"] == 12
    assert got["결과"] == {"계산": {}} and got["판"] == before
    assert layers.version_of(root / "templates/table/구조물집계표.json") == before


def test_프로젝트_만들기_자리에서_시스템_양식을_복사(world: dict[str, Any], tmp_path: Path) -> None:
    from B02_ProjRegister.B02_ProjRegister_Repository import _initialize_project_storage

    layers.write_template(layers.company_dir(7), "table", "구조물집계표", {"회사": 1})
    root = tmp_path / "new_project"
    _initialize_project_storage(root, "new")
    got = layers.read_template(root / "templates", "table", "구조물집계표")
    assert got["문서"] == {"판": 1, "열": []}  # 회사 공식이 있어도 시스템
    assert (root / "templates/_initial/drawing/A1_도각.json").is_file()
    assert (root / "project_manifest.json").is_file()