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()


def test_옛_프로젝트_넣기_도구는_더하기만_폴더_없으면_건너뜀(world: dict[str, Any]) -> None:
    from M02_MasterTemplete import M02_Template_Migrate as migrate

    old = world["storage"] / "7/42/old-project"
    (old / "B05_Profile").mkdir(parents=True)
    (old / "B05_Profile/keep.txt").write_text("x", encoding="utf-8")
    root1 = _root(world, P1)
    layers.write_template(root1 / "templates", "table", "구조물집계표", {"고침": 1})
    (root1 / "templates/drawing/A1_도각.json").unlink()
    projects = [
        {
            "id": "old",
            "name": "옛",
            "company_id": 7,
            "user_id": 42,
            "storage_path": "storage/7/42/old-project",
        },
        {
            "id": P1,
            "name": "첫째",
            "company_id": 7,
            "user_id": 42,
            "storage_path": f"storage/7/42/{P1}",
        },
        {
            "id": "gone",
            "name": "없음",
            "company_id": 7,
            "user_id": 42,
            "storage_path": "storage/7/42/gone",
        },
    ]
    dry = migrate.migrate(projects, dry_run=True)
    assert not (old / "templates").exists() and dry[0]["상태"] == "넣을 것(시험)"
    report = {row["id"]: row for row in migrate.migrate(projects)}
    assert report["old"]["상태"] == "넣음" and (old / "templates/_initial/table").is_dir()
    assert (old / "B05_Profile/keep.txt").read_text(encoding="utf-8") == "x"
    assert report[P1]["작업본"] == ["drawing/A1_도각"]
    assert layers.read_template(root1 / "templates", "table", "구조물집계표")["문서"] == {"고침": 1}
    assert report["gone"]["상태"].startswith("폴더 없음")
    assert not (world["storage"] / "7/42/gone").exists()
    assert migrate.migrate(projects)[0]["상태"] == "이미 있음"
    assert "| 7 | 42 | 옛 |" in migrate.render(list(report.values()), [])


def test_프로젝트_표_저장은_서버가_설계값을_걸러_씀(world: dict[str, Any]) -> None:
    from M02_MasterTemplete import M02_Table_Fill as fill

    client = world["client"]
    master = json.loads(
        (
            config_system.PROJECT_ROOT / "resources/master_template/table/구조물집계표.json"
        ).read_text(encoding="utf-8")
    )
    filled = fill.fill_document(
        master,
        [
            {
                "type_id": "position_sign",
                "placement": "point",
                "chainage_m": 40.0,
                "start_m": None,
                "end_m": None,
                "options": {},
            }
        ],
    )
    filled["줄"][1]["값"]["h_intake"] = 2
    version = client.get(_url()).json()["판"]
    assert client.put(_url(), json={"판": version, "문서": filled}).status_code == 200
    stored = layers.read_template(_root(world, P1) / "templates", "table", "구조물집계표")["문서"]
    assert stored["줄"][1] == {"id": "s40.00", "값": {"sta": "NO.2", "h_intake": 2}}
    assert "알림" not in stored and all("|" not in c["id"] for c in stored["열"])


def test_옛_표_틀은_새_판으로_손_값은_지킴(world: dict[str, Any]) -> None:
    from M02_MasterTemplete import M02_Template_Migrate as migrate

    new = {
        "판": 2,
        "열": [{"id": "a", "식": "1", "펼침틀": {"묶음": "g"}}],
        "줄": [],
        "변수": {"k": 1},
    }
    old = {
        "판": 1,
        "열": [{"id": "a", "식": "1", "바인딩": {"종류": "pipe", "값": "식"}}, {"id": "memo"}],
        "줄": [{"id": "s10.00", "값": {"sta": "NO.0+10", "memo": "손"}}],
        "변수": {"k": 5},
    }
    layers.write_template(layers.system_dir(), "table", "구조물집계표", new)
    root2, root1 = _root(world, P2, user=43), _root(world, P1)
    for folder in (root1 / "templates", root1 / "templates/_initial", root2 / "templates"):
        layers.write_template(folder, "table", "구조물집계표", old)
    layers.write_template(
        root2 / "templates", "table", "구조물집계표", {**old, "줄": [], "변수": {"k": 1}}
    )

    assert migrate.refresh_tables(root1, dry_run=True) == [
        "templates/table/구조물집계표",
        "_initial/table/구조물집계표",
    ]
    migrate.refresh_tables(root1)
    work = layers.read_template(root1 / "templates", "table", "구조물집계표")["문서"]
    assert work["열"] == new["열"] and work["변수"] == {"k": 5}  # 틀만 새 판 · 변수 · 손 값 지킴
    assert work["줄"] == old["줄"]
    system_version = layers.version_of(layers.system_dir() / "table/구조물집계표.json")
    initial = root1 / "templates/_initial"
    assert layers.version_of(initial / "table/구조물집계표.json") != system_version  # 변수 다름
    assert layers.read_manifest(root1 / "templates")["table/구조물집계표"]["판"] == system_version
    assert migrate.refresh_tables(root1) == []  # 다시 돌려도 그대로
    # 손댄 것 없는 작업본은 시스템 파일 그대로(판이 같음)
    migrate.refresh_tables(root2)
    assert layers.version_of(root2 / "templates/table/구조물집계표.json") == system_version