diff --git a/M02_MasterTemplete/M02_MasterTemplete_Router_Layers.py b/M02_MasterTemplete/M02_MasterTemplete_Router_Layers.py new file mode 100644 index 00000000..b6a714be --- /dev/null +++ b/M02_MasterTemplete/M02_MasterTemplete_Router_Layers.py @@ -0,0 +1,417 @@ +"""M02 양식 층 API — 계약 `tmp/M02_분석/6_계약.md` 「층 (sub4)」. + +⚠ 등록은 `main.py`(로그인만) · 층마다 권한은 여기서: + system 읽기만(고치기는 `/api/m02/templates` 시스템 관리자 길) + company 같은 회사 읽기 · 쓰기는 회사 관리자(ADMIN · 마스터 · 시스템 관리자) + personal 본인 읽기·쓰기 · 같은 회사 사람 것은 읽기만(가져오기) + project 같은 회사 프로젝트 · 작업본 쓰기 · `_initial/` 은 안 씀 +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any, Literal + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, ConfigDict, Field + +from common_util.common_util_auth import verify_session +from common_util.common_util_storage import resolve_stored_project_path +from config.config_db import get_db_pool +from M02_MasterTemplete import M02_Template_Layers as layers + +router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Layers"]) + +Layer = Literal["system", "company", "personal", "project"] + + +class SaveBody(BaseModel): + 판: str | None = None + 문서: dict[str, Any] + + +class TargetBody(BaseModel): + 종류: str | None = None + 이름: str | None = None + + +class ApplyBody(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + from_: Layer = Field(alias="from") + user_id: int | None = None + project_id: str | None = None + 종류: str | None = None + 이름: str | None = None + + +class SaveAsBody(BaseModel): + to: Literal["personal", "company"] + 종류: str + 이름: str + + +# ── DB (시험이 바꿔 끼움) ───────────────────────────── + + +async def _project_row(project_id: str) -> dict[str, Any] | None: + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + """SELECT id, name, company_id, user_id, storage_path FROM projects + WHERE id = %s AND deleted_at IS NULL""", + (str(project_id),), + ) + row = await cursor.fetchone() + if not row: + return None + keys = ("id", "name", "company_id", "user_id", "storage_path") + return dict(zip(keys, row, strict=True)) + + +async def _user_company(user_id: int) -> int | None: + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + "SELECT company_id FROM users WHERE id = %s AND deleted_at IS NULL", (user_id,) + ) + row = await cursor.fetchone() + return row[0] if row else None + + +async def _company_users(company_id: int) -> list[dict[str, Any]]: + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + "SELECT id, name FROM users WHERE company_id = %s AND deleted_at IS NULL ORDER BY id", + (company_id,), + ) + rows = await cursor.fetchall() + return [{"user_id": row[0], "name": row[1]} for row in rows] + + +async def _company_projects(company_id: int) -> list[dict[str, Any]]: + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + """SELECT id, name, storage_path FROM projects + WHERE company_id = %s AND deleted_at IS NULL ORDER BY created_at DESC""", + (company_id,), + ) + rows = await cursor.fetchall() + return [{"project_id": str(row[0]), "name": row[1], "storage_path": row[2]} for row in rows] + + +# ── 권한 · 자리 ─────────────────────────────────────── + + +def _is_system_admin(session: dict[str, Any]) -> bool: + return session.get("role") == "SYSTEM_ADMIN" + + +def _is_company_admin(session: dict[str, Any]) -> bool: + return session.get("role") in ("ADMIN", "SYSTEM_ADMIN") or bool(session.get("is_master")) + + +async def _project(session: dict[str, Any], project_id: str | None) -> dict[str, Any]: + """같은 회사 프로젝트 한 건 + `root`(실경로). 아니면 403/404.""" + if not project_id: + raise HTTPException(status_code=400, detail="project_id 가 필요합니다.") + row = await _project_row(project_id) + if row is None: + raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.") + if not _is_system_admin(session) and row["company_id"] != session.get("company_id"): + raise HTTPException(status_code=403, detail="다른 회사의 프로젝트입니다.") + if not row.get("storage_path"): + raise HTTPException(status_code=404, detail="프로젝트 저장 경로를 찾을 수 없습니다.") + row["root"] = Path(resolve_stored_project_path(row["storage_path"])) + return row + + +def _company_of(session: dict[str, Any], project: dict[str, Any] | None) -> int: + company_id = project["company_id"] if project else session.get("company_id") + if company_id is None: + raise HTTPException(status_code=403, detail="회사 연결이 필요합니다.") + return int(company_id) + + +async def _same_company_user(session: dict[str, Any], user_id: int, company_id: int) -> None: + if user_id == session.get("user_id"): + return + if await _user_company(user_id) != company_id: + raise HTTPException(status_code=403, detail="같은 회사 사람의 양식만 볼 수 있습니다.") + + +async def _layer_dir( + session: dict[str, Any], + layer: str, + *, + project_id: str | None, + user_id: int | None = None, + write: bool = False, +) -> Path: + """층 폴더 — 읽기·쓰기 권한까지 여기서 가름.""" + if layer == "system": + if write: + raise HTTPException(status_code=403, detail="시스템 양식은 마스터 템플릿 화면에서만.") + return layers.system_dir() + if layer == "project": + return layers.project_dir((await _project(session, project_id))["root"]) + project = await _project(session, project_id) if project_id else None + company_id = _company_of(session, project) + if layer == "company": + if write and not _is_company_admin(session): + raise HTTPException(status_code=403, detail="회사 공식 양식은 회사 관리자만.") + return layers.company_dir(company_id) + if layer == "personal": + owner = int(user_id) if user_id is not None else int(session["user_id"]) + if write and owner != session.get("user_id"): + raise HTTPException(status_code=403, detail="개인 양식은 본인만 고칩니다.") + await _same_company_user(session, owner, company_id) + return layers.personal_dir(company_id, owner) + raise HTTPException(status_code=404, detail="없는 층입니다.") + + +def _bad(error: ValueError) -> HTTPException: + return HTTPException(status_code=400, detail=str(error)) + + +# ── 층 길 ───────────────────────────────────────────── + + +@router.get("/layers/{layer}/templates") +async def list_layer( + layer: Layer, + project_id: str | None = Query(None), + user_id: int | None = Query(None), + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + folder = await _layer_dir(session, layer, project_id=project_id, user_id=user_id) + rows = await asyncio.to_thread(layers.list_templates, folder) + manifest = await asyncio.to_thread(layers.read_manifest, folder) + for row in rows: + row["출처"] = manifest.get(f"{row['종류']}/{row['이름']}") + return {"층": layer, "양식": rows} + + +@router.get("/layers/{layer}/templates/{kind}/{name}") +async def read_layer_template( + layer: Layer, + kind: str, + name: str, + project_id: str | None = Query(None), + user_id: int | None = Query(None), + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + folder = await _layer_dir(session, layer, project_id=project_id, user_id=user_id) + try: + found = await asyncio.to_thread(layers.read_template, folder, kind, name) + except ValueError as error: + raise _bad(error) from error + if found is None and layer == "project": + # 옛 프로젝트(사본 없음) — 시스템 양식으로 떨어짐 · 저장하면 그때 작업본이 생김 + found = await asyncio.to_thread(layers.read_template, layers.system_dir(), kind, name) + if found is not None: + found.update({"층": "system", "판": None}) + return found + if found is None: + raise HTTPException(status_code=404, detail="양식을 찾을 수 없습니다.") + found["층"] = layer + found["출처"] = layers.read_manifest(folder).get(f"{kind}/{name}") + return found + + +@router.put("/layers/{layer}/templates/{kind}/{name}") +async def save_layer_template( + layer: Layer, + kind: str, + name: str, + body: SaveBody, + project_id: str | None = Query(None), + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + folder = await _layer_dir(session, layer, project_id=project_id, write=True) + try: + version = await asyncio.to_thread( + layers.write_template, + folder, + kind, + name, + body.문서, + version=body.판, + check_version=True, + ) + except layers.StaleTemplate as error: + raise HTTPException( + status_code=409, detail={"message": str(error), "판": error.current} + ) from error + except ValueError as error: + raise _bad(error) from error + return {"종류": kind, "이름": name, "판": version, "층": layer} + + +# ── 프로젝트 길 ─────────────────────────────────────── + + +@router.post("/projects/{project_id}/templates/reset") +async def reset_project_templates( + project_id: str, + body: TargetBody | None = None, + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + project = await _project(session, project_id) + target = body or TargetBody() + try: + done = await asyncio.to_thread( + layers.reset_project, project["root"], kind=target.종류, name=target.이름 + ) + except ValueError as error: + raise _bad(error) from error + return {"초기화": done} + + +@router.post("/projects/{project_id}/templates/apply") +async def apply_project_templates( + project_id: str, + body: ApplyBody, + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + """다른 층 양식을 작업본에 덮어씀 — [회사 양식 적용] · [양식 가져오기]. + + `_initial/` 은 그대로 — 초기화 기준은 안 바뀜. + """ + project = await _project(session, project_id) + ref: dict[str, Any] = {} + if body.from_ == "project": + if not body.project_id or body.project_id == project_id: + raise HTTPException(status_code=400, detail="가져올 다른 프로젝트를 고르세요.") + other = await _project(session, body.project_id) + if other["company_id"] != project["company_id"]: + raise HTTPException(status_code=403, detail="같은 회사 프로젝트만 가져옵니다.") + source = layers.project_dir(other["root"]) + ref = {"project_id": body.project_id, "프로젝트": other.get("name")} + elif body.from_ == "personal": + owner = body.user_id if body.user_id is not None else int(session["user_id"]) + await _same_company_user(session, owner, int(project["company_id"])) + source = layers.personal_dir(project["company_id"], owner) + ref = {"user_id": owner} + elif body.from_ == "company": + source = layers.company_dir(project["company_id"]) + else: + source = layers.system_dir() + try: + done = await asyncio.to_thread( + layers.copy_templates, + source, + layers.project_dir(project["root"]), + source_layer=body.from_, + kind=body.종류, + name=body.이름, + source_ref=ref, + ) + except ValueError as error: + raise _bad(error) from error + if not done: + raise HTTPException(status_code=404, detail="가져올 양식이 없습니다.") + return {"적용": done, "from": body.from_} + + +@router.post("/projects/{project_id}/templates/save-as") +async def save_project_template_as( + project_id: str, + body: SaveAsBody, + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + """작업본 한 벌을 [내 양식으로 저장] · [회사 공식으로 저장].""" + project = await _project(session, project_id) + if body.to == "company" and not _is_company_admin(session): + raise HTTPException(status_code=403, detail="회사 공식 양식은 회사 관리자만.") + if body.to == "company": + target = layers.company_dir(project["company_id"]) + else: + target = layers.personal_dir(project["company_id"], session["user_id"]) + try: + done = await asyncio.to_thread( + layers.copy_templates, + layers.project_dir(project["root"]), + target, + source_layer="project", + kind=body.종류, + name=body.이름, + source_ref={"project_id": project_id, "프로젝트": project.get("name")}, + ) + except ValueError as error: + raise _bad(error) from error + if not done: + raise HTTPException(status_code=404, detail="프로젝트 작업본에 그 양식이 없습니다.") + return {"저장": done, "to": body.to} + + +@router.get("/projects/{project_id}/sources") +async def list_sources( + project_id: str, + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + """가져올 수 있는 것 — 시스템 · 회사 공식 · 같은 회사 사람 개인 · 같은 회사 다른 프로젝트.""" + project = await _project(session, project_id) + company_id = int(project["company_id"]) + + def _people(users: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows = [] + for user in users: + found = layers.list_templates(layers.personal_dir(company_id, user["user_id"])) + if found: + rows.append({**user, "양식": found}) + return rows + + def _projects(projects: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows = [] + for other in projects: + if other["project_id"] == str(project_id) or not other.get("storage_path"): + continue + try: + root = Path(resolve_stored_project_path(other["storage_path"])) + except ValueError: + continue + found = layers.list_templates(layers.project_dir(root)) + if found: + rows.append( + {"project_id": other["project_id"], "name": other["name"], "양식": found} + ) + return rows + + users = await _company_users(company_id) + projects = await _company_projects(company_id) + return { + "system": await asyncio.to_thread(layers.list_templates, layers.system_dir()), + "company": await asyncio.to_thread(layers.list_templates, layers.company_dir(company_id)), + "personal": await asyncio.to_thread(_people, users), + "project": await asyncio.to_thread(_projects, projects), + } + + +@router.get("/projects/{project_id}/tables/{name}/filled") +async def filled_table( + project_id: str, + name: str, + session: dict[str, Any] = Depends(verify_session), +) -> dict[str, Any]: + """설계값을 채운 표 문서 — 저장 안 함(5장 ③ 서버 단독).""" + from M02_MasterTemplete import M02_Table_Fill as fill + + project = await _project(session, project_id) + try: + found = await asyncio.to_thread( + layers.read_template, layers.project_dir(project["root"]), "table", name + ) + if found is None: + found = await asyncio.to_thread( + layers.read_template, layers.system_dir(), "table", name + ) + except ValueError as error: + raise _bad(error) from error + if found is None: + raise HTTPException(status_code=404, detail="양식을 찾을 수 없습니다.") + document = await asyncio.to_thread(fill.fill_table, project["root"], found["문서"]) + return {"이름": name, "판": found["판"], "문서": document} diff --git a/resources/tester/test_m02_template_layers.py b/resources/tester/test_m02_template_layers.py new file mode 100644 index 00000000..ab784375 --- /dev/null +++ b/resources/tester/test_m02_template_layers.py @@ -0,0 +1,183 @@ +"""M02 양식 층 — 자리 · 복사 · 초기화 · 회사 적용 · 가져오기 · 권한 (임시 storage).""" + +from __future__ import annotations + +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)