eomsangdonandClaude Opus 5.5
55f1a7c5cc
- /calc 창구 차례 = 격자 경로 막기 → 단계 · 화면 길 서버 맥락(B-4) → follow 저장 정본 맥락(B-5 · stages 다시 짬) - 까닭 글 = cross 그대로 · pair_B5 걷음(follow 엶) - 시험 511_b1 · 309_2 · 309_2_browser = 잠긴 단계 줄 없음 · 잠금 예시 = form_default · 화면 열린 줄에 P5 P6 F1 F2 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
277 lines
12 KiB
Python
277 lines
12 KiB
Python
"""PLAN 504-2 (요구 309-2 · 310) — 계산 자리 판 · 시스템 관리자 창구 · G1 · D3 연결.
|
|
|
|
못박는 것
|
|
- 묶음 12 줄 = 등록부 # 전부(45 · 겹침 없음) · 바꿀 수 있는 줄(G1 · D3) = browser_now · 고정 줄 = 그 자리.
|
|
- 커밋한 판 = 기본값(전부 both = 등록부 그대로) · 판 없음 · 깨짐 = 기본값.
|
|
- GET = 로그인 · PUT = 시스템 관리자만(ADMIN · USER 403) · 잠금 · 모르는 줄 · 값 = 400 · 바뀐 줄마다
|
|
시스템 로그 CALC_SITE_CHANGE 한 줄 · 같은 값 = 판 · 로그 그대로.
|
|
- ground 단계 = 판(serverOnly) 따라 격자 · 서버 길 · cross 의 격자는 그대로 · D3 = 워커 안 띄움.
|
|
- 카드 = 시스템 관리자 갈래에만.
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_309_2_calc_sites.py -q
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import 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 B01_Dashboard import B01_Dashboard_Router as dash # noqa: E402
|
|
from common_util import common_util_calc_sites as calc_sites # noqa: E402
|
|
from common_util.common_util_auth import verify_session # noqa: E402
|
|
|
|
REGISTRY = (
|
|
ROOT / "A00_Common/design/logic/design_calc_registry.ts",
|
|
ROOT / "A00_Common/design/logic/design_calc_registry_facility.ts",
|
|
)
|
|
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
|
|
|
|
|
|
def _text(rel: str) -> str:
|
|
return (ROOT / rel).read_text(encoding="utf-8")
|
|
|
|
|
|
def _registry_sites() -> dict[str, str]:
|
|
pairs: dict[str, str] = {}
|
|
for path in REGISTRY:
|
|
src = path.read_text(encoding="utf-8")
|
|
for no, site in re.findall(r'no: "(\w+)",[\s\S]*?site: "(\w+)"', src):
|
|
pairs[no] = site
|
|
return pairs
|
|
|
|
|
|
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) == 13 # 511-B 단계 다섯 · F9 폼 기본값 · 511-C haul
|
|
assert len(nos) == len(set(nos)) and set(nos) == set(sites)
|
|
assert calc_sites.SWITCHABLE == {
|
|
"ground_sample",
|
|
"corridor_mesh",
|
|
"quantity",
|
|
"carry",
|
|
"route",
|
|
"follow",
|
|
"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" or group.get("default") == "server":
|
|
assert got <= {"server_canon", "server_only"}, group["key"]
|
|
else:
|
|
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 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": 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 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:
|
|
def __init__(self, log: list):
|
|
self.log = log
|
|
|
|
async def execute(self, sql, args=None):
|
|
self.log.append((sql, args))
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
return False
|
|
|
|
|
|
class _Conn(_Cursor):
|
|
def cursor(self):
|
|
return _Cursor(self.log)
|
|
|
|
async def commit(self):
|
|
self.log.append(("COMMIT", None))
|
|
|
|
|
|
class _Pool:
|
|
def __init__(self, log: list):
|
|
self.log = log
|
|
|
|
def acquire(self):
|
|
return _Conn(self.log)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
board = tmp_path / "calc_sites.json"
|
|
board.write_text(calc_sites.PATH.read_text(encoding="utf-8"), "utf-8")
|
|
monkeypatch.setattr(calc_sites, "PATH", board)
|
|
log: list = []
|
|
monkeypatch.setattr(dash, "get_db_pool", lambda: _Pool(log))
|
|
app = FastAPI()
|
|
app.include_router(dash.router)
|
|
role = {"role": "SYSTEM_ADMIN"}
|
|
app.dependency_overrides[verify_session] = lambda: {"role": role["role"], "user_id": 7}
|
|
return TestClient(app), role, board, log
|
|
|
|
|
|
def test_읽기_로그인_쓰기_시스템_관리자만(client):
|
|
http, role, board, log = client
|
|
before = board.read_text("utf-8")
|
|
for name in ("USER", "ADMIN"):
|
|
role["role"] = name
|
|
got = http.get("/api/dashboard/calc-sites")
|
|
assert got.status_code == 200
|
|
assert got.json()["sites"]["ground_sample"] == "both"
|
|
assert [g["key"] for g in got.json()["groups"]][:2] == ["ground_sample", "corridor_mesh"]
|
|
put = http.put(
|
|
"/api/dashboard/admin/calc-sites", json={"sites": {"ground_sample": "server"}}
|
|
)
|
|
assert put.status_code == 403, name
|
|
assert board.read_text("utf-8") == before and log == []
|
|
|
|
|
|
def test_잠금_모르는_줄_400_감사_한_줄(client):
|
|
http, _role, board, log = client
|
|
url = "/api/dashboard/admin/calc-sites"
|
|
for bad in ({"form_default": "server"}, {"fixed_server": "both"}, {"nope": "server"}):
|
|
assert http.put(url, json={"sites": bad}).status_code == 400, bad
|
|
assert http.put(url, json={"sites": {"ground_sample": "browser"}}).status_code == 400
|
|
assert log == []
|
|
|
|
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 = {**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
|
|
assert inserts[0][:5] == (7, "CALC_SITE_CHANGE", "calc_site", None, "ground_sample=server")
|
|
assert log[-1][0] == "COMMIT"
|
|
|
|
# 같은 값 = 판 번호 · 로그 그대로
|
|
again = http.put(url, json={"sites": {"ground_sample": "server"}})
|
|
assert again.json()["changed"] == [] and json.loads(board.read_text("utf-8"))["version"] == 2
|
|
assert len([1 for sql, _ in log if "system_audit_logs" in sql]) == 1
|
|
# 되돌림
|
|
http.put(url, json={"sites": {"ground_sample": "both"}})
|
|
assert json.loads(board.read_text("utf-8"))["sites"]["ground_sample"] == "both"
|
|
|
|
|
|
_RUNNER = """
|
|
import { createGroundStage } from "./B05_Profile/logic/B05_Profile_Pipeline_RouteGround.js";
|
|
const axis = Float64Array.from({ length: 61 }, (_, i) => i);
|
|
const grid = { x: axis, y: axis, z: new Float32Array(61 * 61).fill(5),
|
|
validBits: new Uint8Array(Math.ceil((61 * 61) / 8)).fill(255) };
|
|
const run = async (serverOnly) => {
|
|
const calls = [];
|
|
const stage = createGroundStage({
|
|
grid: () => ({ grid, design: "las" }),
|
|
remote: async (points) => { calls.push(points.length); return points.map(() => 9); },
|
|
...(serverOnly === undefined ? {} : { serverOnly }),
|
|
});
|
|
const station = { key: "k0", chainageM: 0, x: 30, y: 30, azimuthDeg: 0, left: [0, 1],
|
|
kind: "grid", id: null, ground: null };
|
|
const out = await stage({
|
|
generation: 1, range: null, signal: new AbortController().signal,
|
|
inputs: { route: null }, out: { stations: [station], route: null }, before: {}, prev: null,
|
|
});
|
|
return { calls: calls.length, center: out.stations[0].ground.centerZ };
|
|
};
|
|
console.log(JSON.stringify([
|
|
await run(undefined), await run(() => false), await run(async () => true), await run(() => true),
|
|
]));
|
|
"""
|
|
|
|
|
|
@pytest.mark.skipif(shutil.which("node") is None or not TSC.exists(), reason="node · tsc 없음")
|
|
def test_ground_판_따라_길_고름(tmp_path):
|
|
out = tmp_path / "js"
|
|
subprocess.run( # noqa: S603 — 고정 실행 파일
|
|
["node", str(TSC), str(ROOT / "B05_Profile/logic/B05_Profile_Pipeline_RouteGround.ts"),
|
|
"--outDir", str(out), "--rootDir", str(ROOT), "--module", "esnext", "--target", "es2022",
|
|
"--moduleResolution", "bundler", "--ignoreConfig"],
|
|
cwd=str(ROOT), check=True, capture_output=True,
|
|
) # fmt: skip
|
|
for emitted in out.rglob("*.js"):
|
|
emitted.write_text(
|
|
re.sub(r'(from\s+"\.\.?/[^"]+)"', r'\1.js"', emitted.read_text(encoding="utf-8")),
|
|
encoding="utf-8",
|
|
)
|
|
(out / "runner.mjs").write_text(_RUNNER, encoding="utf-8")
|
|
result = subprocess.run( # noqa: S603
|
|
["node", str(out / "runner.mjs")], capture_output=True, text=True, timeout=60
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
grid, off, server_async, server = json.loads(result.stdout)
|
|
assert grid == off == {"calls": 0, "center": 5} # 판 없음 · both = 격자
|
|
assert server_async == server and server["calls"] == 1 and server["center"] == 9
|
|
|
|
|
|
def test_연결_자리_cross_격자_그대로():
|
|
deps = _text("B05_Profile/B05_Profile_Api_TerrainGrid.ts")
|
|
assert "grid: () => designTerrainGridNow(projectId)," in deps # cross 가 같이 쓰는 격자
|
|
assert 'serverOnly: () => calcOnServer("ground_sample"),' in deps
|
|
stage = _text("B05_Profile/logic/B05_Profile_Pipeline_RouteGround.ts")
|
|
assert "const local = (await deps.serverOnly?.()) ? null : deps.grid();" in stage
|
|
page = _text("A00_Common/design/design_pipeline_page.ts")
|
|
assert len(page.splitlines()) <= 699
|
|
assert "createCrossStage({ grid: groundDeps.grid })" in page
|
|
assert "calc_sites" not in page and "calcOnServer" not in page
|
|
worker = _text("B05_Profile/B05_Profile_UI_Corridor_Worker_Client.ts")
|
|
body = worker.split("export async function buildCorridorOffThread(")[1]
|
|
assert body.index('if (await calcOnServer("corridor_mesh")) return null;') < body.index(
|
|
"getWorker()"
|
|
)
|
|
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)
|
|
|
|
|
|
def test_카드_시스템_관리자만_글_다_있음():
|
|
nav = _text("A00_Common/m_settings_nav.ts")
|
|
links = nav.split("export function mountSettingsDrawer")[0]
|
|
admin, rest = links.split('if (user.role === "SYSTEM_ADMIN") {')[1].split("// 회사 수정 =")
|
|
assert 'key: "calcSites"' in admin and "calcSites" not in rest
|
|
assert 'settingLinks(user, companyId).filter((link) => link.key !== "calcSites")' in nav
|
|
page = _text("B01_Dashboard/B01_Dashboard_UI_Page.ts")
|
|
assert 'link.key === "calcSites" ? void openCalcSitesDialog()' in page
|
|
locale = _text("ui_template/ui_template_locale_m1.ts")
|
|
for group in calc_sites.GROUPS:
|
|
assert f"B01_CalcSites_{group['key']}:" in locale, group["key"]
|
|
assert f"B01_CalcSites_Why_{group['why'] or group['key']}:" in locale, group["key"]
|
|
ui = _text("B01_Dashboard/B01_Dashboard_UI_CalcSites.ts")
|
|
assert 'check(group.fixed !== "browser", true)' in ui # 서버 정본 칸 = 늘 잠금
|
|
assert "saveCalcSites(changes)" in ui and "dirty()" in ui # 바뀐 줄만 PUT
|
|
assert "CALC_REGISTRY" in ui and "group.nos.map" in ui # 묶음 머리 아래 계산 줄(등록부 label)
|