merge(511-B · 511-C 합치기): origin/sub_laptop_4(511-C 유토 배분 E2 · 판 haul 줄) 받음 — 공유 파일 충돌 풂

- Node 입구 design_calc_node.ts 등록부 밖 id = 시험 길 · 단계 · 화면 길(PAGE_IDS) · haul.view
- calc_runner = PAGE_IDS(cross.refresh) · VIEW_IDS(haul.view) 둘 다 받음
- calc_sites.py 머리 글 = B-4 cross · 511-C haul(기본 server) 함께 · 묶음 13 줄
- 시험 309_2 · 309_2_browser · 511_b1 = 열린 줄에 haul 더함 · 화면 시험 「지금 서버」 = haul 기본 server 로 1 줄(바꾼 뒤 2 줄)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-22 15:55:17 +09:00
co-authored by Claude Opus 5.5
19 changed files with 403 additions and 65 deletions
File diff suppressed because one or more lines are too long
+13 -10
View File
@@ -56,7 +56,7 @@ def test_묶음_여덟_줄_등록부_전부():
sites = _registry_sites()
assert len(sites) == 45
nos = [no for group in calc_sites.GROUPS for no in group["nos"]]
assert len(calc_sites.GROUPS) == 12 # 511-B 단계 다섯 · F9 폼 기본값
assert len(calc_sites.GROUPS) == 13 # 511-B 단계 다섯 · F9 폼 기본값 · 511-C haul
assert len(nos) == len(set(nos)) and set(nos) == set(sites)
assert calc_sites.SWITCHABLE == {
"ground_sample",
@@ -67,13 +67,14 @@ def test_묶음_여덟_줄_등록부_전부():
"stations",
"profile", # 511-B B-3 엶
"cross", # 511-B B-4 끝에 엶
"haul", # 511-C E2 엶(기본 server)
"fixed_browser",
} # 511-A 셋 더 · 511-B B-2 stations · B-5 route
for group in calc_sites.GROUPS:
got = {sites[no] for no in group["nos"]}
if group["fixed"] == "browser":
assert got == {"browser_only"}, group["key"]
elif group["fixed"] == "server":
elif group["fixed"] == "server" or group.get("default") == "server":
assert got <= {"server_canon", "server_only"}, group["key"]
else:
assert got == {"browser_now"}, group["key"]
@@ -83,17 +84,19 @@ def test_묶음_여덟_줄_등록부_전부():
def test_커밋한_판_기본값_없음_깨짐(tmp_path, monkeypatch):
committed = json.loads(calc_sites.PATH.read_text(encoding="utf-8"))
assert committed["sites"] == dict.fromkeys(calc_sites.SWITCHABLE, "both")
assert set(calc_sites.read_sites()["sites"].values()) == {"both"}
assert calc_sites.DEFAULTS == {**dict.fromkeys(calc_sites.SWITCHABLE, "both"), "haul": "server"}
assert set(committed["sites"].items()) <= set(calc_sites.DEFAULTS.items())
assert calc_sites.read_sites()["sites"] == calc_sites.DEFAULTS # 판 파일 haul 없음 = server
monkeypatch.setattr(calc_sites, "PATH", tmp_path / "none.json")
assert calc_sites.read_sites() == {
"version": 0,
"sites": dict.fromkeys(calc_sites.SWITCHABLE, "both"),
}
assert calc_sites.read_sites() == {"version": 0, "sites": calc_sites.DEFAULTS}
broken = tmp_path / "broken.json"
broken.write_text('{"sites": {"ground_sample": "server", "corridor_mesh": "x"', "utf-8")
monkeypatch.setattr(calc_sites, "PATH", broken)
assert set(calc_sites.read_sites()["sites"].values()) == {"both"}
assert calc_sites.read_sites()["sites"] == calc_sites.DEFAULTS
odd = tmp_path / "odd.json"
odd.write_text('{"sites": {"haul": "x", "quantity": "server"}}', "utf-8")
monkeypatch.setattr(calc_sites, "PATH", odd)
assert calc_sites.read_sites()["sites"] == {**calc_sites.DEFAULTS, "quantity": "server"}
class _Cursor:
@@ -167,7 +170,7 @@ def test_잠금_모르는_줄_400_감사_한_줄(client):
done = http.put(url, json={"sites": {"ground_sample": "server"}})
assert done.status_code == 200 and done.json()["changed"] == ["ground_sample"]
saved = json.loads(board.read_text("utf-8"))
sites = {**dict.fromkeys(calc_sites.SWITCHABLE, "both"), "ground_sample": "server"}
sites = {**calc_sites.DEFAULTS, "ground_sample": "server"}
assert saved == {"version": 2, "sites": sites}
inserts = [args for sql, args in log if "system_audit_logs" in sql]
assert len(inserts) == 1
@@ -148,7 +148,7 @@ def test_시스템_관리자_카드_창_G1_서버_길(browser):
)
heads = [row for row in rows if row["head"]]
calcs = [row for row in rows if not row["head"]]
assert len(heads) == 12 and len(calcs) == 45 # 511-B 단계 다섯 줄 · F9 폼 기본값
assert len(heads) == 13 and len(calcs) == 45 # 511-B 단계 다섯 줄 · F9 · 511-C haul
whys = {row["name"].split(" · ")[0]: row["why"] for row in calcs}
for no, n in {"F2": 5}.items():
assert whys[no] == f"짝 모으는 중 · B-{n}", no
@@ -159,11 +159,12 @@ def test_시스템_관리자_카드_창_G1_서버_길(browser):
opened += ["R1", "R3", "R5", "R6", "R7", "F3"] # 511-B B-5 노선 · B-2 측점
opened += ["P1", "P2", "P3", "P4", "P8"] # 511-B B-3 종단 엶
opened += ["G2", "G4", "C1", "C2", "C3", "C5", "C6", "F4", "F5", "F6"] # 511-B B-4
opened += ["E2"] # 511-C 유토 배분 엶(기본 server = 체크 꺼짐)
opened += ["R2", "R4", "R8", "P7", "C7", "D4"] # ⑧ 시험용
assert [name.split(" · ")[0] for name in editable] == opened
assert all(row["name"].split(" · ")[1] for row in calcs) # 이름 = 등록부 label
assert all(row["boxes"][1][1] for row in calcs) # 서버 정본 칸은 늘 잠금
assert "지금 서버 0 줄" in dialog.inner_text()
assert "지금 서버 1 줄" in dialog.inner_text() # haul 기본 server(511-C)
SHOTS.mkdir(parents=True, exist_ok=True)
dialog.screenshot(path=str(SHOTS / "light.png"))
page.evaluate("document.documentElement.setAttribute('data-theme', 'dark')")
@@ -171,7 +172,7 @@ def test_시스템_관리자_카드_창_G1_서버_길(browser):
page.evaluate("document.documentElement.setAttribute('data-theme', 'light')")
page.click("input[data-calc-site='ground_sample']")
assert "지금 서버 1 줄" in dialog.inner_text()
assert "지금 서버 2 줄" in dialog.inner_text()
assert not page.is_checked("input[data-calc-no='G1']") # 머리 따라 줄도 꺼짐
page.locator(".b01-dashboard__calc-sites .ui-modal__actions button").click()
page.wait_for_selector(".b01-dashboard__calc-sites", state="detached", timeout=30000)
+3 -3
View File
@@ -23,15 +23,15 @@ def _patch(monkeypatch, outputs: list) -> list:
async def _none(_project_id):
return {}
def _run(_bundle, _script, payload):
calls.append(payload)
def _run(result, context):
calls.append(result)
return outputs.pop(0)
monkeypatch.setattr(router, "haul_inputs_for", _none)
monkeypatch.setattr(router, "conversion_factors_for", _none)
monkeypatch.setattr(router, "haul_limits_for", _none)
monkeypatch.setattr(router, "_mass_haul_context", lambda *_: {"limits": []})
monkeypatch.setattr(router, "run_bundle_json", _run)
monkeypatch.setattr(router, "_run_view", _run)
router._KEPT.clear()
return calls
@@ -48,12 +48,12 @@ def node(monkeypatch, tmp_path):
bundle.write_text("//", encoding="utf-8")
calls: list = []
def fake_run(path, script, payload):
def fake_run(payload):
calls.append(payload)
return {"areas": [], "mass_haul": {"total": len(calls)}}
monkeypatch.setattr(prebuild, "BUNDLE", bundle)
monkeypatch.setattr(prebuild, "run_bundle_json", fake_run)
monkeypatch.setattr(prebuild, "NODE_BUNDLE", bundle)
monkeypatch.setattr(prebuild, "_run_node", fake_run)
monkeypatch.setattr(prebuild, "bundle_stale", lambda path: False)
monkeypatch.setattr(prebuild, "_NODE_CACHE", {})
return SimpleNamespace(calls=calls, bundle=bundle)
@@ -105,7 +105,7 @@ def test_TS_가_번들보다_새로우면_다시(node, monkeypatch):
def test_실패는_기억_안_함(node, monkeypatch):
monkeypatch.setattr(prebuild, "run_bundle_json", lambda *a: node.calls.append(a) or None)
monkeypatch.setattr(prebuild, "_run_node", lambda *a: node.calls.append(a) or None)
assert prebuild._node_output(_input()) is None
assert prebuild._node_output(_input()) is None
assert len(node.calls) == 2
@@ -241,13 +241,13 @@ def test_같은_입력_두_번이면_Node_0_쓰기_0(node, monkeypatch, tmp_path
written_haul.append(data_patch)
stored["data"]["mass_haul"] = data_patch["mass_haul"]
def fake_run(path, script, payload):
def fake_run(payload):
node.calls.append(payload)
# Node 에 간 입력이 곧 지문 입력 — 다른 입력 길 없음
assert set(payload) == {"detail", "context"}
return {"areas": [{"chainage_m": 0.0, "cut_area_m2": 5.0}], "mass_haul": {"total": 9.0}}
monkeypatch.setattr(prebuild, "run_bundle_json", fake_run)
monkeypatch.setattr(prebuild, "_run_node", fake_run)
monkeypatch.setattr(b06_router, "get_section_detail", detail)
monkeypatch.setattr(prebuild, "run_with_connection", run_with)
monkeypatch.setattr(prebuild, "resolve_stored_project_path", lambda path: str(tmp_path))
+10 -1
View File
@@ -266,7 +266,16 @@ def test_서버_쪽_코드_브라우저_밖(loaded, reach) -> None:
if row["site"] in ("server_canon", "server_only") and row["ts"]
]
assert server # 유토 배분(common_util_mass_haul_balance)
assert [path for path in server if path in browser] == []
# 511-C(요구 321) — 유토 배분은 판 haul = both 때만 dynamic import(첫 묶음 밖) · 정적 import 0
lazy = "common_util/common_util_mass_haul_balance.ts"
assert [path for path in server if path in browser] == [lazy]
static = re.compile(
r'^import\s+(?!type\s)[^;]*?from\s+"@util/common_util_mass_haul_balance"', re.M
)
ts_files = [p for p in browser if p.endswith(".ts")]
assert [p for p in ts_files if static.search((ROOT / p).read_text(encoding="utf-8"))] == []
haul_api = (ROOT / "B05_Profile" / "B05_Profile_Api_HaulPlan.ts").read_text(encoding="utf-8")
assert 'await import("@util/common_util_mass_haul_balance")' in haul_api
assert NODE_ENTRY not in browser
assert reach[REGISTRY] == sorted(REGISTRY_PARTS) # 순수 자료 — 제 몫 파일만(나눔 · 함수 안 끔)
@@ -98,6 +98,7 @@ def test_판_다섯_줄_잠금_기본값_그대로():
"stations", # B-2 끝 = 엶(511-B B-2)
"profile", # B-3 엶(511-B · test_511_b3_profile_node)
"cross",
"haul", # 511-C E2 엶(기본 server)
"fixed_browser",
}
committed = subprocess.run( # noqa: S603 — 고정 실행 파일
+199
View File
@@ -0,0 +1,199 @@
# -*- coding: utf-8 -*-
"""511-C (요구 320-4) — 유토 배분 E2 = 늘 떠 있는 Node(TS 한 벌) · 옛 server-calc 묶음 값과 0 차.
못박는 것
- /haul-plan 창구 새 길(`_run_view` → 시험 길 `haul.view`) = 옛 server-calc `haul_plan_for` 갈래
= 브라우저 자리 함수(`computeHaulPlan` 바로 부름) — 프로젝트 일곱
(저장 유토곡선 · DB 에서 한 번 뜬 `fixtures/haul_511_c.json`)
× 문맥(기본 · 장비 경계 없음 · 구조물 잔토 · 채집석) × 그림 · 수량.
- 끝 경우 — 측점 하나(배분 없음 null) · 둘 · 전부 0 · 전부 절토 · 전부 성토.
- [저장] 정본 새 길(`Server_Calc_Prebuild._node_output` → 늘 떠 있는 Node 셋) = 옛 server-calc
`{detail, context}` 답(면적 · 다단 벽 · 유토곡선 + 배분 두 벌) — golden_472 · 문맥 셋.
./venv/Scripts/python.exe -m pytest resources/tester/test_511_c_haul_node.py -q
"""
from __future__ import annotations
import copy
import json
import shutil
import subprocess
import sys
from pathlib import Path
import pytest
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from B06_Section import B06_Section_Router_HaulPlan as haul_router # noqa: E402
from B06_Section import B06_Section_Server_Calc_Prebuild as prebuild # noqa: E402
from common_util.common_util_node_bundle import run_bundle_json # noqa: E402
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not (ROOT / "config" / "node_modules" / "vite").is_dir(),
reason="node · 프론트엔드 의존성 없음",
)
PROJECTS = json.loads((HERE / "fixtures" / "haul_511_c.json").read_text(encoding="utf-8"))
GOLDEN = HERE / "fixtures" / "golden_472"
SPOIL_POINTS = [
{"chainage_m": 40.0, "spoil_m3": 35.5, "ground_type": "soil"},
{"chainage_m": 260.0, "spoil_m3": 12.25, "ground_type": "ripping_rock"},
{"chainage_m": 600.0, "spoil_m3": 8.0, "ground_type": None},
]
def _result(payload: dict) -> dict:
"""저장 유토곡선 → 누가토량 결과 꼴(최소 · 최대 누계는 측점에서)."""
cums = [p["cumulative_volume_m3"] for p in payload["points"]]
return {
**payload,
"min_cumulative_m3": min(cums, default=0.0),
"max_cumulative_m3": max(cums, default=0.0),
}
def _contexts() -> dict[str, dict]:
base = prebuild._mass_haul_context()
return {
"기본": base,
"장비 경계 없음": {**base, "haul_equipment_limits": None},
"구조물 잔토": {
**base,
"structure_spoil_points": SPOIL_POINTS,
"structure_spoil_m3": 55.75,
},
"채집석": {
**base,
"collected_stone_deduction_m3": 120.0,
"collected_stone_by_ground_m3": {"soil": 40.0, "blasting_rock": 30.0},
"collected_stone_ground_unknown_m3": 5.0,
},
}
def _options(context: dict, drawing: bool) -> dict:
keys = (
"collected_stone_deduction_m3",
"collected_stone_by_ground_m3",
"collected_stone_ground_unknown_m3",
"structure_spoil_m3",
"structure_spoil_points",
)
return {
**{k: context.get(k) for k in keys},
"conversion": context.get("earthwork_conversion"),
"drawing": drawing,
}
def _browser(tmp_path: Path, calls: list[dict]) -> list:
job = tmp_path / "job.json"
job.write_text(json.dumps(calls, ensure_ascii=False), encoding="utf-8")
proc = subprocess.run( # noqa: S603 — 고정 실행 파일
["node", str(HERE / "helper_511_a_mirror.cjs"), str(job)],
cwd=ROOT,
capture_output=True,
text=True,
encoding="utf-8",
timeout=300,
)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def _old_view(result: dict, context: dict):
out = run_bundle_json(
prebuild.BUNDLE, prebuild._NPM_SCRIPT, {"haul_plan_for": result, "context": context}
)
assert isinstance(out, dict), out
return out["haul_plan"]
def _edge_results() -> dict[str, dict]:
conv = PROJECTS["483"]["conversion"]
def make(nets: list[float]) -> dict:
points, cum = [], 0.0
for index, net in enumerate(nets):
cum += net
points.append(
{
"station_id": f"s{index}",
"chainage_m": index * 20.0,
"net_volume_m3": net,
"cumulative_volume_m3": cum,
"cut_soil_m3": max(net, 0.0),
"cut_rock_m3": 0.0,
"cut_rr_m3": 0.0,
"cut_br_m3": 0.0,
"net_area_m2": net / 20.0,
"fill_m3": max(-net, 0.0),
}
)
return _result({"points": points, "conversion": conv})
return {
"측점 하나": make([0.0]),
"측점 둘": make([0.0, 50.0]),
"전부 0": make([0.0] * 6),
"전부 절토": make([0.0, 30.0, 45.0, 10.0, 80.0]),
"전부 성토": make([0.0, -30.0, -45.0, -10.0, -80.0]),
"오르내림": make([0.0, 120.0, -200.0, 90.0, -15.0, 60.0, -140.0, 70.0]),
}
def test_창구_새_길_옛_묶음_브라우저_같음(tmp_path):
cases = [
(f"{rid}·{name}", _result(payload), context)
for rid, payload in PROJECTS.items()
for name, context in _contexts().items()
] + [(f"끝·{name}", result, _contexts()["기본"]) for name, result in _edge_results().items()]
calls = [
{
"mod": "common_util/common_util_mass_haul_balance.ts",
"fn": "computeHaulPlan",
"args": [result, context.get("haul_equipment_limits"), _options(context, drawing)],
}
for _, result, context in cases
for drawing in (True, False)
]
browser = _browser(tmp_path, calls)
checked = 0
for index, (label, result, context) in enumerate(cases):
new = haul_router._run_view(result, context)
assert new is not None, label
assert new["haul_plan"] == _old_view(result, context), label
assert new["haul_plan"] == browser[2 * index], label
quantity = haul_router.run_calc_served(
["haul.view"],
{
"args": {
"haul.view": [
result,
context.get("haul_equipment_limits"),
_options(context, False),
]
}
},
)["results"]["haul.view"]
assert quantity == browser[2 * index + 1], label
checked += 1
assert checked == len(PROJECTS) * 4 + 6
assert browser[2 * len(PROJECTS) * 4] is None # 측점 하나 = 배분 없음
assert any(item and item["transfers"] for item in browser) # 빈 배분만 견준 것 아님
@pytest.mark.parametrize("name", ["기본", "구조물 잔토", "채집석"])
def test_저장_정본_새_길_옛_묶음_같음(name, monkeypatch):
monkeypatch.setattr(prebuild, "_NODE_CACHE", {})
detail = json.loads((GOLDEN / "detail.json").read_text(encoding="utf-8"))
node_input = {"detail": detail, "context": _contexts()[name]}
old = run_bundle_json(prebuild.BUNDLE, prebuild._NPM_SCRIPT, copy.deepcopy(node_input))
new = prebuild._node_output(copy.deepcopy(node_input))
assert new == old
assert new["mass_haul"]["haul_plan"] and new["mass_haul"]["haul_plan_drawing"]