Merge remote-tracking branch 'origin/dev' into sub_laptop_3
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
"""M02 마스터 템플릿 API — 시스템 층(계약 `6_계약.md` 서버 길 첫 묶음).
|
||||
|
||||
⚠ 권한은 등록하는 쪽(`main.py`)이 `system_admin_only` 로 붙임.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from M02_MasterTemplete import M02_MasterTemplete_Store as store
|
||||
|
||||
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete"])
|
||||
|
||||
|
||||
class SaveBody(BaseModel):
|
||||
판: str = ""
|
||||
문서: Any
|
||||
|
||||
|
||||
def _call(fn, *args):
|
||||
try:
|
||||
return fn(*args)
|
||||
except store.StoreError as e:
|
||||
raise HTTPException(status_code=e.status, detail=e.detail) from e
|
||||
|
||||
|
||||
@router.get("/templates")
|
||||
def get_templates() -> list[dict]:
|
||||
return _call(store.list_all)
|
||||
|
||||
|
||||
@router.get("/templates/{kind}/{name}")
|
||||
def get_template(kind: str, name: str) -> dict:
|
||||
return _call(store.read, kind, name)
|
||||
|
||||
|
||||
@router.put("/templates/{kind}/{name}")
|
||||
def put_template(kind: str, name: str, body: SaveBody) -> dict:
|
||||
return _call(store.write, kind, name, body.판, body.문서)
|
||||
|
||||
|
||||
@router.delete("/templates/{kind}/{name}")
|
||||
def delete_template(kind: str, name: str, 판: str | None = None) -> dict:
|
||||
_call(store.delete, kind, name, 판)
|
||||
return {"ok": True}
|
||||
@@ -0,0 +1,10 @@
|
||||
"""M02 마스터 템플릿 — 도면 길 틀(빈 라우터).
|
||||
|
||||
⚠ 로그인만으로 `main.py` 에 붙음 — 권한은 길 안에서 봄. 길은 계약 `6_계약.md` 를 따름.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Drawing"])
|
||||
@@ -0,0 +1,10 @@
|
||||
"""M02 마스터 템플릿 — 층 길 틀(빈 라우터).
|
||||
|
||||
⚠ 로그인만으로 `main.py` 에 붙음 — 권한은 길 안에서 봄. 길은 계약 `6_계약.md` 를 따름.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Layers"])
|
||||
@@ -0,0 +1,86 @@
|
||||
"""M02 마스터 템플릿 — 시스템 층 저장소 (`resources/master_template/{table,drawing}/<이름>.json`).
|
||||
|
||||
M01 `Store` 방식 — 판(파일 sha256 앞 16자) · 판이 다르면 409 · 원자 쓰기.
|
||||
권한은 등록하는 쪽(`main.py`)이 붙임.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from common_util.common_util_json import atomic_write_json
|
||||
|
||||
FOLDER: Path = Path(__file__).resolve().parent.parent / "resources" / "master_template"
|
||||
KINDS = ("table", "drawing") # 시험은 FOLDER 를 사본으로 바꿈
|
||||
_BAD_NAME = re.compile(r'[\/:*?"<>|\x00-\x1f]')
|
||||
# ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로
|
||||
_LOCK = threading.Lock()
|
||||
|
||||
|
||||
class StoreError(Exception):
|
||||
def __init__(self, status: int, detail):
|
||||
super().__init__(detail)
|
||||
self.status, self.detail = status, detail
|
||||
|
||||
|
||||
def version_of(raw: bytes) -> str:
|
||||
return hashlib.sha256(raw).hexdigest()[:16]
|
||||
|
||||
|
||||
def _path(kind: str, name: str) -> Path:
|
||||
if kind not in KINDS:
|
||||
raise StoreError(404, f"없는 종류 「{kind}」")
|
||||
if not name or name != name.strip() or name.startswith(".") or _BAD_NAME.search(name):
|
||||
raise StoreError(422, f"쓸 수 없는 이름 「{name}」")
|
||||
return FOLDER / kind / f"{name}.json"
|
||||
|
||||
|
||||
def _info(kind: str, path: Path) -> dict[str, Any]:
|
||||
raw = path.read_bytes()
|
||||
stamp = datetime.fromtimestamp(path.stat().st_mtime).isoformat(timespec="seconds")
|
||||
return {"종류": kind, "이름": path.stem, "판": version_of(raw), "수정일": stamp}
|
||||
|
||||
|
||||
def list_all() -> list[dict[str, Any]]:
|
||||
rows = [
|
||||
_info(kind, p)
|
||||
for kind in KINDS
|
||||
for p in sorted((FOLDER / kind).glob("*.json"))
|
||||
if not p.name.startswith(".")
|
||||
]
|
||||
return rows
|
||||
|
||||
|
||||
def read(kind: str, name: str) -> dict[str, Any]:
|
||||
path = _path(kind, name)
|
||||
if not path.is_file():
|
||||
raise StoreError(404, f"없는 양식 「{name}」")
|
||||
raw = path.read_bytes()
|
||||
return {"종류": kind, "이름": name, "판": version_of(raw), "문서": json.loads(raw)}
|
||||
|
||||
|
||||
def write(kind: str, name: str, version: str, doc: Any) -> dict[str, Any]:
|
||||
"""`version` 이 빈 글이면 새로 만듦(이미 있으면 409) · 아니면 그 판일 때만 덮어씀."""
|
||||
path = _path(kind, name)
|
||||
with _LOCK:
|
||||
have = version_of(path.read_bytes()) if path.is_file() else ""
|
||||
if have != (version or ""):
|
||||
raise StoreError(409, {"stale": [name], "판": have})
|
||||
atomic_write_json(path, doc)
|
||||
return _info(kind, path)
|
||||
|
||||
|
||||
def delete(kind: str, name: str, version: str | None = None) -> None:
|
||||
path = _path(kind, name)
|
||||
with _LOCK:
|
||||
if not path.is_file():
|
||||
raise StoreError(404, f"없는 양식 「{name}」")
|
||||
if version is not None and version_of(path.read_bytes()) != version:
|
||||
raise StoreError(409, {"stale": [name], "판": version_of(path.read_bytes())})
|
||||
path.unlink()
|
||||
@@ -61,6 +61,9 @@ from B06_Section.B06_Section_Router_HaulPlan import (
|
||||
from B07_DesignDetail.B07_DesignDetail_Router import router as b07_design_router
|
||||
from B07_DesignDetail.B07_DesignDetail_Router_Frame import router as b07_frame_router
|
||||
from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
|
||||
from M02_MasterTemplete.M02_MasterTemplete_Router import router as m02_master_template_router
|
||||
from M02_MasterTemplete.M02_MasterTemplete_Router_Drawing import router as m02_drawing_router
|
||||
from M02_MasterTemplete.M02_MasterTemplete_Router_Layers import router as m02_layers_router
|
||||
from common_util.common_util_audit import note_api_call, record_call_burst
|
||||
from common_util.common_util_auth import (
|
||||
require_company,
|
||||
@@ -632,6 +635,10 @@ app.include_router(b07_frame_router, dependencies=protected_with_company)
|
||||
# 마스터 데이터 — 회사·프로젝트가 아니라 시스템 관리자만 본다.
|
||||
system_admin_only = [Depends(verify_session), Depends(require_system_admin)]
|
||||
app.include_router(m01_master_data_router, dependencies=system_admin_only)
|
||||
# 마스터 템플릿 — 시스템 층은 시스템 관리자만 · 층(회사·개인·프로젝트)·도면 길은 로그인만(권한은 길 안에서).
|
||||
app.include_router(m02_master_template_router, dependencies=system_admin_only)
|
||||
app.include_router(m02_layers_router, dependencies=protected)
|
||||
app.include_router(m02_drawing_router, dependencies=protected)
|
||||
# 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근).
|
||||
# 그 위에 서버가 환경까지 한 번 더 본다.
|
||||
app.include_router(dev_unlock_router, dependencies=protected_with_company)
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
"""M02 시스템 층 저장소·길 — 새로·저장·다시 열기·지우기·409 (사본 폴더로)."""
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from M02_MasterTemplete import M02_MasterTemplete_Store as store
|
||||
from M02_MasterTemplete.M02_MasterTemplete_Router import router
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client(tmp_path, monkeypatch):
|
||||
for kind in store.KINDS:
|
||||
(tmp_path / kind).mkdir()
|
||||
monkeypatch.setattr(store, "FOLDER", tmp_path)
|
||||
app = FastAPI()
|
||||
app.include_router(router)
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_new_save_reopen_delete_and_stale(client):
|
||||
assert client.get("/api/m02/templates").json() == []
|
||||
|
||||
made = client.put("/api/m02/templates/table/집계표", json={"판": "", "문서": {"열": [1]}})
|
||||
assert made.status_code == 200
|
||||
v1 = made.json()["판"]
|
||||
assert (
|
||||
client.put("/api/m02/templates/table/집계표", json={"판": "", "문서": {}}).status_code
|
||||
== 409
|
||||
)
|
||||
|
||||
got = client.get("/api/m02/templates/table/집계표").json()
|
||||
assert got["판"] == v1 and got["문서"] == {"열": [1]}
|
||||
|
||||
saved = client.put("/api/m02/templates/table/집계표", json={"판": v1, "문서": {"열": [1, 2]}})
|
||||
assert saved.status_code == 200 and saved.json()["판"] != v1
|
||||
stale = client.put("/api/m02/templates/table/집계표", json={"판": v1, "문서": {}})
|
||||
assert stale.status_code == 409 and stale.json()["detail"]["stale"] == ["집계표"]
|
||||
|
||||
rows = client.get("/api/m02/templates").json()
|
||||
assert [(r["종류"], r["이름"]) for r in rows] == [("table", "집계표")]
|
||||
|
||||
assert client.delete("/api/m02/templates/table/집계표").status_code == 200
|
||||
assert client.get("/api/m02/templates/table/집계표").status_code == 404
|
||||
|
||||
|
||||
def test_bad_names_and_kinds(client):
|
||||
assert (
|
||||
client.put("/api/m02/templates/table/a..b", json={"판": "", "문서": {}}).status_code == 200
|
||||
)
|
||||
assert (
|
||||
client.put("/api/m02/templates/table/.숨김", json={"판": "", "문서": {}}).status_code == 422
|
||||
)
|
||||
assert client.put("/api/m02/templates/other/x", json={"판": "", "문서": {}}).status_code == 404
|
||||
Reference in New Issue
Block a user