# -*- coding: utf-8 -*- """511-C (요구 320-4) — 유토 배분 E2 = 늘 떠 있는 Node(TS 한 벌) = 브라우저 자리 TS 0 차. 못박는 것 - /haul-plan 창구 새 길(`_run_view` → 시험 길 `haul.view`) = 브라우저 자리 함수(`computeHaulPlan` 바로 부름) — 프로젝트 일곱 (저장 유토곡선 · DB 에서 한 번 뜬 `fixtures/haul_511_c.json`) × 문맥(기본 · 장비 경계 없음 · 구조물 잔토 · 채집석) × 그림 · 수량. - 끝 경우 — 측점 하나(배분 없음 null) · 둘 · 전부 0 · 전부 절토 · 전부 성토. - [저장] 정본 새 길(`Server_Calc_Prebuild._node_output` → 늘 떠 있는 Node 셋) = 브라우저 자리 TS(`structureAreaRows` · `extraWallRows`) · 유토곡선 끝값 · 배분 두 벌 — golden_472 · 문맥 셋 (옛 server-calc 묶음 = 511-B B-6 걷음). ./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 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 _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"] == 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, tmp_path): monkeypatch.setattr(prebuild, "_NODE_CACHE", {}) detail = json.loads((GOLDEN / "detail.json").read_text(encoding="utf-8")) node_input = {"detail": detail, "context": _contexts()[name]} new = prebuild._node_output(copy.deepcopy(node_input)) layouts = "B06_Section/B06_Section_Structure_Layouts.ts" sections = detail["cross_sections"] areas, walls = _browser( tmp_path, [ {"mod": layouts, "fn": "structureAreaRows", "args": [sections]}, {"mod": layouts, "fn": "extraWallRows", "args": [sections]}, ], ) assert new["areas"] == areas and areas assert new["extra_walls"] == walls assert new["mass_haul"]["haul_plan"] and new["mass_haul"]["haul_plan_drawing"]