"""프로젝트 집계 항목 복사본 완전체 (PLAN 25-2b · 25-2e · 26-2b · 임시 저장소 · 임시 마스터 사본). 복사본은 구조물마다(`templates/structures/{구조물 id}/` · 관은 `pipe@측점`). - 지정 → [저장] → 폴더에 새 파일(입력연결 · 입력값 · 고른자재 · 단가) · 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 = "pipe@40.0" # 관 지점 복사본 폴더 = 구조물 id(26-2b) PIPE_PATH = ["관공", "(관종·관경마다)", "관매설"] 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/structures" / 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/structures" / 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, "경로": PIPE_PATH}, "원본판": 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/structures" / 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] # 안 고쳤으면 지움 assert not (root / "templates/structures/_지난").exists() _save(root, [_pipe()]) current = copies.read_file(root, FOLDER, "입력값") copies.write_file(root, FOLDER, "입력값", current["판"], {"GF000176": {"관경": {"값": 600}}}) assert _save(root, [_pipe(None)])["지움"] == [FOLDER] assert copies.list_items(root) == [] past = list((root / "templates/structures/_지난").iterdir()) assert len(past) == 1 and past[0].name.startswith("pipe@40.0_") assert _read(past[0] / "입력값.json") == {"GF000176": {"관경": {"값": 600}}} def test_미리_가져오기(world): root = world["root"] _save(root, []) made = copies.prefetch(root, FOLDER, PIPE) assert made["폴더"] == FOLDER and made["만듦"] is True assert copies.prefetch(root, FOLDER, PIPE)["만듦"] is False # 같은 항목이면 그대로 assert copies.list_items(root)[0]["구조물"] == [FOLDER] assert _save(root, [])["지움"] == [FOLDER] # [저장] 뒤에도 안 쓰면 지움 with pytest.raises(ValueError): copies.prefetch(root, "../x", PIPE) def test_옛_복사본은_읽을_때_채움(world): root = world["root"] _save(root, [_pipe()]) folder = root / "templates/structures" / 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()["멈춤"] wall = {"structure": "st9", "layer": "system", "owner": None} assert client.post(base, json={**wall, "summary_item": {"layer": "x"}}).status_code == 400 assert client.post(base, json={"layer": "system", "key": "SS000022"}).status_code == 400 made = client.post(base, json={**wall, "key": "SS000022"}).json() assert made["폴더"] == "st9" and made["만듦"] is True names = {i["폴더"]: i["이름"] for i in client.get(base).json()["항목"]} assert names["st9"] == "돌쌓기 (찰·메마다) H=(높이마다)" # 층 SS 줄 이름(경로 이음) assert client.post(base, json={**wall, "key": "SS999999"}).status_code == 404 assert client.post(f"{base}/{FOLDER}/reapply").json()["덮임"] == [] session["company_id"] = 4 assert client.get(base).status_code == 403 assert client.post(f"{base}/{FOLDER}/refresh").status_code == 403 # ── 구조물마다 복사본 (26-2b) ───────────────────────── MS = {"layer": "system", "owner": None, "key": "SS000022", "name": "돌쌓기"} def _wall(sid: str, summary: dict | None = MS) -> dict[str, Any]: return { "structure_id": sid, "type_id": "masonry_dry", "placement": "interval", "chainage_m": 85, "start_m": 80, "end_m": 90, "options": {"height_m": 2.0, "length_m": 10}, "summary_item": summary, } def _save_all(root: Path, walls: list[dict], pipes: list[dict]) -> dict[str, list[str]]: (root / "B05_Profile/route/structures.json").write_text( json.dumps({"revision": 1, "structures": walls}, ensure_ascii=False), 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 _memo(root: Path, folder: str, text: str) -> None: """면적 시트에 표식 한 칸 — 한쪽만 고쳤는지 보려는 것(계산값 풀이는 건너뜀).""" area = copies.read_file(root, folder, "면적") body = {**area["내용"], "메모": text} copies.write_file(root, folder, "면적", area["판"], body, recalc=lambda doc: doc) def test_같은_항목_구조물_둘은_따로(world): root = world["root"] second = {**_pipe(), "chainage_m": 60.0} made = _save_all(root, [_wall("a1"), _wall("a2")], [_pipe(), second])["만듦"] assert made == ["a1", "a2", FOLDER, "pipe@60.0"] _memo(root, "a1", "a1 만 고침") assert copies.read_file(root, "a1", "면적")["내용"]["메모"] == "a1 만 고침" assert "메모" not in copies.read_file(root, "a2", "면적")["내용"] # 다른 쪽 그대로 _fill_values(root) # pipe@40.0 에만 값 assert copies.calc(root, FOLDER)["멈춤"] == [] and copies.calc(root, FOLDER)["계"] > 0 assert len(copies.calc(root, "pipe@60.0")["멈춤"]) == 3 # 계산도 각자 flags = {i["폴더"]: i["고침"] for i in copies.list_items(root)} assert flags == {"a1": True, "a2": False, FOLDER: True, "pipe@60.0": False} def test_원본_항목을_지워도_복사본_계산_그대로(world): root, system = world["root"], world["system"] _save(root, [_pipe()]) _fill_values(root) before = copies.calc(root, FOLDER) combo = _read(system / layers.COMBO_FILE) combo["줄"] = [r for r in combo["줄"] if r["키"] != PIPE["key"]] (system / layers.COMBO_FILE).write_text(json.dumps(combo, ensure_ascii=False), "utf-8") assert copies.calc(root, FOLDER) == before row = copies.list_items(root)[0] assert row["원본없음"] is True and row["마스터바뀜"] is False assert _save(root, [_pipe()])["만듦"] == [] # 다시 저장해도 복사본 그대로 with pytest.raises(LookupError): copies.reapply(root, FOLDER) # 원본이 없으면 재적용 못 함(404) def test_재적용은_시트_고침을_덮고_알림_입력값은_살림(world): root = world["root"] _save(root, [_pipe()]) values = _fill_values(root) _memo(root, FOLDER, "x") assert copies.reapply(root, FOLDER)["덮임"] == ["면적"] assert "메모" not in copies.read_file(root, FOLDER, "면적")["내용"] assert copies.read_file(root, FOLDER, "입력값")["내용"] == values assert copies.reapply(root, FOLDER)["덮임"] == [] def test_개인에_저장하면_개인_층에_항목(world): root = world["root"] _save(root, [_pipe()]) rows = [{"로직": "GF000176", "입력": "관경", "자리": "선택"}] current = copies.read_file(root, FOLDER, "입력연결") copies.write_file(root, FOLDER, "입력연결", current["판"], {"판": 3, "줄": rows}) made = copies.save_as(root, FOLDER, 3, 12, ["관공", "우리 현장"], "파형강관 매설") assert made["summary_item"] == { "layer": "personal", "owner": "3/12", "key": made["key"], "name": "관공 우리 현장 파형강관 매설", } folder = layers.personal_dir(3, 12) row = next(r for r in _read(folder / layers.COMBO_FILE)["줄"] if r["키"] == made["key"]) assert [row["종류"], row["분류1"], row["분류2"]] == ["관공", "우리 현장", "파형강관 매설"] assert row["일위대가"] == "UA000036" and row["단위"] == "m" copy = root / "templates/structures" / FOLDER drawing = _read(folder / "structure" / f"{made['열']}.json") assert drawing["도면"] == _read(copy / "도면.json")["도면"] assert _read(folder / "inputs" / f"{made['열']}.json") == {"판": 1, "줄": rows} assert not list(folder.rglob("단가.json")) # 단가 묶음은 안 가져감 # 개인 층 항목을 다른 구조물에 지정하면 그 내용으로 복사본이 섬 assert _save_all(root, [_wall("b1", made["summary_item"])], [])["만듦"] == ["b1"] def test_프로젝트에서_새_항목(world): root = world["root"] _save_all(root, [_wall("c1")], []) made = copies.new_item(root, "c1", 3, 12, ["신규 공사"], "돌붙임", "㎡") assert made["폴더"] == "c1" and made["summary_item"]["layer"] == "personal" origin = _read(root / "templates/structures/c1/manifest.json")["원본"] assert origin["key"] == made["key"] and origin["경로"] == ["신규 공사", "돌붙임"] assert copies.list_items(root)[0]["이름"] == "신규 공사 돌붙임" # 화면이 새 항목을 지정해 [저장] 하면 그대로 남음 assert _save_all(root, [_wall("c1", made["summary_item"])], [])["만듦"] == [] with pytest.raises(ValueError): copies.new_item(root, "c2", 3, 12, [], "", None) # 경로 · 이름 빈 값 def test_옛_항목_폴더는_쓰는_구조물마다_옮겨_받음(world): root = world["root"] _save_all(root, [_wall("d1"), _wall("d2")], []) base = root / "templates/structures" old = root / "templates/items/system_none_SS000022" old.parent.mkdir(parents=True) shutil.move(str(base / "d1"), str(old)) shutil.rmtree(base / "d2") manifest = _read(old / "manifest.json") manifest.update({"구조물": ["d1", "d2"], "고침": True}) (old / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False), "utf-8") (old / "입력값.json").write_text(json.dumps({"GF": {"a": {"값": 1}}}), "utf-8") assert [i["폴더"] for i in copies.list_items(root)] == ["d1", "d2"] assert not (root / "templates/items").exists() for sid in ("d1", "d2"): assert _read(base / sid / "입력값.json") == {"GF": {"a": {"값": 1}}} # 고친 값 유지 assert _read(base / sid / "manifest.json")["구조물"] == [sid] assert _save_all(root, [_wall("d1"), _wall("d2")], [])["만듦"] == [] def test_지정을_바꾸면_고친_옛_복사본은_지난에(world): root = world["root"] _save_all(root, [_wall("e1")], []) _memo(root, "e1", "y") rv = {"layer": "system", "owner": None, "key": "SS000001", "name": "돌기슭막이"} result = _save_all(root, [_wall("e1", rv)], []) assert result["만듦"] == ["e1"] and result["지움"] == ["e1"] assert _read(root / "templates/structures/e1/manifest.json")["원본"]["key"] == "SS000001" past = list((root / "templates/structures/_지난").iterdir()) assert len(past) == 1 and _read(past[0] / "면적.json")["메모"] == "y" def test_새_항목_길_회사_없으면_409(world, monkeypatch): root = world["root"] _save_all(root, [_wall("f1")], []) async def fake_row(project_id: str) -> dict[str, Any]: return {"company_id": None, "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": "SYSTEM_ADMIN", "company_id": None, "user_id": 12} app.dependency_overrides[verify_session] = lambda: session client = TestClient(app) body = {"structure": "f1", "경로": ["새"], "이름": "것"} assert client.post("/api/projects/p1/items/new", json=body).status_code == 409 save_as = client.post("/api/projects/p1/items/f1/save-as", json={"경로": ["a"], "이름": "b"}) assert save_as.status_code == 409