eomsangdonandClaude Opus 5.5
b038e99d8d
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BU9gbNE3aWFUvE61pS2WGt
343 lines
15 KiB
Python
343 lines
15 KiB
Python
"""프로젝트 집계 항목 복사본 완전체 (PLAN 25-2b · 25-2e · 임시 저장소 · 임시 마스터 사본).
|
|
|
|
- 지정 → [저장] → 폴더에 새 파일(입력연결 · 입력값 · 고른자재 · 단가) · manifest 가져온날 · 원본판
|
|
- 단가 묶음이 닫혀 있음 — 조합 37개 · 로직 170개를 마스터와 복사본으로 돌려 같은 값
|
|
- 마스터 단가 한 줄을 바꿔도 프로젝트 계산 그대로 · 목록에 「마스터 바뀜」 · 새로 가져오기 뒤 바뀜
|
|
(입력값 · 고른자재는 같은 키면 살림)
|
|
- 쓰기 뒤 다시 읽기 같음 · 판 다르면 409 · 모양 틀리면 400 · 고친 복사본은 안 지움
|
|
- 옛 복사본은 읽을 때 채움 · 미리 가져오기 · 남의 프로젝트 403
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
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 M01_MasterData import M01_MasterData_Store as m01
|
|
from M02_MasterTemplete import M02_Item_Copy as copies
|
|
from M02_MasterTemplete import M02_Item_Prices as prices
|
|
from M02_MasterTemplete import M02_Item_Snapshot as snap
|
|
from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layer_routes
|
|
from M02_MasterTemplete import M02_Project_Items_Router as items_router
|
|
from M02_MasterTemplete import M02_Template_Layers as layers
|
|
|
|
REAL = config_system.PROJECT_ROOT / "resources/master_template"
|
|
MASTER = config_system.PROJECT_ROOT / "resources/master_data"
|
|
PIPE = {"layer": "system", "owner": None, "key": "SS000011", "name": "관공 관매설"}
|
|
FOLDER = "system_none_SS000011"
|
|
LABOR_PRICE = "LB000002" # 보통인부 — 되메우기 · 파형강관 부설 호표가 씀
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def master_copy(tmp_path_factory) -> Path:
|
|
target = tmp_path_factory.mktemp("master_data")
|
|
for path in MASTER.glob("*.json"):
|
|
shutil.copy2(path, target / path.name)
|
|
return target
|
|
|
|
|
|
@pytest.fixture
|
|
def world(tmp_path: Path, monkeypatch, master_copy: Path) -> dict[str, Any]:
|
|
system = tmp_path / "master"
|
|
system.mkdir()
|
|
for name in (layers.COMBO_FILE, "자재조합.json"):
|
|
shutil.copy2(REAL / name, system / name)
|
|
for kind in ("table", "structure", "area", "material"):
|
|
shutil.copytree(REAL / kind, system / kind)
|
|
monkeypatch.setattr(layers, "SYSTEM_ROOT", system)
|
|
monkeypatch.setattr(m01, "FOLDER", master_copy)
|
|
storage = tmp_path / "storage"
|
|
monkeypatch.setattr(config_system, "STORAGE_BASE_DIR", str(storage))
|
|
root = storage / "3" / "12" / "proj"
|
|
(root / "B05_Profile/route").mkdir(parents=True)
|
|
(root / "B04_PreProcess/drainage/edits").mkdir(parents=True)
|
|
return {"system": system, "root": root, "master": master_copy}
|
|
|
|
|
|
def _save(root: Path, pipes: list[dict]) -> dict[str, list[str]]:
|
|
(root / "B05_Profile/route/structures.json").write_text(
|
|
json.dumps({"revision": 1, "structures": []}), encoding="utf-8"
|
|
)
|
|
(root / "B04_PreProcess/drainage/edits/pipe_points.json").write_text(
|
|
json.dumps({"points": pipes}, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
return snap.sync_snapshots(root)
|
|
|
|
|
|
def _pipe(summary: dict | None = PIPE) -> dict[str, Any]:
|
|
return {"chainage_m": 40.0, "source": "user", "summary_item": summary}
|
|
|
|
|
|
def _read(path: Path) -> Any:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _auto_values(root: Path) -> dict[str, Any]:
|
|
"""담은 로직마다 자동값을 확인한 값으로 — 사용자가 다 채운 것과 같게."""
|
|
labor = _read(root / "templates/items" / FOLDER / "일위대가.json")
|
|
_, whole = m01.loaded()
|
|
out: dict[str, Any] = {}
|
|
for row in labor["로직"]:
|
|
values = m01.mau.values(whole, whole.logic(row["키"]))
|
|
out[row["키"]] = {k: {"값": prices.to_json(v), "확인": True} for k, v in values.items()}
|
|
return out
|
|
|
|
|
|
def _fill_values(root: Path) -> dict[str, Any]:
|
|
current = copies.read_file(root, FOLDER, "입력값")
|
|
values = _auto_values(root)
|
|
copies.write_file(root, FOLDER, "입력값", current["판"], values)
|
|
return values
|
|
|
|
|
|
# ── 새 파일 ──────────────────────────────────────────
|
|
|
|
|
|
def test_지정하면_새_파일까지_복사(world):
|
|
root = world["root"]
|
|
assert _save(root, [_pipe()])["만듦"] == [FOLDER]
|
|
folder = root / "templates/items" / FOLDER
|
|
assert _read(folder / "입력연결.json") == {"판": None, "줄": []} # 층 연결이 아직 없음
|
|
assert _read(folder / "입력값.json") == {} and _read(folder / "고른자재.json") == {}
|
|
price = _read(folder / "단가.json")
|
|
rows = {k: len(v.get("줄") or v.get("표")) for k, v in price.items()}
|
|
assert rows["인력.json"] >= 2 and rows["재료_자재품목.json"] > 0 and "_빠진" not in price
|
|
keys = {row["키"] for data in price.values() for row in data.get("줄") or data.get("표")}
|
|
assert {"GF000176", "GC000088", "GF000113", "GC000267", LABOR_PRICE} <= keys
|
|
manifest = _read(folder / "manifest.json")
|
|
assert manifest["가져온날"] and len(manifest["원본판"]) == 16
|
|
assert copies.list_items(root) == [
|
|
{
|
|
"폴더": FOLDER,
|
|
"이름": "관공 관매설",
|
|
"원본": PIPE,
|
|
"원본판": manifest["원본판"],
|
|
"가져온날": manifest["가져온날"],
|
|
"마스터바뀜": False,
|
|
"원본없음": False,
|
|
"구조물": ["pipe@40.0"],
|
|
"고침": False,
|
|
}
|
|
]
|
|
|
|
|
|
def test_층_입력연결을_복사(world):
|
|
root, system = world["root"], world["system"]
|
|
(system / "inputs").mkdir()
|
|
link = {"판": "a1", "줄": [{"로직": "GF000176", "입력": "관경", "자리": "면적", "알약": "x"}]}
|
|
column = next(
|
|
r["열"] for r in _read(system / layers.COMBO_FILE)["줄"] if r["키"] == PIPE["key"]
|
|
)
|
|
(system / "inputs" / f"{column}.json").write_text(json.dumps(link, ensure_ascii=False), "utf-8")
|
|
_save(root, [_pipe()])
|
|
assert copies.read_file(root, FOLDER, "입력연결")["내용"] == link
|
|
|
|
|
|
def test_단가_묶음은_닫혀_있음(world):
|
|
"""조합마다 복사본 단가 묶음만으로 돌린 값 = 마스터로 돌린 값(자동값 · 고르기 끝값)."""
|
|
files, whole = prices.master()
|
|
combos = [r for d in files.values() if d.get("그룹") == "일위대가조합" for r in d["줄"]]
|
|
runs = 0
|
|
for ua in combos:
|
|
keys = prices.mcb.logic_keys(ua)
|
|
snapshot = prices.load_prices(prices.to_json(prices.collect(keys)))
|
|
for key in keys:
|
|
try:
|
|
row = whole.logic(key)
|
|
except prices.mf.FormulaError:
|
|
continue
|
|
for pick in (0, -1):
|
|
values = m01.mau.values(whole, row)
|
|
for spec in row.get("입력", []):
|
|
if spec.get("고르기"):
|
|
values[spec["이름"]] = spec["고르기"][pick]
|
|
assert _run(whole, key, values) == _run(snapshot, key, values), (ua["키"], key)
|
|
runs += 1
|
|
assert len(combos) >= 30 and runs >= 300
|
|
|
|
|
|
def _run(master: Any, key: str, values: dict) -> str:
|
|
try:
|
|
got = prices.mf.run(master, key, dict(values))
|
|
except Exception as error: # 까닭도 같아야 함
|
|
return f"멈춤 {error}"
|
|
return str(got.get("계", got.get("결과")))
|
|
|
|
|
|
# ── 계산 · 마스터 바뀜 · 새로 가져오기 ───────────────────
|
|
|
|
|
|
def test_입력이_없으면_빠진_입력_목록(world):
|
|
root = world["root"]
|
|
_save(root, [_pipe()])
|
|
got = copies.calc(root, FOLDER)
|
|
assert got["계"] == 0 and len(got["멈춤"]) == 3
|
|
assert {"로직": "GF000176", "입력": "관경"} in got["빠진입력"]
|
|
|
|
|
|
def test_마스터_단가를_바꿔도_계산_그대로_새로_가져오면_바뀜(world):
|
|
root, master_dir = world["root"], world["master"]
|
|
_save(root, [_pipe()])
|
|
values = _fill_values(root)
|
|
before = copies.calc(root, FOLDER)
|
|
assert before["빠진입력"] == [] and before["멈춤"] == [] and before["계"] > 0
|
|
files, whole = m01.loaded()
|
|
ua = next(
|
|
r
|
|
for d in files.values()
|
|
if d.get("그룹") == "일위대가조합"
|
|
for r in d["줄"]
|
|
if r["키"] == "UA000036"
|
|
)
|
|
given = {k: {n: v["값"] for n, v in one.items()} for k, one in values.items()}
|
|
expected = prices.to_json(prices.mcb.run(whole, ua, given))
|
|
assert before["계"] == pytest.approx(expected["계"]) and before["노무비"] > 0
|
|
labor_file = master_dir / "인력.json"
|
|
original = labor_file.read_bytes()
|
|
try:
|
|
data = _read(labor_file)
|
|
row = next(r for r in data["줄"] if r["키"] == LABOR_PRICE)
|
|
row["값"] = row["값"] * 2
|
|
labor_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
assert copies.calc(root, FOLDER) == before # 복사본만 씀
|
|
assert copies.list_items(root)[0]["마스터바뀜"] is True
|
|
copies.refresh(root, FOLDER)
|
|
after = copies.calc(root, FOLDER)
|
|
assert after["노무비"] > before["노무비"] and after["재료비"] == before["재료비"]
|
|
assert copies.read_file(root, FOLDER, "입력값")["내용"] == values # 같은 키 살림
|
|
assert copies.list_items(root)[0]["마스터바뀜"] is False
|
|
finally:
|
|
labor_file.write_bytes(original)
|
|
|
|
|
|
def test_새로_가져오면_없는_키는_버림(world):
|
|
root = world["root"]
|
|
_save(root, [_pipe()])
|
|
folder = root / "templates/items" / FOLDER
|
|
(folder / "입력값.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"GF000176": {"관경": {"값": 600, "확인": True}, "옛입력": {"값": 1}},
|
|
"GX999999": {"a": {"값": 1}},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
"utf-8",
|
|
)
|
|
(folder / "고른자재.json").write_text(
|
|
json.dumps({"GF000176": {"0": "MT000001", "99": "MT000002"}}), "utf-8"
|
|
)
|
|
copies.refresh(root, FOLDER)
|
|
assert _read(folder / "입력값.json") == {"GF000176": {"관경": {"값": 600, "확인": True}}}
|
|
assert _read(folder / "고른자재.json") == {"GF000176": {"0": "MT000001"}}
|
|
|
|
|
|
# ── 쓰기 · 판 · 모양 · 지우기 규칙 ─────────────────────────
|
|
|
|
|
|
def test_쓰기_뒤_다시_읽기_같음_판_다르면_409(world):
|
|
root = world["root"]
|
|
_save(root, [_pipe()])
|
|
first = copies.read_file(root, FOLDER, "고른자재")
|
|
wrote = copies.write_file(
|
|
root, FOLDER, "고른자재", first["판"], {"GF000176": {"0": "MT000010"}}
|
|
)
|
|
again = copies.read_file(root, FOLDER, "고른자재")
|
|
assert again["내용"] == {"GF000176": {"0": "MT000010"}} and again["판"] == wrote["판"]
|
|
with pytest.raises(copies.Stale):
|
|
copies.write_file(root, FOLDER, "고른자재", first["판"], {})
|
|
for kind, bad in (
|
|
("입력값", {"GF000176": {"관경": 600}}),
|
|
("입력값", {"GF000176": {"관경": {"값": 1, "확인": "네"}}}),
|
|
("고른자재", {"GF000176": {"첫": "MT000001"}}),
|
|
("입력연결", {"줄": [{"로직": "GF000176", "입력": "관경", "자리": "어딘가"}]}),
|
|
("도면", {}),
|
|
):
|
|
with pytest.raises(ValueError):
|
|
copies.write_file(root, FOLDER, kind, again["판"], bad)
|
|
with pytest.raises(LookupError):
|
|
copies.read_file(root, "../../etc", "입력값")
|
|
|
|
|
|
def test_고친_복사본은_구조물에서_빠져도_안_지움(world):
|
|
root = world["root"]
|
|
_save(root, [_pipe()])
|
|
assert _save(root, [_pipe(None)])["지움"] == [FOLDER] # 안 고쳤으면 지움(지금 규칙)
|
|
_save(root, [_pipe()])
|
|
current = copies.read_file(root, FOLDER, "입력값")
|
|
copies.write_file(root, FOLDER, "입력값", current["판"], {"GF000176": {"관경": {"값": 600}}})
|
|
assert _save(root, [_pipe(None)])["지움"] == []
|
|
assert copies.list_items(root)[0]["구조물"] == [] and copies.list_items(root)[0]["고침"]
|
|
|
|
|
|
def test_미리_가져오기(world):
|
|
root = world["root"]
|
|
_save(root, [])
|
|
made = copies.prefetch(root, PIPE)
|
|
assert made["폴더"] == FOLDER and made["만듦"] is True
|
|
assert copies.prefetch(root, PIPE)["만듦"] is False # 있으면 그대로
|
|
assert copies.list_items(root)[0]["구조물"] == []
|
|
assert _save(root, [])["지움"] == [FOLDER] # 안 고쳤으면 [저장] 때 지움
|
|
|
|
|
|
def test_옛_복사본은_읽을_때_채움(world):
|
|
root = world["root"]
|
|
_save(root, [_pipe()])
|
|
folder = root / "templates/items" / FOLDER
|
|
for name in ("입력연결.json", "입력값.json", "고른자재.json", "단가.json"):
|
|
(folder / name).unlink()
|
|
manifest = _read(folder / "manifest.json")
|
|
for key in ("원본판", "가져온날"):
|
|
manifest.pop(key)
|
|
(folder / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False), "utf-8")
|
|
assert copies.read_file(root, FOLDER, "입력값")["내용"] == {}
|
|
filled = _read(folder / "manifest.json")
|
|
assert filled["가져온날"] == filled["적용일"] and filled["옮겨받음"]
|
|
assert (folder / "단가.json").is_file() and copies.calc(root, FOLDER)["멈춤"]
|
|
assert copies.list_items(root)[0]["마스터바뀜"] is False # 원본이 안 바뀐 옛 복사본
|
|
|
|
|
|
# ── 길 · 권한 ────────────────────────────────────────
|
|
|
|
|
|
def test_길_권한_남의_프로젝트는_403(world, monkeypatch):
|
|
root = world["root"]
|
|
_save(root, [_pipe()])
|
|
|
|
async def fake_row(project_id: str) -> dict[str, Any]:
|
|
return {"company_id": 3, "storage_path": str(root)}
|
|
|
|
monkeypatch.setattr(layer_routes, "_project_row", fake_row)
|
|
monkeypatch.setattr(layer_routes, "resolve_stored_project_path", lambda path: path)
|
|
app = FastAPI()
|
|
app.include_router(items_router.router)
|
|
session = {"role": "USER", "company_id": 3}
|
|
app.dependency_overrides[verify_session] = lambda: session
|
|
client = TestClient(app)
|
|
base = "/api/projects/p1/items"
|
|
assert client.get(base).json()["항목"][0]["폴더"] == FOLDER
|
|
got = client.get(f"{base}/{FOLDER}/입력값").json()
|
|
put = client.put(f"{base}/{FOLDER}/입력값", json={"판": got["판"], "내용": {}})
|
|
assert put.status_code == 200
|
|
stale = client.put(f"{base}/{FOLDER}/입력값", json={"판": got["판"] + "x", "내용": {}})
|
|
assert stale.status_code == 409 and stale.json()["detail"]["판"] == put.json()["판"]
|
|
assert client.get(f"{base}/{FOLDER}/모름").status_code == 422
|
|
assert client.post(f"{base}/{FOLDER}/calc").json()["멈춤"]
|
|
assert client.post(base, json={"summary_item": {"layer": "x"}}).status_code == 400
|
|
made = client.post(base, json={"layer": "system", "owner": None, "key": "SS000022"}).json()
|
|
assert made["폴더"] == "system_none_SS000022" and made["만듦"] is True
|
|
names = {i["폴더"]: i["이름"] for i in client.get(base).json()["항목"]}
|
|
assert names["system_none_SS000022"] == "돌쌓기 (찰·메마다) H=(높이마다)" # 층 SS 줄 이름
|
|
assert client.post(base, json={"layer": "system", "key": "SS999999"}).status_code == 404
|
|
session["company_id"] = 4
|
|
assert client.get(base).status_code == 403
|
|
assert client.post(f"{base}/{FOLDER}/refresh").status_code == 403
|