Merge remote-tracking branch 'origin/sub_laptop_4' into sub_laptop_2
This commit is contained in:
@@ -217,7 +217,37 @@ def check_saved(files: dict, changed: list[str], book: dict, narrowed=None) -> l
|
||||
out += cm.check_logics(files, whole, book, narrowed and narrowed[:2])
|
||||
if narrowed and narrowed[2]: # 요소·표를 고쳤으면 연결 · 준용도 봄(로직 좁힘과 따로)
|
||||
out += cm.check_links(whole)
|
||||
return out + cm.mcb.check_all(files, whole)
|
||||
return out + cm.mcb.check_all(files, whole) + labor_errors(whole)
|
||||
|
||||
|
||||
def _labor_files() -> list[tuple[str, Path]]:
|
||||
"""모든 층(마스터 · 회사 · 개인)의 일위대가 문서 `labor/<열>.json` — (어디, 파일)."""
|
||||
from M02_MasterTemplete import M02_Template_Layers as layers # M02 가 M01 을 부름 — 늦게
|
||||
|
||||
root = layers.storage_root()
|
||||
found = [("마스터", p) for p in sorted((layers.system_dir() / "labor").glob("*.json"))]
|
||||
found += [(f"회사 {p.parts[-4]}", p) for p in sorted(root.glob("*/templates/labor/*.json"))]
|
||||
personal = sorted(root.glob("*/*/templates/labor/*.json"))
|
||||
return found + [(f"개인 {p.parts[-5]}/{p.parts[-4]}", p) for p in personal]
|
||||
|
||||
|
||||
def labor_errors(whole) -> list[str]:
|
||||
"""일위대가 문서가 담은 로직이 마스터에 없음(39-1d) — 로직 지우기를 막음 · 어디서 쓰는지 알림.
|
||||
전부터 깨진 것은 `save` 가 고치기 전 검사와 견주어 뺌."""
|
||||
out = []
|
||||
for where, path in _labor_files():
|
||||
try:
|
||||
doc = json.loads(path.read_bytes())
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
name = (doc.get("일위대가번") if isinstance(doc, dict) else None) or path.stem
|
||||
for one in (doc.get("담은로직") if isinstance(doc, dict) else None) or []:
|
||||
key = str(one.get("로직") or "") if isinstance(one, dict) else ""
|
||||
try:
|
||||
whole.logic(key)
|
||||
except cm.mf.FormulaError:
|
||||
out.append(f"일위대가 {where} 「{name}」 이 로직 「{key}」 을 씀 — 거기서 먼저 뺌")
|
||||
return out
|
||||
|
||||
|
||||
def warm(folder: Path) -> threading.Thread:
|
||||
|
||||
@@ -138,18 +138,6 @@ def post_combo(
|
||||
return _call(folder, combos.create, body.combo, body.판 or None)
|
||||
|
||||
|
||||
@router.put("/layers/{layer}/combos/{key}")
|
||||
def put_combo(
|
||||
layer: Layer,
|
||||
key: str,
|
||||
body: ComboBody,
|
||||
owner: str | None = None,
|
||||
session: Session = Depends(verify_session),
|
||||
):
|
||||
folder = _root(session, layer, owner, write=True)
|
||||
return _call(folder, combos.edit, key, body.combo, body.판 or "")
|
||||
|
||||
|
||||
@router.delete("/layers/{layer}/combos/{key}")
|
||||
def delete_combo(
|
||||
layer: Layer,
|
||||
|
||||
@@ -11,11 +11,14 @@ PLAN 27-1c) · `owner` = 회사 `{C}` · 개인 `{C}/{U}`(없으면 자기 층)
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Callable, Coroutine
|
||||
from contextvars import ContextVar
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
import anyio.from_thread
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
|
||||
from fastapi.routing import APIRoute
|
||||
from pydantic import BaseModel
|
||||
|
||||
from common_util.common_util_audit import record_audit
|
||||
@@ -37,7 +40,32 @@ from M02_MasterTemplete.M02_MasterTemplete_Router import (
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"])
|
||||
#: 이 요청이 쓴 남의 층(`others_write`) — 길이 성공해 답한 뒤에만 시스템 로그(`NotedRoute`)
|
||||
_WRITES: ContextVar[list[tuple[dict, access.Grant]] | None] = ContextVar("m02_writes", default=None)
|
||||
|
||||
|
||||
class NotedRoute(APIRoute):
|
||||
"""남의 층 쓰기 기록 = 성공한 쓰기만(39-1d) — 막히거나 실패한 부름(예외 · 4xx · 5xx)은 뺌."""
|
||||
|
||||
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
|
||||
handler = super().get_route_handler()
|
||||
|
||||
async def run(request: Request) -> Response:
|
||||
notes: list[tuple[dict, access.Grant]] = []
|
||||
token = _WRITES.set(notes)
|
||||
try:
|
||||
response = await handler(request)
|
||||
finally:
|
||||
_WRITES.reset(token)
|
||||
if response.status_code < 400:
|
||||
for session, got in notes:
|
||||
await _note(session, got)
|
||||
return response
|
||||
|
||||
return run
|
||||
|
||||
|
||||
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"], route_class=NotedRoute)
|
||||
|
||||
Layer = Literal["system", "company", "personal"]
|
||||
Session = dict[str, Any]
|
||||
@@ -114,12 +142,17 @@ async def _note(session: Session, got: access.Grant) -> None:
|
||||
|
||||
|
||||
async def others_write(session: Session, got: access.Grant, folder: Path) -> None:
|
||||
"""남의 층 쓰기 — 처음(집계표 없음)이면 주인이 있는지 본 뒤 뼈대 · 쓸 때마다 시스템 로그."""
|
||||
"""남의 층 쓰기 — 처음(집계표 없음)이면 주인이 있는지 본 뒤 뼈대 · 시스템 로그는 길이 성공한
|
||||
뒤(`NotedRoute` · 그 밖의 길은 바로)."""
|
||||
if not (folder / "table" / f"{store.STRUCTURE_TABLE}.json").is_file():
|
||||
if not await _member(str(got.owner)):
|
||||
raise HTTPException(status_code=403, detail="그 회사 · 그 회사 사람이 아님")
|
||||
items_store.ensure(folder)
|
||||
await _note(session, got)
|
||||
pending = _WRITES.get()
|
||||
if pending is None:
|
||||
await _note(session, got)
|
||||
else:
|
||||
pending.append((session, got))
|
||||
|
||||
|
||||
def _root(session: Session, layer: str, owner: str | None = None, write: bool = False) -> Path:
|
||||
|
||||
@@ -34,7 +34,9 @@ from M02_MasterTemplete import M02_Template_Layers as layers
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Layers"])
|
||||
router = APIRouter(
|
||||
prefix="/api/m02", tags=["M02 MasterTemplete Layers"], route_class=items_router.NotedRoute
|
||||
)
|
||||
|
||||
Layer = Literal["system", "company", "personal", "project"]
|
||||
|
||||
|
||||
@@ -176,3 +176,46 @@ def test_만든_자체_로직이_시험_계산에_돈다(client: TestClient) ->
|
||||
assert got["ok"] is True and Decimal(str(got["sums"]["노무비"])) > 0
|
||||
listed = _get(client, "/api/m01/logics", owner="공용")["logics"]
|
||||
assert made["key"] in [x["키"] for x in listed]
|
||||
|
||||
|
||||
def test_일위대가가_담은_로직은_못_지운다(client: TestClient, tmp_path: Path, monkeypatch) -> None:
|
||||
"""39-1d — 로직 지우기 검사 = 모든 층(마스터 · 회사 · 개인) labor 담은로직 + 실무문서(마스터
|
||||
일위대가 조합) · 쓰는 곳을 알림 · 다 빼면 지워짐(임시 storage · 임시 마스터)."""
|
||||
import json
|
||||
|
||||
from config import config_system
|
||||
from M02_MasterTemplete import M02_Template_Layers as layers
|
||||
|
||||
system, storage = tmp_path / "system", tmp_path / "storage"
|
||||
monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
|
||||
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
|
||||
made = _post(client, "/api/m01/logic/new", {"logic": _draft(), "owner": "현장"})
|
||||
key = made["key"]
|
||||
docs = {
|
||||
"회사 3": storage / "3/templates/labor/item1.json",
|
||||
"개인 3/15": storage / "3/15/templates/labor/item2.json",
|
||||
"마스터": system / "labor/item3.json",
|
||||
}
|
||||
for at, path in enumerate(docs.values()):
|
||||
path.parent.mkdir(parents=True)
|
||||
doc = {"일위대가번": f"SU-00000{at}", "담은로직": [{"로직": SAMPLE}, {"로직": key}]}
|
||||
path.write_text(json.dumps(doc, ensure_ascii=False), encoding="utf-8")
|
||||
combos = store.FOLDER / store.mcb.FILE
|
||||
raw = json.loads(combos.read_text(encoding="utf-8"))
|
||||
raw["줄"].append({**raw["줄"][0], "키": "UA999999", "담은로직": [{"로직": key, "메모": None}]})
|
||||
combos.write_text(json.dumps(raw, ensure_ascii=False), encoding="utf-8")
|
||||
|
||||
def delete(want: int) -> dict:
|
||||
version = _get(client, "/api/m01/logic", key=key)["version"]
|
||||
return _post(client, "/api/m01/logic/delete", {"key": key, "version": version}, want)
|
||||
|
||||
errors = " ".join(delete(422)["detail"]["errors"])
|
||||
for where in docs:
|
||||
assert f"일위대가 {where} 「SU-" in errors and key in errors # 어디서 쓰는지 알림
|
||||
assert store.mcb.FILE in errors # 실무문서(마스터 일위대가 조합)
|
||||
for path in docs.values(): # 일위대가에서 뺌 — 실무문서만 남음
|
||||
path.write_text(json.dumps({"담은로직": [{"로직": SAMPLE}]}), encoding="utf-8")
|
||||
assert "일위대가 " not in " ".join(delete(422)["detail"]["errors"])
|
||||
raw["줄"].pop()
|
||||
combos.write_text(json.dumps(raw, ensure_ascii=False), encoding="utf-8")
|
||||
assert delete(200)["file"] == make.OWN # 쓰는 곳 0 = 지워짐
|
||||
|
||||
@@ -60,7 +60,7 @@ def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
||||
|
||||
|
||||
def test_조합_길은_시스템_관리자만(world):
|
||||
"""39-6 — 일위대가 조합 = 실무문서 · 조합 길 전부 시스템 관리자만 · 담을 로직 읽기는 로그인만."""
|
||||
"""39-6 — 일위대가 조합 = 실무문서 · 조합 길 전부 시스템 관리자만 · 담을 로직은 로그인만."""
|
||||
world["client"].app.dependency_overrides.pop(combos_router._door)
|
||||
for role in ("USER", "ADMIN"):
|
||||
client = _as(world, role)
|
||||
@@ -149,10 +149,9 @@ def test_검사_빈_조합_이름_겹침_본뜬키_판(world):
|
||||
branched = _new(me, "personal", _combo("마스터 본뜸", 본뜬키=master_key))
|
||||
assert branched["combo"]["본뜬키"] == master_key # 층을 넘는 계보
|
||||
url = f"{URL.format('personal')}/{first['키']}"
|
||||
assert me.put(url, json={"combo": _combo("원본 고침"), "판": "낡은판"}).status_code == 409
|
||||
now = _list(me, "personal")["판"]
|
||||
res = me.put(url, json={"combo": _combo("원본 고침", (B,)), "판": now})
|
||||
assert res.status_code == 200 and res.json()["combo"]["담은로직"][0]["로직"] == B
|
||||
assert (
|
||||
me.put(url, json={"combo": _combo("원본 고침")}).status_code == 405
|
||||
) # 조합 고치기 은퇴(39-1d)
|
||||
|
||||
|
||||
# ── 나무 ─────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -160,7 +160,7 @@ def test_남의_층_첫_쓰기는_소속_확인_뒤_뼈대(world: dict[str, Any]
|
||||
add["판"] = ""
|
||||
assert client.post("/api/m02/layers/company/tree?owner=9", json=add).status_code == 409
|
||||
assert (storage / "9/templates/table/구조물집계표.json").is_file()
|
||||
assert [n[2] for n in world["notes"]] == ["7/44", "7/44", "9"]
|
||||
assert [n[2] for n in world["notes"]] == ["7/44"] # 성공한 쓰기만(39-1d) · 409 · 403 은 안 남김
|
||||
|
||||
|
||||
def test_남의_층_목록(world: dict[str, Any]):
|
||||
|
||||
Reference in New Issue
Block a user