merge(M01): origin/sub_laptop_7 받음 — 로직 지우기 검사 두 벌을 sub4 한 벌로(labor_errors · 실무문서 · 알림 글)
- sub7 의 LABOR_ROOTS · 겹친 _labor_files · labor_errors 지움 · sub7 시험은 층 자리(layers.SYSTEM_ROOT) · 알림 글만 맞춤 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BVPp2o6Mh2FZJr72yW3ofZ
This commit is contained in:
@@ -141,8 +141,9 @@ def copy_item(
|
||||
origin: dict[str, Any],
|
||||
path: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""다른 층 항목(SS `key`) → `target` 층 새 열 하나 — 머리 · 단위 · 일위대가 · 바인딩 그대로 ·
|
||||
도면 · 면적산출 · 재료산출 내용(알약 연결 포함) · 항목 연결 · 자재 조합 줄(새 UM) 복사 ·
|
||||
"""다른 층 항목(SS `key`) → `target` 층 새 열 하나 — 머리 · 단위 · 바인딩 그대로 ·
|
||||
도면 · 일위대가(새 SU · 조합 키는 안 받음 · 39-1b) · 면적산출 · 재료산출 내용(알약 연결 포함) ·
|
||||
항목 연결 · 자재 조합 줄(새 UM) 복사 ·
|
||||
새 번호 · 새 SS 키 ·
|
||||
열 `본뜬` = `origin`(원래 층 · 주인 · 키) · `path` = 새 자리(폴더 + 이름 · 없으면 원래 경로 ·
|
||||
붙여넣기 PLAN 30-2). 돌려줌 `{key, 열, 이름}`."""
|
||||
@@ -179,7 +180,13 @@ def copy_item(
|
||||
table = store.read("table", store.STRUCTURE_TABLE)
|
||||
doc = table["문서"]
|
||||
new_id = _fresh_id(col_id, {c.get("id") for c in doc.get("열", []) if isinstance(c, dict)})
|
||||
made = {**json.loads(json.dumps(col)), "id": new_id, "자재조합": new_um, "본뜬": origin}
|
||||
made = {
|
||||
**json.loads(json.dumps(col)),
|
||||
"id": new_id,
|
||||
"일위대가": None, # 정본 = 일위대가 문서(아래 복사)
|
||||
"자재조합": new_um,
|
||||
"본뜬": origin,
|
||||
}
|
||||
if path:
|
||||
made.update({"경로": list(path), "머리": list(path)}) # 머리 줄 수는 저장 맞춤이 채움
|
||||
doc["열"] = [*doc.get("열", []), made]
|
||||
@@ -192,6 +199,7 @@ def copy_item(
|
||||
body = store._server(kind, body)
|
||||
with store._LOCK:
|
||||
atomic_write_json(store._path(kind, new_id), body)
|
||||
store._refresh_labor(new_id) # 면적이 일위대가 뒤에 서므로 칸 값으로 다시
|
||||
ss = next(r for r in store.read_combos()["줄"] if r.get("열") == new_id)
|
||||
inputs_store.copy_between(source, target, col_id, new_id) # 항목 연결도 같이(판 1)
|
||||
return {"key": ss["키"], "열": new_id, "이름": ss["이름"]}
|
||||
|
||||
@@ -400,18 +400,36 @@ def post_use_combo(
|
||||
return _call(folder, branch.use_combo, column, body.종류, body.키, body.판)
|
||||
|
||||
|
||||
def _copy_practice(column: str, key: str) -> dict[str, Any]:
|
||||
"""실무문서 조합(마스터 UA) → 이 열 일위대가 문서(입력값 빈 값 · 39-1b)."""
|
||||
from M02_MasterTemplete import M02_Labor_Doc as labor
|
||||
|
||||
if not key.startswith("UA"):
|
||||
raise store.StoreError(400, f"실무문서 조합 키가 아님 「{key}」")
|
||||
return store.copy_from("labor", column, key, body=labor.from_combo(key))
|
||||
|
||||
|
||||
@router.post("/layers/{layer}/{kind}/{column}/copy")
|
||||
def post_copy(
|
||||
layer: Layer,
|
||||
kind: Literal["structures", "area", "material"],
|
||||
kind: Literal["structures", "labor", "area", "material"],
|
||||
column: str,
|
||||
source: str = Query(alias="from"),
|
||||
owner: str | None = None,
|
||||
from_layer: Literal["system", "company", "personal", "practice"] | None = None,
|
||||
from_owner: str | None = None,
|
||||
session: Session = Depends(verify_session),
|
||||
):
|
||||
"""항목 하나 복사해 가져오기(덮어씀 · 번호 · 열 그대로) — 원본 = 같은 층(`from_layer` 없음) ·
|
||||
다른 층(원본 읽기 권한) · 실무문서 조합(`practice` · 일위대가만 · 시스템 관리자만)."""
|
||||
folder = _root(session, layer, owner, write=True)
|
||||
part = "structure" if kind == "structures" else kind
|
||||
return _call(folder, store.copy_from, part, column, source)
|
||||
if from_layer == "practice":
|
||||
if part != "labor" or not access.is_system_admin(session):
|
||||
raise HTTPException(status_code=403, detail="실무문서는 시스템 관리자가 일위대가로만")
|
||||
return _call(folder, _copy_practice, column, source)
|
||||
origin = _root(session, from_layer, from_owner) if from_layer else None
|
||||
return _call(folder, store.copy_from, part, column, source, origin)
|
||||
|
||||
|
||||
@router.get("/layers/{layer}/material-combos")
|
||||
|
||||
@@ -186,6 +186,7 @@ def write(kind: str, name: str, version: str, doc: Any, screen: bool = False) ->
|
||||
for column, source in origins.items(): # 빈 파일 채우기 앞 — 원본 내용으로 만듦
|
||||
for part in _KEY:
|
||||
_ensure(part, column, source)
|
||||
_refresh_labor(column) # 면적이 일위대가 뒤에 서므로 칸 값으로 다시
|
||||
_copy_inputs(source, column)
|
||||
backfill_missing_structures() # 지우기 앞 — 같은 저장의 새 열이 지운 번호를 안 받게
|
||||
prune_removed_structures()
|
||||
@@ -335,7 +336,8 @@ def _ensure(kind: str, column: str, source: str | None = None) -> bool:
|
||||
|
||||
def _take_origins(doc: Any) -> dict[str, str]:
|
||||
"""표 열의 `본뜬열` 을 떼어 냄 → `{새 열: 원본 열}` (구조물 파일이 아직 없는 열만) ·
|
||||
그 열에 일위대가 · 자재조합 키가 비었으면 원본 열 것을 받음."""
|
||||
그 열에 자재조합 키가 비었으면 원본 열 것을 받음 · 일위대가는 원본 열 문서를 새 번호로 복사
|
||||
(`_ensure` · 39-1b — 조합 키는 안 받음)."""
|
||||
cols = (
|
||||
[c for c in (doc.get("열") or []) if isinstance(c, dict)] if isinstance(doc, dict) else []
|
||||
)
|
||||
@@ -348,8 +350,7 @@ def _take_origins(doc: Any) -> dict[str, str]:
|
||||
if (root() / "structure" / f"{col['id']}.json").is_file():
|
||||
continue
|
||||
out[col["id"]] = source
|
||||
for field in ("일위대가", "자재조합"):
|
||||
col[field] = col.get(field) or by_id[source].get(field)
|
||||
col["자재조합"] = col.get("자재조합") or by_id[source].get("자재조합")
|
||||
return out
|
||||
|
||||
|
||||
@@ -365,17 +366,32 @@ def create_structure(column: str) -> dict[str, Any]:
|
||||
return read("structure", column)
|
||||
|
||||
|
||||
def copy_from(kind: str, column: str, source: str) -> dict[str, Any]:
|
||||
def copy_from(
|
||||
kind: str,
|
||||
column: str,
|
||||
source: str,
|
||||
source_root: Path | None = None,
|
||||
body: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""다른 열(`source`) 내용을 받는 열(`column`)로 가져옴 — 번호 · 열은 받는 쪽 그대로(두 열은
|
||||
따로 움직임) · 통합문서는 알약 `연결` 도 같이 오고 계산값은 서버가 다시 풂."""
|
||||
따로 움직임) · 통합문서는 알약 `연결` 도 같이 오고 계산값은 서버가 다시 풂 ·
|
||||
`source_root` = 원본 층 폴더(없으면 지금 층 · 권한은 부르는 쪽) · `body` = 원본 문서를 바로 줌
|
||||
(실무문서 조합 → 일위대가 · 39-1b)."""
|
||||
if kind not in _KEY:
|
||||
raise StoreError(404, f"없는 종류 「{kind}」")
|
||||
target = read(kind, column)["문서"]
|
||||
doc = {**read(kind, source)["문서"], **{k: target.get(k) for k in ("열", _KEY[kind])}}
|
||||
if body is None and source_root is not None:
|
||||
with at(source_root):
|
||||
body = read(kind, source)["문서"]
|
||||
body = read(kind, source)["문서"] if body is None else body
|
||||
doc = {**body, **{k: target.get(k) for k in ("열", _KEY[kind])}}
|
||||
doc = _server(kind, {k: v for k, v in doc.items() if k != "계산값"})
|
||||
with _LOCK:
|
||||
atomic_write_json(_path(kind, column), doc)
|
||||
return _info(kind, _path(kind, column))
|
||||
info = _info(kind, _path(kind, column))
|
||||
if kind == "area":
|
||||
_refresh_labor(column)
|
||||
return info
|
||||
|
||||
|
||||
def _copy_inputs(source: str, column: str) -> None:
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
"""일위대가 복사(PLAN 39-1b B5 · B6) — 항목 하나 복사(같은 층 · 다른 층 · 실무문서 조합) ·
|
||||
구조물 조합 복사 · 분류 더하기 = 새 일위대가번(조합 키는 안 받음).
|
||||
|
||||
시스템 층 = 견본 표(임시 폴더) · 조합 · 로직은 M01 마스터(읽기만) · 개인 층 = 임시 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.common_util_auth import verify_session
|
||||
from config import config_system
|
||||
from M02_MasterTemplete import M02_MasterTemplete_Items as items_store
|
||||
from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router
|
||||
from M02_MasterTemplete import M02_MasterTemplete_Store as store
|
||||
from M02_MasterTemplete import M02_Template_Layers as layers
|
||||
from resources.tester.test_m02_structure import write_sample_table
|
||||
|
||||
TEN = "GC000421"
|
||||
URL = "/api/m02/layers/{}/{}/{}/copy"
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def world(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
||||
system = tmp_path / "system"
|
||||
system.mkdir()
|
||||
write_sample_table(system)
|
||||
monkeypatch.setattr(store, "FOLDER", system)
|
||||
monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
|
||||
(tmp_path / "storage").mkdir()
|
||||
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(tmp_path / "storage"))
|
||||
with store.at(system):
|
||||
for cid in ("rv", "c1"):
|
||||
store.create_structure(cid)
|
||||
got = store.read("labor", "rv")
|
||||
logic = [{"로직": TEN, "입력값": {}, "수량": {"손": 2}}]
|
||||
store.write("labor", "rv", got["판"], {**got["문서"], "단위": "개소", "담은로직": logic})
|
||||
personal = layers.personal_dir(7, 42)
|
||||
items_store.ensure(personal)
|
||||
with store.at(personal):
|
||||
got = store.read("table", store.STRUCTURE_TABLE)
|
||||
got["문서"]["열"] = [{"id": "a", "머리": ["돌", "가"], "꼴": "수"}]
|
||||
store.write("table", store.STRUCTURE_TABLE, got["판"], got["문서"])
|
||||
session: dict[str, Any] = {"user_id": 42, "company_id": 7, "role": "USER", "is_master": False}
|
||||
app = FastAPI()
|
||||
app.include_router(items_router.router)
|
||||
app.dependency_overrides[verify_session] = lambda: session
|
||||
return {"client": TestClient(app), "session": session, "system": system, "personal": personal}
|
||||
|
||||
|
||||
def _as(world: dict, role: str = "USER") -> TestClient:
|
||||
world["session"]["role"] = role
|
||||
return world["client"]
|
||||
|
||||
|
||||
def _labor(folder: Path, cid: str) -> dict:
|
||||
return json.loads((folder / "labor" / f"{cid}.json").read_bytes())
|
||||
|
||||
|
||||
def test_같은_층_일위대가_하나_복사_번호_열은_받는_쪽(world):
|
||||
system = world["system"]
|
||||
before = _labor(system, "c1")
|
||||
got = _as(world, "SYSTEM_ADMIN").post(URL.format("system", "labor", "c1") + "?from=rv")
|
||||
assert got.status_code == 200
|
||||
c1 = _labor(system, "c1")
|
||||
assert (c1["열"], c1["일위대가번"]) == ("c1", before["일위대가번"])
|
||||
assert (c1["단위"], c1["담은로직"][0]["로직"]) == ("개소", TEN)
|
||||
assert c1["계산값"]["반영합"]["계"] > 0 # 서버가 다시 셈
|
||||
|
||||
|
||||
def test_다른_층에서_내_개인으로_저장_뒤에도_덮어씀(world):
|
||||
client, personal = _as(world), world["personal"]
|
||||
url = URL.format("personal", "labor", "a") + "?from=rv&from_layer=system"
|
||||
for _ in range(2): # 저장된 데이터도 다시 바꿈
|
||||
assert client.post(url).status_code == 200
|
||||
assert _labor(personal, "a")["담은로직"][0]["로직"] == TEN
|
||||
area = URL.format("personal", "area", "a") + "?from=rv&from_layer=system"
|
||||
assert client.post(area).status_code == 200
|
||||
|
||||
|
||||
def test_권한_없는_원본_층_403_없는_원본_404(world):
|
||||
client = _as(world)
|
||||
other = URL.format("personal", "labor", "a") + "?from=x&from_layer=personal&from_owner=7/99"
|
||||
assert client.post(other).status_code == 403
|
||||
missing = URL.format("personal", "labor", "a") + "?from=없는열&from_layer=system"
|
||||
assert client.post(missing).status_code == 404
|
||||
# 시스템 층 쓰기는 시스템 관리자만
|
||||
assert client.post(URL.format("system", "labor", "c1") + "?from=rv").status_code == 403
|
||||
|
||||
|
||||
def test_실무문서_조합에서_일위대가로_시스템_관리자만(world):
|
||||
url = URL.format("system", "labor", "c1") + "?from=UA000008&from_layer=practice"
|
||||
assert _as(world).post(url).status_code == 403
|
||||
admin = _as(world, "SYSTEM_ADMIN")
|
||||
assert admin.post(url).status_code == 200
|
||||
c1 = _labor(world["system"], "c1")
|
||||
assert c1["본뜬"] == {"층": "system", "주인": None, "키": "UA000008"}
|
||||
assert len(c1["담은로직"]) == 14 and all(x["입력값"] == {} for x in c1["담은로직"])
|
||||
area = URL.format("system", "area", "c1") + "?from=UA000008&from_layer=practice"
|
||||
assert admin.post(area).status_code == 403 # 실무문서는 일위대가로만
|
||||
bad = URL.format("system", "labor", "c1") + "?from=UP1-000001&from_layer=practice"
|
||||
assert admin.post(bad).status_code == 400
|
||||
|
||||
|
||||
def test_구조물_조합_복사는_새_일위대가번_조합_키는_안_받음(world):
|
||||
system, personal = world["system"], world["personal"]
|
||||
with store.at(system):
|
||||
key = next(r["키"] for r in store.read_combos()["줄"] if r["열"] == "rv")
|
||||
body = {"from_layer": "system", "key": key}
|
||||
made = _as(world).post("/api/m02/layers/personal/items/copy", json=body).json()
|
||||
got = _labor(personal, made["열"])
|
||||
src = _labor(system, "rv")
|
||||
assert got["담은로직"] == src["담은로직"] and got["일위대가번"] != src["일위대가번"]
|
||||
assert got["일위대가번"].startswith("SU-")
|
||||
table = json.loads((personal / "table" / f"{store.STRUCTURE_TABLE}.json").read_bytes())
|
||||
assert next(c for c in table["열"] if c["id"] == made["열"]).get("일위대가") is None
|
||||
|
||||
|
||||
def test_분류_더하기_복사도_새_번호로_일위대가_문서(world):
|
||||
system = world["system"]
|
||||
with store.at(system):
|
||||
got = store.read("table", store.STRUCTURE_TABLE)
|
||||
doc = got["문서"]
|
||||
rv = next(c for c in doc["열"] if c["id"] == "rv")
|
||||
doc["열"].append({**rv, "id": "rv2", "본뜬열": "rv", "일위대가": None})
|
||||
store.write("table", store.STRUCTURE_TABLE, got["판"], doc)
|
||||
new, src = _labor(system, "rv2"), _labor(system, "rv")
|
||||
assert new["담은로직"] == src["담은로직"] and new["일위대가번"] != src["일위대가번"]
|
||||
@@ -211,3 +211,32 @@ def test_길_틀린_문서_400_없는_열_404_남의_개인_층_403(client):
|
||||
assert client.post(url.format("system", "없는열"), json={}).status_code == 404
|
||||
other = client.post(url.format("personal", "rv") + "?owner=7/99", json={})
|
||||
assert other.status_code == 403
|
||||
|
||||
|
||||
def test_로직_지우기_검사는_모든_층_일위대가를_봄(tmp_path, monkeypatch):
|
||||
from config import config_system
|
||||
from M01_MasterData import M01_MasterData_Store_Cache as cache
|
||||
|
||||
system, storage = tmp_path / "system", tmp_path / "storage"
|
||||
for folder, code in (
|
||||
(system / "labor", STONE),
|
||||
(storage / "7" / "templates" / "labor", TEN), # 회사 층
|
||||
(storage / "7" / "42" / "templates" / "labor", STONE), # 개인 층
|
||||
(storage / "7" / "42" / "p1" / "templates" / "labor", STONE), # 프로젝트 — 안 봄
|
||||
):
|
||||
folder.mkdir(parents=True)
|
||||
body = {"담은로직": [{"로직": code}]}
|
||||
(folder / "a.json").write_text(json.dumps(body), encoding="utf-8")
|
||||
from M02_MasterTemplete import M02_Template_Layers as layers
|
||||
|
||||
monkeypatch.setattr(layers, "SYSTEM_ROOT", system) # 검사는 39-1d 한 벌(층 자리 = layers)
|
||||
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
|
||||
|
||||
class Gone: # STONE 을 지운 마스터
|
||||
def logic(self, code):
|
||||
if code == STONE:
|
||||
raise cache.cm.mf.FormulaError("없음")
|
||||
|
||||
got = cache.labor_errors(Gone())
|
||||
assert len(got) == 2 and all(STONE in x for x in got)
|
||||
assert any("마스터" in x for x in got) and any("개인 7/42" in x for x in got)
|
||||
|
||||
@@ -175,7 +175,8 @@ def test_회사_항목을_개인으로_복사(world):
|
||||
made = got.json()
|
||||
assert made["열"] == "cw" and made["key"] == "SP42-000001" # 응답 = 개인 층 새 키(바로 고름)
|
||||
mine_rows = user.get("/api/m02/items?layer=personal").json()
|
||||
assert mine_rows[0]["key"] == made["key"] and mine_rows[0]["names"]["일위대가"] == ua_name
|
||||
assert mine_rows[0]["key"] == made["key"]
|
||||
assert mine_rows[0]["names"]["일위대가"] is None # 조합 키는 안 받음 — 일위대가 문서로(39-1b)
|
||||
mine = "/api/m02/layers/personal"
|
||||
copy = user.get(f"{mine}/templates/structure/cw").json()["문서"]
|
||||
assert copy["도면"]["entities"] == [{"type": "line"}] and copy["열"] == "cw"
|
||||
|
||||
Reference in New Issue
Block a user