Merge remote-tracking branch 'origin/sub_laptop_2' into sub_laptop_3

This commit is contained in:
2026-10-22 13:09:50 +09:00
41 changed files with 1589 additions and 83 deletions
+4 -1
View File
@@ -2,6 +2,9 @@
"version": 1,
"sites": {
"ground_sample": "both",
"corridor_mesh": "both"
"corridor_mesh": "both",
"quantity": "both",
"carry": "both",
"fixed_browser": "both"
}
}
@@ -58,6 +58,13 @@ const STUBS = {
"../A00_Common/design/design_profile_free_view": {
createFreeView: () => ({ mount: () => undefined, redraw: () => undefined }),
},
// 계산 자리 「서버」 길(511-A) — 판 both 그대로 · 이 흠과 무관
"../A00_Common/design/design_calc_sites": { calcOnServerNow: () => false },
"../A00_Common/design/design_calc_server": {
calcChecked: (_key, _id, fn, args) => fn(...args),
calcOnServerOnly: async () => undefined,
trialRequest: () => ({}),
},
};
const { loadTs } = makeLoader({
@@ -27,7 +27,16 @@ const session = memoryStorage();
globalThis.window = { sessionStorage: session, localStorage: memoryStorage() };
globalThis.fetch = async () => ({ ok: false });
const STUBS = {};
// 계산 자리 「서버」 길(511-A) — 판 both 그대로
const CALC_STUBS = {
design_calc_sites: { calcOnServerNow: () => false, calcProjectId: () => "" },
design_calc_server: {
calcChecked: (_key, _id, fn, args) => fn(...args),
calcOnServerOnly: async () => undefined,
trialRequest: () => ({}),
},
};
const STUBS = { ...CALC_STUBS };
const { load, loadTs } = makeLoader({
root: ROOT,
requireFor: (full) => (request) => {
@@ -51,6 +51,13 @@ const STUBS = {
"../A00_Common/design/design_profile_free_view": {
createFreeView: () => ({ mount: () => undefined, redraw: () => undefined }),
},
// 계산 자리 「서버」 길(511-A) — 판 both 그대로
"../A00_Common/design/design_calc_sites": { calcOnServerNow: () => false },
"../A00_Common/design/design_calc_server": {
calcChecked: (_key, _id, fn, args) => fn(...args),
calcOnServerOnly: async () => undefined,
trialRequest: () => ({}),
},
};
const { loadTs } = makeLoader({
+90
View File
@@ -0,0 +1,90 @@
/* PLAN 511-A 브라우저 길 헬퍼 — `design_calc_server.ts` 를 그 자리에서 트랜스파일 · 판(`design_calc_sites`)과
* fetch 를 가짜로 끼워 판 따라 길 고름 · 잠잠 · 끊음 · 대조 · 기록을 잼. 결과 = 표준출력 JSON 한 벌.
* 짝 = `test_511_a_calc_browser.py`. */
const path = require("path");
const { makeLoader } = require("./helper_ts_stub_loader.cjs");
const ROOT = path.join(__dirname, "..", "..");
const board = {};
const fetched = [];
let answer = (body) => ({ [body.run[0]]: body.inputs.args?.[body.run[0]]?.[0] ?? null });
globalThis.fetch = async (url, init) => {
const body = JSON.parse(init.body);
fetched.push({ url, run: body.run, at: Date.now() });
if (init.signal?.aborted) throw new DOMException("끊김", "AbortError");
return {
ok: true,
status: 200,
json: async () => ({ results: answer(body), ms: 1, wall_ms: 2 }),
};
};
const plain = (value) => JSON.parse(JSON.stringify(value));
const { loadTs } = makeLoader({
root: ROOT,
stubs: {
design_calc_sites: {
calcOnServerNow: (key) => board[key] === "server",
calcProjectId: () => "p1",
},
design_calc_node_trial: { plain },
},
});
const mod = loadTs(path.join(ROOT, "A00_Common", "design", "design_calc_server.ts"));
const tick = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const echo = (value) => value;
(async () => {
const out = {};
// 1 판 both — 브라우저 값 · 서버 부름 0 · 브라우저 ms 기록
out.both = mod.calcChecked("quantity", "cross.fit", echo, [5]);
await tick(500);
out.bothFetched = fetched.length;
out.local = globalThis.__aisloCalcServer.local["cross.fit"].length;
// 2 판 server — 브라우저 값 바로 · 잠잠 뒤 한 번(빠른 둘째 부름이 첫째를 끊음) · 같음
board.quantity = "server";
let applied = 0;
mod.calcChecked("quantity", "cross.fit", echo, [6], { apply: () => (applied += 1) });
out.server = mod.calcChecked("quantity", "cross.fit", echo, [7], { apply: () => (applied += 1) });
out.beforeQuiet = fetched.length;
await tick(500);
out.afterQuiet = fetched.length;
const calls = globalThis.__aisloCalcServer.calls;
out.same = calls[calls.length - 1].same;
out.applied = applied;
// 3 서버 값이 다름 — apply(서버 값)
answer = () => ({ "cross.fit": 99 });
let got = null;
mod.calcChecked("quantity", "cross.fit", echo, [8], { apply: (v) => (got = v) });
await tick(500);
out.diffApplied = got;
out.diffSame = calls[calls.length - 1].same;
// 4 그리기 자리 — 처음 undefined · 답 오면 redraw · 같은 입력 = fresh
answer = (body) => ({ "masshaul.series": ["s", body.inputs.args["masshaul.series"][0]] });
let redraws = 0;
const first = mod.calcFromServer("quantity", "masshaul.series", [1], () => (redraws += 1));
await tick(500);
const second = mod.calcFromServer("quantity", "masshaul.series", [1], () => (redraws += 1));
const third = mod.calcFromServer("quantity", "masshaul.series", [2], () => (redraws += 1));
out.draw = { first, second, third, redraws };
await tick(500);
// 5 비동기 자리 — outer 끊음 = undefined · 부름 0
const before = fetched.length;
const outer = new AbortController();
const pending = mod.calcOnServerOnly("quantity", "slot", mod.areasRequest([]), outer.signal);
outer.abort();
out.aborted = (await pending) === undefined && fetched.length === before;
// 6 왕복 흉내 — delayMs 만큼 늦음
globalThis.__aisloCalcServer.delayMs = 80;
answer = () => ({ "structure.layouts": { areas: [1] } });
const t0 = Date.now();
out.onlyValue = await mod.calcOnServerOnly("quantity", "slot2", mod.areasRequest([]));
out.delayed = Date.now() - t0 >= 400 + 80;
out.roundMs = calls[calls.length - 1].roundMs;
out.url = fetched[fetched.length - 1].url;
process.stdout.write(JSON.stringify(out));
})();
+10
View File
@@ -0,0 +1,10 @@
/* PLAN 511-A 거울 — R9 브라우저 자리(`carryStationDrafts`)가 부르는 꼴 그대로(읽기 = 칸 이름 → 값 표). */
import {
carryKeyedStates,
type Keyed,
type KeyedState,
type Raw,
} from "../../B06_Section/logic/B06_Section_Route_Carry";
export const carryWith = (from: Record<KeyedState, Raw>, before?: Keyed[], after?: Keyed[]) =>
carryKeyedStates((name) => from[name], before, after);
+21
View File
@@ -0,0 +1,21 @@
/* PLAN 511-A 거울 헬퍼 — 브라우저 자리가 부르는 TS 함수를 그 자리에서 트랜스파일해 Node 로 바로 부름.
* 입력 = argv[2] JSON `[{ mod: 뿌리 기준 경로, fn: export 이름, args: [...] }]` · 출력 = 표준출력 JSON 목록
* (Map · Set → 목록 = 서버 `plain` 과 같은 꼴). 짝 = `test_511_a_calc_server.py`(서버 창구 답과 견줌). */
const fs = require("fs");
const path = require("path");
const { makeLoader, memoryStorage } = require("./helper_ts_stub_loader.cjs");
const ROOT = path.join(__dirname, "..", "..");
globalThis.window = { sessionStorage: memoryStorage(), localStorage: memoryStorage() };
const { loadTs } = makeLoader({ root: ROOT });
const plain = (value) =>
JSON.parse(
JSON.stringify(value, (_key, item) =>
item instanceof Map ? [...item.entries()] : item instanceof Set ? [...item] : item,
) ?? "null",
);
const job = JSON.parse(fs.readFileSync(process.argv[2], "utf8"));
const out = job.map(({ mod, fn, args }) => plain(loadTs(path.join(ROOT, mod))[fn](...args)));
process.stdout.write(JSON.stringify(out));
+6 -1
View File
@@ -172,7 +172,12 @@ def test_배선():
panel = _read("B05_Profile/B05_Profile_UI_Profile_Panel_Tools.ts")
assert "tools.handleStationPick(station ?? chainageM)" in panel
assert "straightSubRun(alignment, fromM, toM)" in panel
assert "straightenBetween(base, ctx.edits(), fromM, toM, ctx.savedRadii())" in panel
# 511-A — 인자 한 벌(계산 자리 길이 함수째 받음)
args = (
"Parameters<typeof straightenBetween> = [\n base,\n ctx.edits(),\n"
" fromM,\n toM,\n ctx.savedRadii(),"
)
assert args in panel
assert "autoRadii: ctx.savedRadii" in panel
assert 'window.addEventListener("keydown", onEscape, true)' in panel
assert "createRangeBandView(" in panel
+6 -1
View File
@@ -77,7 +77,12 @@ def test_끄는_중_미리보기_놓을_때_같음(got) -> None:
def test_편집기_이음() -> None:
edit = (B05 / "B05_Profile_UI_RouteEdit.ts").read_text(encoding="utf-8")
assert "fit.fitAfter(before, snapshotNow(), fitLimits())" in edit
# 511-A 계산 자리 대조(같은 인자)
call = (
'"route.station_fit", fit.fitAfter, [\n'
" before,\n snapshotNow(),\n fitLimits(),"
)
assert call in edit
assert "fit.fitDrag(snapshotNow(), base.planned, index, fitLimits(), base)" in edit
assert "stationM: stationIntervalM" in edit
assert len(edit.splitlines()) <= 695
+15 -8
View File
@@ -58,27 +58,33 @@ def test_묶음_여덟_줄_등록부_전부():
nos = [no for group in calc_sites.GROUPS for no in group["nos"]]
assert len(calc_sites.GROUPS) == 8
assert len(nos) == len(set(nos)) and set(nos) == set(sites)
assert calc_sites.SWITCHABLE == {"ground_sample", "corridor_mesh"}
assert calc_sites.SWITCHABLE == {
"ground_sample",
"corridor_mesh",
"quantity",
"carry",
"fixed_browser",
} # 511-A 셋 더
for group in calc_sites.GROUPS:
got = {sites[no] for no in group["nos"]}
if group["switchable"] or group["fixed"] == "both":
assert got == {"browser_now"}, group["key"]
if group["fixed"] == "browser":
assert got == {"browser_only"}, group["key"]
elif group["fixed"] == "server":
assert got <= {"server_canon", "server_only"}, group["key"]
else:
assert got == {"browser_only"}, group["key"]
assert got == {"browser_now"}, group["key"]
by_key = {g["key"]: g["nos"] for g in calc_sites.GROUPS}
assert by_key["ground_sample"] == ["G1"] and by_key["corridor_mesh"] == ["D3"]
def test_커밋한_판_기본값_없음_깨짐(tmp_path, monkeypatch):
committed = json.loads(calc_sites.PATH.read_text(encoding="utf-8"))
assert committed["sites"] == {"ground_sample": "both", "corridor_mesh": "both"}
assert committed["sites"] == dict.fromkeys(calc_sites.SWITCHABLE, "both")
assert set(calc_sites.read_sites()["sites"].values()) == {"both"}
monkeypatch.setattr(calc_sites, "PATH", tmp_path / "none.json")
assert calc_sites.read_sites() == {
"version": 0,
"sites": {"ground_sample": "both", "corridor_mesh": "both"},
"sites": dict.fromkeys(calc_sites.SWITCHABLE, "both"),
}
broken = tmp_path / "broken.json"
broken.write_text('{"sites": {"ground_sample": "server", "corridor_mesh": "x"', "utf-8")
@@ -157,7 +163,8 @@ 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"))
assert saved == {"version": 2, "sites": {"ground_sample": "server", "corridor_mesh": "both"}}
sites = {**dict.fromkeys(calc_sites.SWITCHABLE, "both"), "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
assert inserts[0][:5] == (7, "CALC_SITE_CHANGE", "calc_site", None, "ground_sample=server")
@@ -237,7 +244,7 @@ def test_연결_자리_cross_격자_그대로():
assert body.index('if (await calcOnServer("corridor_mesh")) return null;') < body.index(
"getWorker()"
)
assert "void openCalcSites();" in _text("A00_Common/design/design_page.ts")
assert "void openCalcSites(projectId);" in _text("A00_Common/design/design_page.ts")
# 저장 길은 판을 안 읽음
for rel in ("A00_Common/design/design_save.ts", "A00_Common/design/design_save_api.ts"):
assert "calc_sites" not in _text(rel) and "calcOnServer" not in _text(rel)
@@ -151,7 +151,9 @@ def test_시스템_관리자_카드_창_G1_서버_길(browser):
assert len(heads) == 8 and len(calcs) == 45
assert heads[0]["boxes"] == [[True, False], [True, True]] # G1 묶음 머리 = 바꿀 수 있음
editable = [row["name"] for row in calcs if not row["boxes"][0][1]]
assert [name.split(" · ")[0] for name in editable] == ["G1", "D3"] # 줄도 둘만 켬
opened = ["G1", "D3", "E1", "F7", "F8", "C4", "R9", "F10"] # 511-A ③ ④ 더
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()
@@ -132,7 +132,7 @@ def test_노선_반영_단추_없앰_배선():
def test_넘김_알림은_결과_한_줄에_합침():
"""346-3 덧 — [저장] 뒤 「노선 밖 고침」 두 줄 → 한 줄 · 넘김은 조용히 수만 · 기본은 알림 그대로."""
keys = _read("B06_Section/B06_Section_Station_Keys.ts")
assert "{ quiet = false }: { quiet?: boolean } = {}," in keys
assert "{ quiet = false, applyTo = toRouteId }: { quiet?: boolean; applyTo?: number }" in keys
assert "if (outside.size && !quiet)" in keys
patch = _read("B06_Section/B06_Section_Route_Carry_Patch.ts")
assert "outside: keyed.outside.size + chain.outside.size" in patch
@@ -48,6 +48,8 @@ def test_기준_횡단만_계산(r):
def test_그리기는_횡단만_부름():
mass = (ROOT / "B05_Profile/B05_Profile_UI_Profile_MassHaul.ts").read_text(encoding="utf-8")
calls = re.findall(r"computeMassHaulSeries\(([^;]*?)\)", mass, re.S)
# 511-A — 부름 인자 한 벌(계산 자리 길이 함수째 받음)
calls += re.findall(r"Parameters<typeof computeMassHaulSeries> = \[([^;]*?)\];", mass, re.S)
assert calls and all('["cross"]' in call for call in calls)
+6 -1
View File
@@ -89,7 +89,12 @@ def test_시작_자리는_고친_꺾임점_앞(got) -> None:
def test_편집기_이음() -> None:
edit = (B05 / "B05_Profile_UI_RouteEdit.ts").read_text("utf-8")
assert "fit.fitAfter(before, snapshotNow(), fitLimits())" in edit # 280-4
# 280-4 · 511-A 계산 자리 대조(같은 인자)
call = (
'"route.station_fit", fit.fitAfter, [\n'
" before,\n snapshotNow(),\n fitLimits(),"
)
assert call in edit
# 되돌리기 · 추천 [적용] · 초안 되살림이 지나는 restoreSnapshot 은 안 맞춤
at = edit.index("function restoreSnapshot(")
assert "applyEdit(message, record, false);" in edit[at : at + 700]
@@ -0,0 +1,94 @@
"""PLAN 511-A (요구 320-4) — 브라우저 「서버」 길(`design_calc_server.ts`) · 판 따라 길 · 자리.
못박는 것
- 판 both = 브라우저 값 · 서버 부름 0(넣어도 바뀌는 것 0) · server = 잠잠 400 ms 뒤 한 번
(앞 부름 끊음).
- 대조 = 같으면 그대로 · 다르면 apply(서버 값) · 그리기 자리 = 답 오면 redraw · 같은 입력 = fresh.
- 끊긴 부름 = undefined · 왕복 흉내 delayMs · 창구 주소 = /projects/{id}/calc.
- 자리마다 판 열쇠 · [저장] · [확정] 길은 서버 길을 안 부름.
./venv/Scripts/python.exe -m pytest resources/tester/test_511_a_calc_browser.py -q
"""
from __future__ import annotations
import json
import shutil
import subprocess
from pathlib import Path
import pytest
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
def _text(rel: str) -> str:
return (ROOT / rel).read_text(encoding="utf-8")
@pytest.mark.skipif(shutil.which("node") is None, reason="node 없음")
def test_판_따라_길_고름_잠잠_대조_기록():
proc = subprocess.run( # noqa: S603 — 고정 실행 파일
["node", str(HERE / "helper_511_a_browser.cjs")],
cwd=ROOT,
capture_output=True,
text=True,
encoding="utf-8",
timeout=60,
)
assert proc.returncode == 0, proc.stderr
out = json.loads(proc.stdout)
assert out["both"] == 5 and out["bothFetched"] == 0 and out["local"] == 1
assert out["server"] == 7 and out["beforeQuiet"] == 0 and out["afterQuiet"] == 1
assert out["same"] is True and out["applied"] == 0
assert out["diffApplied"] == 99 and out["diffSame"] is False
draw = out["draw"]
assert draw["first"] == {"fresh": False} and draw["redraws"] == 1
assert draw["second"] == {"value": ["s", 1], "fresh": True}
assert draw["third"] == {"value": ["s", 1], "fresh": False} # 새 입력 = 앞 답 + 다시 물음
assert out["aborted"] is True
assert out["onlyValue"] == [1] and out["delayed"] and out["roundMs"] >= 80
assert out["url"] == "http://fake/projects/p1/calc" # API_BASE_URL(가짜) + 창구
def test_자리마다_판_열쇠_저장_길_안_부름():
refresh = _text("B06_Section/B06_Section_Cross_Refresh.ts")
assert 'calcOnServerNow("quantity")' in refresh and '"F7.saved"' in refresh
shared = _text("A00_Common/design/design_shared.ts")
assert 'calcChecked("quantity", "structure.layouts"' in shared
assert 'calcChecked("quantity", "cross.berm_spans"' in shared
lane = _text("A00_Common/design/design_facility_lane.ts")
assert '"C4.copy", bermRequest(structures)' in lane
mass = _text("B05_Profile/B05_Profile_UI_Profile_MassHaul.ts")
assert 'calcFromServer<MassHaulSeries[]>("quantity", "masshaul.series"' in mass
assert "openCalcSites(projectId)" in _text("A00_Common/design/design_page.ts")
for rel in (
"A00_Common/design/design_save.ts",
"A00_Common/design/design_save_api.ts",
"A00_Common/design/design_reset.ts",
):
assert "design_calc_server" not in _text(rel), rel
page = _text("A00_Common/design/design_pipeline_page.ts")
assert len(page.splitlines()) <= 699 and "design_calc_server" not in page # 연결 = 다른 파일
def test_이월_R9_자리_판_열쇠():
keys = _text("B06_Section/B06_Section_Station_Keys.ts")
assert '"carry",\n "stations.carry_keyed",' in keys
assert "carryStationDrafts(projectId, routeId, NEXT_ROUTE, { applyTo: routeId })" in keys
def test_고정_브라우저_자리_판_열쇠():
sites = {
"B05_Profile/B05_Profile_UI_RouteEdit_Pointer.ts": '"fixed_browser", "route.handle"',
"B05_Profile/B05_Profile_UI_RouteEdit.ts": '"fixed_browser", "route.station_fit"',
"B06_Section/B06_Section_Section_Store.ts": '"fixed_browser", "stations.swap"',
"B05_Profile/B05_Profile_UI_Profile_Panel_Tools.ts": '"fixed_browser",\n "P7",',
"B05_Profile/B05_Profile_UI_RouteEdit_Cross_Draw.ts": '"fixed_browser", "cross.fit"',
"B05_Profile/B05_Profile_Footprint_Server.ts": '"fixed_browser", "D4"',
}
for rel, needle in sites.items():
assert needle in _text(rel), rel
page = _text("A00_Common/design/design_pipeline_page.ts")
assert "footprint: footprintStageSited," in page
+213
View File
@@ -0,0 +1,213 @@
"""PLAN 511-A (요구 320-4) — 묶음마다 서버 창구 답 = 브라우저 자리 TS 답(거울).
브라우저 쪽 = `helper_511_a_mirror.cjs` 가 그 자리 함수를 바로 부름 · 서버 쪽 = 늘 떠 있는 Node
(`run_calc_served` · 창구와 같은 본문 꼴 — `design_calc_server.ts` `areasRequest` · `bermRequest` ·
시험 길 args).
고정 자료 = golden_472 · 272 소단 사례 · 손 자료(공유 DB · 실제 마스터 안 씀).
./venv/Scripts/python.exe -m pytest resources/tester/test_511_a_calc_mirror.py -q
"""
from __future__ import annotations
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))
if str(HERE) not in sys.path:
sys.path.insert(0, str(HERE))
from common_util import common_util_calc_runner as runner # noqa: E402
FIXTURE = HERE / "fixtures" / "golden_472"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not (ROOT / "config" / "node_modules" / "vite").is_dir(),
reason="node · 프론트엔드 의존성 없음",
)
def _fixture(name: str):
return json.loads((FIXTURE / name).read_text(encoding="utf-8"))
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=120,
)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def _served(run: list[str], inputs: dict) -> dict:
answer = runner.run_calc_served(run, inputs)
assert answer and "results" in answer, answer
return answer["results"]
def _trial(id_: str, args: list):
return _served([id_], {"args": {id_: args}})[id_]
# ── ③ 토량 · 구조물 셈(E1 · F7 · C4 · F8 = 브라우저 부름 없음) ───────────────────────
def test_quantity_F7_C4_E1_서버_브라우저_같음(tmp_path):
from test_272_berm_source import STRUCTURE_CASES
detail = _fixture("detail.json")
sections = detail["cross_sections"]
conv = _fixture("context.json")["earthwork_conversion"]
long_ = detail["longitudinal"]
calls = [
{"mod": "B06_Section/B06_Section_Structure_Layouts.ts", "fn": "structureAreaRows",
"args": [sections]},
{"mod": "common_util/common_util_mass_haul.ts", "fn": "computeMassHaulSeries",
"args": [long_, sections, conv, None, ["cross"]]},
] + [
{"mod": "common_util/common_util_berm_spans.ts", "fn": "bermSpansFromStructures",
"args": [case]}
for case in STRUCTURE_CASES
] # fmt: skip
browser = _browser(tmp_path, calls)
areas = _served(["structure.layouts"], {"detail": {"cross_sections": sections}})
assert areas["structure.layouts"]["areas"] == browser[0] and browser[0]
series = _trial("masshaul.series", [long_, sections, conv, None, ["cross"]])
assert series == browser[1] and series[0]["result"]["final_cumulative_m3"]
for index, case in enumerate(STRUCTURE_CASES):
got = _served(["cross.berm_spans"], {"berm": {"structures": case, "chainages": []}})
assert got["cross.berm_spans"]["spans"] == browser[2 + index], index
# ── ④ 이월 · 다시 얹기(R9 · F10 = 브라우저 부름 없음) ──────────────────────────────────
def test_carry_R9_서버_브라우저_같음(tmp_path):
seat = {"x": 10.0, "y": 5.0, "left": [0.0, 1.0]}
states = {
"crossdesign": {"grid:1": {"ditch_type": "U"}, "20.00": {"paved": True}},
"shoulderext": {"grid:2": 0.5, "@at": {"grid:2": 40.0}, "@seat": {"grid:2": seat}},
"bermdepth": {"at:55.00": 1.0},
"rockb": {"30.00": -2.5},
"cutslope": {},
}
before = [
{"key": "grid:1", "chainageM": 20.0, "seat": seat},
{"key": "grid:2", "chainageM": 40.0},
{"key": "end:EP", "chainageM": 60.0},
]
after = [
{"key": "grid:1", "chainageM": 22.0},
{"key": "grid:2", "chainageM": 41.0, "seat": seat},
{"key": "end:EP", "chainageM": 58.0},
]
empty = dict.fromkeys(states, {}) # 브라우저 읽기 = 빈 칸도 {}(readRaw)
cases = [[states, before, after], [states, before, None], [empty, None, after]]
browser = _browser(
tmp_path,
[
{"mod": "resources/tester/helper_511_a_carry.ts", "fn": "carryWith", "args": case}
for case in cases
],
)
for index, case in enumerate(cases):
assert _trial("stations.carry_keyed", case) == browser[index], index
assert browser[0]["states"]["crossdesign"] # 넘김이 실제로 일어남
# ── ⑧ 고정 브라우저(시험용 · R2 R4 C7 대조 · P7 서버 값 · R8 · D4 = 화면 검증) ──────────────────
_ROUTE = [[0, 0], [93, 7], [171, 61], [262, 48], [337, 112], [431, 101], [503, 160]]
_LIMITS = {"minRadiusM": 30, "limitRadiusM": 12, "limitArcM": 0, "stationM": 20, "intervalM": 10}
def _columns(planned):
n = len(planned)
return {
"planned": [list(p) for p in planned],
"curveOn": [True] * n,
"curveRadius": [None] * n,
"curveLock": [None] * n,
"curveArc": [None] * n,
"apexLock": [i in (0, n - 1) for i in range(n)],
}
def _profile_base():
import math
stations = [i * 20 for i in range(16)]
ground = [round(100 + 0.03 * s + 0.6 * math.sin(s / 13), 6) for s in stations]
policy = {
"station_interval_m": 20, "curve_radius_ratio": 1, "curve_tangent_max_ratio": 0.5,
"curve_skip_legal_exception": False, "default_curve_length_m": 20,
"default_curve_radius_m": 300, "balance_tolerance_percent": 10, "edit_step_m": 0.1,
"grade_violation_policy": "warn", "max_grade_pct": 14, "curve_skip_delta_pct": 0,
"paved": True, "min_vertical_radius_m": 100, "min_curve_length_m": 20,
} # fmt: skip
return {
"policy": policy,
"onlyExplicitCurves": False,
"basePvi": [{"chainage_m": s, "elevation_m": z} for s, z in zip(stations, ground)],
"chainage": stations,
"ground": ground,
"stations": [{"station_id": f"S{s}", "chainage_m": s} for s in stations],
}
def test_fixed_browser_R2_R4_P7_C7_서버_브라우저_같음(tmp_path):
pushed = [list(p) for p in _ROUTE]
pushed[2] = [174.0, 58.0]
base = _profile_base()
empty = {"station_offsets": {}, "curve_radii": {}}
saved = {"40.000": 5330, "200.000": 1084}
sections = _fixture("detail.json")["cross_sections"][:12]
cases = {
"route.handle": [
[[0, 0], [100, 0], [100, 100], "end", [105, 60]],
[[0, 0], [100, 0], [100, 100], "start", [60, -4]],
[[0, 0], [100, 0], [200, 0], "end", [150, 0]], # 직선 = null
],
"route.station_fit": [
[_columns(_ROUTE), _columns(pushed), _LIMITS],
[None, _columns(_ROUTE), {**_LIMITS, "intervalM": 20}],
],
"profile.tools": [
[base, empty, 40, 200, {}],
[base, {"station_offsets": {}, "curve_radii": dict(saved)}, 40, 200, saved],
[base, empty, 0, 300, {}],
],
"cross.fit": [[section] for section in sections],
}
b05 = "B05_Profile/logic/B05_Profile_"
where = {
"route.handle": (f"{b05}UI_RouteEdit_Curve.ts", "dragHandleTo"),
"route.station_fit": (f"{b05}RouteEdit_StationFit.ts", "fitAfter"),
"profile.tools": (f"{b05}UI_Profile_Straighten.ts", "straightenBetween"),
"cross.fit": ("B06_Section/logic/B06_Section_UI_Cross_Fit.ts", "toeOffsets"),
}
flat = [(id_, args) for id_, rows in cases.items() for args in rows]
browser = _browser(
tmp_path,
[{"mod": where[id_][0], "fn": where[id_][1], "args": args} for id_, args in flat],
)
for (id_, args), want in zip(flat, browser):
assert _trial(id_, args) == want, id_
assert browser[0] and browser[2] is None # 직선 = null
toes = [want for (id_, _), want in zip(flat, browser) if id_ == "cross.fit"]
assert any(toe["left"] is not None for toe in toes)
+139
View File
@@ -0,0 +1,139 @@
"""PLAN 511-A (요구 320-4) — 계산 자리 「서버」 창구 `POST /api/projects/{id}/calc` · 떠 있는 Node.
못박는 것
- 창구 = 프로젝트 권한(다른 회사 403 · 시스템 관리자 통과) · 본문 상한 413 · 모르는 id · 꼴 400.
- 늘 떠 있는 Node 답 = 파일 길(`run_calc`) 답 · 죽여도 다음 부름에 다시 띄움.
- 시험 길 id(`TRIAL_IDS`) = TS `design_calc_node_trial.ts` 표 · `available` 에 안 섞임.
- 공유 DB 안 씀(가짜 풀) · 실제 마스터 안 씀(고정 자료 golden_472).
./venv/Scripts/python.exe -m pytest resources/tester/test_511_a_calc_server.py -q
"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from B06_Section import B06_Section_Router_Calc as calc_router # noqa: E402
from common_util import common_util_auth as auth # noqa: E402
from common_util import common_util_calc_runner as runner # noqa: E402
FIXTURE = ROOT / "resources" / "tester" / "fixtures" / "golden_472"
TRIAL_TS = ROOT / "A00_Common" / "design" / "logic" / "design_calc_node_trial.ts"
PROJECT = "11111111-2222-3333-4444-555555555555"
pytestmark = pytest.mark.skipif(
not (ROOT / "config" / "node_modules" / "vite").is_dir(), reason="프론트엔드 의존성 미설치"
)
def _fixture(name: str):
return json.loads((FIXTURE / name).read_text(encoding="utf-8"))
def _inputs() -> dict:
ctx = _fixture("context.json")
keys = ("earthwork_conversion", "haul_equipment_limits", "natural_spoil_min_ground_slope")
return {"detail": _fixture("detail.json"), "context": {k: ctx[k] for k in keys}}
class _Cursor:
async def execute(self, sql, args=None):
self.args = args
async def fetchone(self):
return (1,) if self.args == (PROJECT,) else None # 회사 1 프로젝트 하나
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
class _Conn(_Cursor):
def cursor(self):
return _Cursor()
class _Pool:
def acquire(self):
return _Conn()
@pytest.fixture
def client(monkeypatch):
monkeypatch.setattr(auth, "get_db_pool", lambda: _Pool())
app = FastAPI()
app.include_router(
calc_router.router,
dependencies=[Depends(auth.verify_session), Depends(auth.require_project_access)],
)
session = {"role": "USER", "company_id": 1, "user_id": 7}
app.dependency_overrides[auth.verify_session] = lambda: dict(session)
return TestClient(app), session
def test_main_프로젝트_권한으로_붙음():
main = (ROOT / "main.py").read_text(encoding="utf-8")
assert "app.include_router(b06_calc_router, dependencies=protected_with_company)" in main
assert "note_api_call(session_id)" in main # 호출량 = 미들웨어(/api 전부)
def test_권한_다른_회사_403_관리자_통과(client):
http, session = client
body = {"run": ["cross.fit"], "inputs": {"args": {"cross.fit": [{}]}}}
ok = http.post(f"/api/projects/{PROJECT}/calc", json=body)
assert ok.status_code == 200, ok.text
assert ok.json()["results"] == {"cross.fit": {"left": None, "right": None}}
session["company_id"] = 2
assert http.post(f"/api/projects/{PROJECT}/calc", json=body).status_code == 403
missing = "00000000-0000-0000-0000-000000000000"
assert http.post(f"/api/projects/{missing}/calc", json=body).status_code == 404
session["role"] = "SYSTEM_ADMIN"
assert http.post(f"/api/projects/{PROJECT}/calc", json=body).status_code == 200
def test_본문_상한_꼴_모르는_id(client, monkeypatch):
http, _ = client
url = f"/api/projects/{PROJECT}/calc"
monkeypatch.setattr(calc_router, "MAX_BODY", 100)
big = {"run": ["cross.fit"], "inputs": {"pad": "x" * 200}}
assert http.post(url, json=big).status_code == 413
monkeypatch.setattr(calc_router, "MAX_BODY", 8 * 1024 * 1024)
assert http.post(url, content=b"{nope").status_code == 400
assert http.post(url, json={"run": "cross.fit"}).status_code == 400
assert http.post(url, json={"run": ["cross.fit"], "inputs": []}).status_code == 400
bad = http.post(url, json={"run": ["stations.list"], "inputs": {}})
assert bad.status_code == 400 and "stations.list" in bad.json()["message"]
def test_떠_있는_Node_답_파일_길과_같음_죽어도_다시():
ids = ["structure.layouts", "masshaul.cumulative", "haul.plan", "excavation.trench"]
want = runner.run_calc(ids, _inputs())
got = runner.run_calc_served(ids, _inputs())
assert got["results"] == want and want["masshaul.cumulative"]["final_cumulative_m3"]
assert got["ms"] >= 0 and got["wall_ms"] >= got["ms"]
runner._SERVED.proc.kill()
runner._SERVED.proc.wait()
again = runner.run_calc_served(["structure.layouts"], _inputs())
assert again["results"]["structure.layouts"] == want["structure.layouts"]
def test_시험_길_id_TS_표와_같음_available_밖():
src = TRIAL_TS.read_text(encoding="utf-8")
table = src.split("const TRIAL: Record<string, Call> = {")[1].split("\n};")[0]
assert set(re.findall(r'^ "([a-z_.]+)":', table, re.M)) == runner.TRIAL_IDS
out = runner.run_bundle_json(runner.BUNDLE, runner.NPM_SCRIPT, {"run": [], "inputs": {}})
assert set(out["available"]) == runner.NODE_IDS
assert not runner.NODE_IDS & runner.TRIAL_IDS
@@ -0,0 +1,235 @@
"""PLAN 511-A 화면 — 자기 포트 Playwright 머리 없는 브라우저 · 서버가 떠 있을 때만 · 검증_sub2.
- 판 기본값(both) = B05 계획고 ▲ · 교각점 끌기 동안 /calc 0 · 브라우저 셈 기록만
- 묶음 하나씩(quantity · carry · fixed_browser) server → 같은 조작 → /calc 요청 · 답마다
그 페이지 TS 로 다시 셈한 값 = 서버 값 · 그 묶음 id 만 감 → 판 기본값으로(파일 바이트 그대로)
- 저장 안 함(세션 초안만 · 창 닫으면 버림) · 시험 브라우저 표식 시스템 로그 행만 지움
- 시간 표 = `tmp/511a_times.json`(묶음마다 브라우저 ms · 왕복 ms · Node ms · 흉내 0 · 30 · 80 ms)
./venv/Scripts/python.exe -m pytest resources/tester/test_511_a_calc_sites_browser.py -q -s
"""
from __future__ import annotations
import asyncio
import json
import statistics
import sys
import urllib.request
from pathlib import Path
import pytest
HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
if str(HERE) not in sys.path:
sys.path.insert(0, str(HERE))
from helper_verify_project import own_ports # noqa: E402
from test_505_3_cross_card_screen import JS_NODES # noqa: E402
API, BASE = own_ports()
BOARD = ROOT / "resources" / "master_template" / "calc_sites.json"
PROJECT = "cb15892a-5788-4a26-9316-20a15421f68b" # 검증_sub2(OWNERS.md)
MARK = "AisloTest-511-A"
TIMES = ROOT / "tmp" / "511a_times.json"
GROUP_IDS = {
"quantity": {"structure.layouts", "cross.berm_spans", "masshaul.series"},
"carry": {"stations.carry_keyed"},
"fixed_browser": {"route.handle", "route.station_fit", "stations.swap", "profile.tools"}
| {"cross.fit", "footprint.plan"},
}
def _started(base: str) -> str | None:
try:
with urllib.request.urlopen(f"{base}/api/health", timeout=3) as res:
return json.loads(res.read()).get("started_at")
except Exception:
return None
_UP = _started(BASE) if BASE else None
pytestmark = pytest.mark.skipif(
_UP is None or _UP != _started(API), reason="이 폴더 서버 없음 · vite 가 이 폴더 백엔드로 안 감"
)
JS_GRADE = r"""async () => {
const url = performance.getEntriesByType('resource').map((e) => e.name)
.find((n) => /design_pipeline_page\.ts/.test(n));
const m = await import(url);
const pid = localStorage.getItem('frd_current_project_id');
const pipe = m.pagePipeline(pid);
const cur = pipe.input('alignment') ?? { station_offsets: {}, curve_radii: {} };
const st = pipe.output('ground').stations;
const key = st[Math.floor(st.length / 2)].chainageM.toFixed(3);
m.pipelineGrade(pid, { ...cur, station_offsets: { ...cur.station_offsets,
[key]: (cur.station_offsets?.[key] ?? 0) + 0.3 } });
}"""
# 잡은 /calc 하나를 이 페이지의 같은 TS 로 다시 셈 → 같은지
JS_MIRROR = r"""async (calls) => {
const urls = performance.getEntriesByType('resource').map((e) => e.name);
const mod = async (re) => {
const u = urls.find((n) => re.test(n));
if (!u) throw new Error('모듈 없음 ' + re);
return import(u);
};
const plain = (v) => JSON.parse(JSON.stringify(v, (_k, x) =>
x instanceof Map ? [...x.entries()] : x instanceof Set ? [...x] : x) ?? 'null');
const where = {
'structure.layouts': [/B06_Section_Structure_Layouts\.ts/, 'structureAreaRows'],
'cross.berm_spans': [/common_util_berm_spans\.ts/, 'bermSpansFromStructures'],
'masshaul.series': [/common_util_mass_haul\.ts/, 'computeMassHaulSeries'],
'route.handle': [/B05_Profile_UI_RouteEdit_Curve\.ts/, 'dragHandleTo'],
'route.station_fit': [/B05_Profile_RouteEdit_StationFit\.ts/, 'fitAfter'],
'stations.swap': [/B06_Section_Station_Swap\.ts/, 'swapCrossSections'],
'profile.tools': [/B05_Profile_UI_Profile_Straighten\.ts/, 'straightenBetween'],
'cross.fit': [/B06_Section_UI_Cross_Fit\.ts/, 'toeOffsets'],
'footprint.plan': [/B05_Profile_Footprint_Stage\.ts/, 'footprintStage'],
'stations.carry_keyed': [/B06_Section_Route_Carry\.ts/, 'carryKeyedStates'],
};
const out = [];
for (const { run, inputs, results } of calls) {
for (const id of run) {
const fn = (await mod(where[id][0]))[where[id][1]];
const args = inputs.args?.[id];
const local = id === 'structure.layouts' ? fn(inputs.detail.cross_sections)
: id === 'cross.berm_spans' ? fn(inputs.berm.structures)
: id === 'footprint.plan' ? fn({ out: args[0], range: null, prev: null })
: id === 'stations.carry_keyed' ? fn((n) => args[0][n] ?? {}, args[1], args[2])
: fn(...args);
const theirs = id === 'structure.layouts' ? results[id].areas
: id === 'cross.berm_spans' ? results[id].spans : results[id];
out.push({ id, same: JSON.stringify(plain(local)) === JSON.stringify(theirs) });
}
}
return out;
}"""
def _drop_test_logs() -> None:
from config.config_db import close_db_pool, init_db_pool
async def run():
pool = await init_db_pool()
try:
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute("DELETE FROM system_audit_logs WHERE user_agent = %s", (MARK,))
await connection.commit()
finally:
await close_db_pool()
asyncio.run(run())
@pytest.fixture(scope="module")
def ctx():
sync = pytest.importorskip("playwright.sync_api")
committed = BOARD.read_bytes()
cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
try:
with sync.sync_playwright() as play:
chrome = play.chromium.launch(channel="msedge", headless=True)
context = chrome.new_context(viewport={"width": 1600, "height": 1000}, user_agent=MARK)
login = {k: cred[k] for k in ("email", "password")}
r = context.request.post(f"{BASE}/api/auth/login/request", data=login, timeout=60000)
assert r.ok, f"로그인 실패 {r.status}" # 비밀번호는 안 찍음
try:
yield context
finally:
for key in GROUP_IDS:
_put(context, key, "both")
chrome.close()
finally:
BOARD.write_bytes(committed)
_drop_test_logs()
def _put(context, key: str, value: str) -> None:
r = context.request.put(f"{BASE}/api/dashboard/admin/calc-sites", data={"sites": {key: value}})
assert r.ok, r.status
def _session(context, delay: int = 0):
page = context.new_page()
page.add_init_script(
f"localStorage.setItem('frd_current_project_id', '{PROJECT}');"
" performance.setResourceTimingBufferSize(20000);"
)
calls: list[dict] = []
def on_response(res):
if res.url.endswith("/calc") and res.request.method == "POST":
calls.append({**json.loads(res.request.post_data), "results": res.json()["results"]})
page.on("response", on_response)
page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
page.wait_for_timeout(9000)
seed = "{ calls: [], local: {}, delayMs: 0 }"
page.evaluate(f"(d) => {{ (window.__aisloCalcServer ??= {seed}).delayMs = d; }}", delay)
return page, calls
def _act(page) -> dict:
page.evaluate(JS_GRADE) # 계획고 ▲ 한 번(가운데 측점 +0.3 m)
page.wait_for_timeout(3500)
nodes = page.evaluate(JS_NODES)
assert len(nodes) >= 3, "지도에서 교각점을 못 찾음"
node = nodes[len(nodes) // 2]
page.mouse.move(node["x"], node["y"])
page.mouse.down()
page.mouse.move(node["x"] + 14, node["y"] + 9, steps=10) # 교각점 끌기(노선 초안)
page.mouse.up()
page.wait_for_timeout(5000)
return page.evaluate("() => window.__aisloCalcServer")
def _median(values: list[float]) -> float | None:
values = [v for v in values if v is not None]
return round(statistics.median(values), 2) if values else None
def test_기본값_서버_부름_0(ctx):
page, calls = _session(ctx)
log = _act(page)
page.close()
assert calls == [] and log["calls"] == []
assert {"structure.layouts", "masshaul.series", "footprint.plan"} <= set(log["local"])
assert {"route.station_fit", "stations.swap", "stations.carry_keyed"} <= set(log["local"])
TIMES.write_text(json.dumps({"browser": log["local"]}, ensure_ascii=False), encoding="utf-8")
@pytest.mark.parametrize("group", list(GROUP_IDS))
def test_묶음_서버_요청_값_같음_시간(ctx, group):
_put(ctx, group, "server")
table = json.loads(TIMES.read_text(encoding="utf-8")) if TIMES.exists() else {}
try:
for delay in (0, 30, 80):
page, calls = _session(ctx, delay)
log = _act(page)
mirror = page.evaluate(JS_MIRROR, calls)
page.close()
sent = {id_ for call in calls for id_ in call["run"]}
assert sent and sent <= GROUP_IDS[group], (group, sent)
assert mirror and all(row["same"] for row in mirror), (group, mirror)
assert all(c["same"] is not False for c in log["calls"]), log["calls"]
rows = {}
for call in log["calls"]:
rows.setdefault(call["id"], []).append(call)
table[f"{group}@{delay}"] = {
id_: {
"n": len(group_calls),
"round_ms": _median([c["roundMs"] for c in group_calls]),
"node_ms": _median([c["nodeMs"] for c in group_calls]),
"wall_ms": _median([c["wallMs"] for c in group_calls]),
"kb": round(_median([c["bytes"] for c in group_calls]) / 1024),
}
for id_, group_calls in rows.items()
}
print(group, delay, table[f"{group}@{delay}"])
finally:
_put(ctx, group, "both")
TIMES.write_text(json.dumps(table, ensure_ascii=False, indent=1), encoding="utf-8")
assert json.loads(BOARD.read_text("utf-8"))["sites"][group] == "both"
@@ -77,7 +77,7 @@ def test_cross_refresh_lays_structure_areas_once():
유토 뱃지가 갈렸다(실측 B05 −14,144.5 · B06 −12,533.2㎥)."""
body = REFRESH[REFRESH.index("export async function refreshCrossDesigns") :]
body = body[: body.index("\n}\n")]
assert "applyStructureAreaRows(" in body and "structureAreaRows(" in body
assert "applyStructureAreaRows(" in body and "structureAreaRows" in body # 511-A 함수째 넘김
assert "applyStructureAreaRows(" not in B05_PREVIEW
# 화면(횡단 보기 · 카드 그리기)은 면적을 안 고침 — 카드 안 그려도 면적 같음(387-4)
assert "applyStructureAreaRows(" not in B06_PAGE