Merge remote-tracking branches 'origin/sub_laptop_1', 'origin/sub_laptop_3' and 'origin/sub_laptop_4' into main_laptop_1

This commit is contained in:
2026-10-21 15:24:14 +09:00
12 changed files with 374 additions and 41 deletions
+6 -1
View File
@@ -124,6 +124,10 @@ export function createSpecForm(links: SpecLinks | null, hooks: SpecFormHooks = {
return slot;
};
/** 관경 빈칸 = 빈 보기 그대로 — 기본값 1000 을 몰래 채우지 않음(505-4 · 295). */
const blankKept = (c: SpecContainer, item: SpecItem): boolean =>
spec?.typeId === "pipe" && c.id === "pipe" && item.id === "diameter";
const control = (
c: SpecContainer,
item: SpecItem,
@@ -140,7 +144,7 @@ export function createSpecForm(links: SpecLinks | null, hooks: SpecFormHooks = {
const blank = !tag ? "" : name ? `${name} (${tag})` : tag;
// 필수 상자 = 빈 보기 없음(54-43) · 선택 상자 빈 보기 = 빈 글
select.replaceChildren(
...(c.required && !tag ? [] : [new Option(blank, "")]),
...(c.required && !tag && !blankKept(c, item) ? [] : [new Option(blank, "")]),
...choices.map((ch) => new Option(choiceLabel(c, item, ch, names), ch.value)),
);
// 선택지 밖 옛 값은 그 값만 남겨 보임(지운 값을 몰래 바꾸지 않음)
@@ -223,6 +227,7 @@ export function createSpecForm(links: SpecLinks | null, hooks: SpecFormHooks = {
for (const item of c.items) {
if (item.kind !== "select" || values[item.id] || (item.by && !branchOf(item, values)))
continue;
if (blankKept(c, item)) continue;
const choices = choicesOf(item, values);
const key = branchOf(item, values)?.key;
const wanted = String((key ? defaults[key] : undefined) ?? item.default ?? "");
@@ -8,8 +8,8 @@
「종단 반영」 켬이면 1 m 균형 종단을 꼭짓점에 얹음(`seat_base_pvi` → `rebuild_alignment_profile`).
[적용]은 이 값을 받아 적지 않음 — 서버가 체인에서 같은 조각으로 다시 셈(CLAUDE.md 5).
저장본 = `B05_Profile/route_recommend/initial/<결과 키>.json` `{fingerprint, profile}` — 결과 정리
(`*.json` glob)가 안 보는 하위 폴더 · 결과 짝 없는 파일은 쓸 때 지움 · 지문 다르면 다시 셈.
저장본 = `B05_Profile/route_recommend/initial/<결과 키>.json` `{fingerprint, source, profile}` —
결과 정리(`*.json` glob)가 안 보는 하위 폴더 · 결과 짝 없는 파일은 쓸 때 지움 · 지문 다르면 다시 셈.
응답 칸 `initial_profile`(화면 476-2) = `{stations_m, ground_m, plan_m, line: [[s, z]…], pvi_count,
curve_count, straightened, max_grade_pct, curve_length_m, warnings, with_profile, seconds}` · 실패 =
`{error}`.
@@ -209,8 +209,11 @@ def stat_marks(paths: list[Path]) -> list[str]:
return marks
def fingerprint(parts: dict[str, Any]) -> str:
text = json.dumps({**parts, "source": source_fingerprint()}, sort_keys=True, default=str)
def fingerprint(parts: dict[str, Any], source: str | None = None) -> str:
"""입력 `parts` + 코드 지문(`source` 없으면 지금 코드)."""
text = json.dumps(
{**parts, "source": source or source_fingerprint()}, sort_keys=True, default=str
)
return hashlib.blake2b(text.encode(), digest_size=16).hexdigest()
@@ -218,21 +221,35 @@ def _path(project_root: Path, key: str) -> Path:
return Path(project_root) / INITIAL_DIR / f"{key}.json"
def read_initial(project_root: Path, key: str, mark: str) -> dict[str, Any] | None:
"""저장본 — 없거나 지문 다르면 None."""
def read_initial(
project_root: Path, key: str, mark: str, parts: dict[str, Any] | None = None
) -> dict[str, Any] | None:
"""저장본 — 없거나 지문 다르면 None. `parts` 있음 = 프로젝트 접근(505-1 덧) — 코드 지문은
안 견주고 입력만(그 파일의 코드 지문으로 다시 냄) · 코드 지문 없는 옛 파일은 그대로 꺼냄."""
try:
data = json.loads(_path(project_root, key).read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
if not isinstance(data, dict):
return None
if data.get("fingerprint") == mark:
return data.get("profile")
if parts is None:
return None
source = data.get("source")
return (
data.get("profile") if isinstance(data, dict) and data.get("fingerprint") == mark else None
data.get("profile")
if source is None or fingerprint(parts, source) == data.get("fingerprint")
else None
)
def write_initial(project_root: Path, key: str, mark: str, profile: dict[str, Any]) -> None:
"""한 결과에 하나(덮어씀) · 결과 저장본 짝 없는 파일 지움."""
path = _path(project_root, key)
atomic_write_json(path, {"fingerprint": mark, "profile": profile})
atomic_write_json(
path, {"fingerprint": mark, "source": source_fingerprint(), "profile": profile}
)
results = path.parent.parent
for other in path.parent.glob("*.json"):
if not (results / other.name).exists():
@@ -11,7 +11,9 @@
기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관
지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을
올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 다른 판 · 지문 파일은 안 읽고 남기며 30일 지난
것만 미리 셈 때 지운다(420-2 · storage 를 두 PC · 여러 워크트리가 같이 씀).
것만 미리 셈 때 지운다(420-2 · storage 를 두 PC · 여러 워크트리가 같이 씀). 프로젝트 접근(화면
요청)은 같은 판 · 같은 입력이면 다른 소스 지문 저장본도 꺼냄(`read_stored_any` · 505-1 덧) — 지문
견줌 · 다시 셈은 전처리 체인에서만.
기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
"""
@@ -25,7 +27,7 @@ import time
from contextlib import asynccontextmanager
from datetime import datetime
from pathlib import Path
from typing import Any, AsyncIterator
from typing import Any, AsyncIterator, Callable
from B05_Profile.B05_Profile_Engine_RouteInitial import (
DEFAULT_WEIGHTS,
@@ -91,8 +93,11 @@ def request_fingerprint(
with_profile: bool,
design_speed: int,
standard: dict[str, Any] | None = None,
source: str | None = None,
) -> tuple[str, str]:
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
"""(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키.
`source` = 그 엔진 소스 지문의 키(없으면 지금 코드 · `read_stored_any`)."""
pool = _pool_key(
expected,
terrain,
@@ -109,7 +114,7 @@ def request_fingerprint(
int(design_speed),
json.dumps(standard, sort_keys=True, default=str) if standard else None,
ALGO_VERSION,
source_fingerprint(),
source or source_fingerprint(),
)
key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest()
return key, pool
@@ -156,6 +161,28 @@ def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
return data if _current(data) else None
def read_stored_any(
project_root: Path, pool: str, key_of: Callable[[str], str]
) -> tuple[str, dict[str, Any]] | None:
"""(키, 저장본) — 같은 판 · 묶음이고 그 파일의 소스 지문으로 다시 낸 키(`key_of`)가 파일 이름과
같은 것 중 최근 것 · 없으면 None. 프로젝트 접근 전용(505-1 덧 · 엔진 고쳐도 안내 안 쏟아짐)."""
found: list[tuple[float, str, dict[str, Any]]] = []
for path in store_dir(project_root).glob("*.json"):
try:
data = json.loads(path.read_text(encoding="utf-8"))
mtime = path.stat().st_mtime
except (OSError, ValueError):
continue
if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
continue
if data.get("pool") == pool and key_of(str(data.get("source"))) == path.stem:
found.append((mtime, path.stem, data))
if not found:
return None
_, key, data = max(found, key=lambda item: item[0])
return key, data
def write_stored(
project_root: Path,
key: str,
+7 -1
View File
@@ -63,6 +63,8 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B05 Route Design"])
_PROJECT_PATH_MISSING = {"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."}
# 고른 분석 지형의 지표면 모델이 DB 에 없음(505-2 5d)
TERRAIN_MODEL_MISSING = "고른 분석 지형의 지표면 모델이 없습니다 — 전처리에서 만든 뒤 다시 하세요."
async def _project_paths(project_id: UUID) -> tuple[Path, str] | None:
@@ -336,7 +338,11 @@ async def _recompute(
pool = get_db_pool()
async with pool.acquire() as connection:
await connection.begin() # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음
switched = await switch_design_terrain(connection, project_id, terrain)
try:
switched = await switch_design_terrain(connection, project_id, terrain)
except LookupError: # 고른 지형의 DB 모델 없음 — 409 사유 한 줄(505-2 5d)
await connection.rollback()
return {"error": TERRAIN_MODEL_MISSING}
await connection.commit()
selection = await get_surface_confirmation_params(connection, str(project_id))
latest = await get_latest_route(connection, project_id)
+16 -3
View File
@@ -49,7 +49,8 @@ async def compute_initial(
응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
`recompute` = [재계산] — 묶음 잠금 안에서 이 묶음 저장본 · 메모리 묶음을 버리고 새로 셈(420).
미리 셈 갈래(기본 · 저장 제약 · 종단 켬 · 설계 지형)는 `precompute`(전처리 체인)만 셈 — 화면
요청은 저장본만 · 없으면 409 `NOT_PREPARED`(505-1). 사용자가 바꾼 값 · [재계산]은 그 자리 셈.
요청은 저장본만(엔진 소스 지문이 달라도 · 505-1 덧) · 없으면 409 `NOT_PREPARED`(505-1).
사용자가 바꾼 값 · [재계산]은 그 자리 셈.
"""
from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
from B05_Profile.B05_Profile_Engine_RouteInitial import (
@@ -67,6 +68,7 @@ async def compute_initial(
is_default,
pool_lock,
read_stored,
read_stored_any,
request_fingerprint,
write_stored,
)
@@ -154,9 +156,20 @@ async def compute_initial(
forget_pool_on_standard(pool_key, inputs["standard"])
stored = await asyncio.to_thread(read_stored, project_root, key)
if stored is None and not may_compute:
return JSONResponse(
status_code=409, content={"status": "error", "message": NOT_PREPARED}
# 프로젝트 접근 — 엔진 코드 판이 달라도 같은 입력 저장본을 꺼냄(505-1 덧)
found = await asyncio.to_thread(
read_stored_any,
project_root,
pool_key,
lambda source: request_fingerprint(
route, terrain, streams, **inputs, source=source
)[0],
)
if found is None:
return JSONResponse(
status_code=409, content={"status": "error", "message": NOT_PREPARED}
)
key, stored = found
if stored is not None:
result = stored["result"]
else:
@@ -94,27 +94,29 @@ async def initial_profile_of(
crs = f"EPSG:{epsg}" if epsg is not None else None
surface = {k: selection.get(k) for k in ("source_filter", "method", "smooth")}
models = project_root / _MODELS_SUBDIR
mark = ip.fingerprint(
{
"key": key,
"nodes": result["nodes"],
"curves": result.get("curves"),
"radius": radius_m,
"surface": surface,
"models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))),
"streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]),
# 저장 노선 파일은 안 넣음(505-1) — 초기화(492) 셈은 노선 파일 · 관 · 구조물 ·
# 추가 측점을 안 읽음(선 = 후보 노드) · 넣으면 노선 [저장]마다 다시 셈
"options": asdict(options),
"grade": grade_options.as_dict() if grade_options else None,
"crs": crs,
"with_profile": request.with_profile,
"criteria": request.criteria.model_dump() if request.criteria else None,
"standard": await project_standard(project_id),
"road": await replan._road_settings(project_id),
}
parts = {
"key": key,
"nodes": result["nodes"],
"curves": result.get("curves"),
"radius": radius_m,
"surface": surface,
"models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))),
"streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]),
# 저장 노선 파일은 안 넣음(505-1) — 초기화(492) 셈은 노선 파일 · 관 · 구조물 ·
# 추가 측점을 안 읽음(선 = 후보 노드) · 넣으면 노선 [저장]마다 다시 셈
"options": asdict(options),
"grade": grade_options.as_dict() if grade_options else None,
"crs": crs,
"with_profile": request.with_profile,
"criteria": request.criteria.model_dump() if request.criteria else None,
"standard": await project_standard(project_id),
"road": await replan._road_settings(project_id),
}
mark = ip.fingerprint(parts)
# 프로젝트 접근(compute 거짓) = 코드 지문 안 견줌(505-1 덧)
stored = await asyncio.to_thread(
ip.read_initial, project_root, key, mark, None if compute else parts
)
stored = await asyncio.to_thread(ip.read_initial, project_root, key, mark)
if stored is not None:
return stored
if not compute:
+14 -3
View File
@@ -116,7 +116,11 @@ async def read_route_elevations(project_id: UUID, request: ElevationRequest) ->
@router.get("/{project_id}/route/terrain-sources", response_model=None)
async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse:
"""편집 모달이 고를 수 있는 분석 지형 — 둘이면 고르기를 띄움 · 하나면 숨김."""
"""편집 모달이 고를 수 있는 분석 지형 — 둘이면 고르기를 띄움 · 하나면 숨김.
설계 지형 아닌 쪽은 DB 완료 모델이 있어야 둠 — [확인]이 그 모델로 바꾸므로(505-2 5d)."""
from B04_PreProcess.B04_PreProcess_Service import find_surface_model_for_selection
pool = get_db_pool()
async with pool.acquire() as connection:
stored = await get_project_storage_relative_path(connection, project_id)
@@ -127,8 +131,15 @@ async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse:
)
params = await get_surface_confirmation_params(connection, str(project_id))
confirmed = await _confirmed_model(connection, project_id)
models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR
sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir)
models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR
sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir)
for source in [s for s in sources if s != design_terrain(params)]:
try:
await find_surface_model_for_selection(
connection, project_id, analysis_terrain_params(params, source)
)
except LookupError:
sources.remove(source)
return {
"status": "success",
"project_id": str(project_id),
+2 -1
View File
@@ -191,7 +191,8 @@ export async function mountRouteEditor(
station_interval_m: stationIntervalM,
station_snap: stationSnap,
}),
onShown: () => draw(), // 횡단이 뜨면 그 측점 반폭 선을 그린다.
// 횡단이 뜨면 그 측점 반폭 선을 그린다 — 새로 고침 뒤 고른 측점은 만드는 도중 불려 아래 선언이 아직이라 한 박자 뒤에(505-4).
onShown: () => queueMicrotask(draw),
});
/** 구간 재기 — 계획·예상 노선 위 두 점. 셈·서버 묻기는 `_Measure` 몫(계획서 0-9 ⑤). */
const measure = createMeasureTool({
@@ -94,6 +94,13 @@ def test_라이다_있는_프로젝트_두_지형_다_고름(tmp_path, monkeypat
models = root / "B04_PreProcess" / "models"
assert analysis_terrain_sources(LAS_PARAMS, models) == ["las", "sheet"]
_patch_db(monkeypatch, root, LAS_PARAMS)
import B04_PreProcess.B04_PreProcess_Service as service
async def models_in_db(*args): # sheet 고르기 = DB 완료 모델도 있어야(505-2 5d)
params = {"source_filter": SHEET_ANALYSIS_FILTER}
return [{"id": 2, "status": "COMPLETE", "model_type": "DTM", "generation_params": params}]
monkeypatch.setattr(service, "list_surface_models", models_in_db)
answer = asyncio.run(terrain_router.read_terrain_sources(PID))
assert answer["sources"] == ["las", "sheet"]
@@ -139,3 +139,44 @@ def test_초기_종단_지문에_노선_파일_없음(gather, monkeypatch) -> No
doc["properties"] = {**(doc.get("properties") or {}), "saved": "노선 [저장]"}
route.write_text(json.dumps(doc), "utf-8") # 노선 [저장] = 노선 파일만 바뀜
assert gather["run"]("k1", False) == first
def test_엔진_지문_다른_저장본_접근은_읽음_체인은_다시_셈(wired, monkeypatch) -> None: # noqa: F811
"""505-1 덧 — 엔진 코드를 고쳐도(소스 지문 바뀜) 화면은 옛 저장본을 꺼냄 · 체인만 다시 셈."""
import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store
asyncio.run(router.run_precompute(uuid4()))
assert wired.count == 5
monkeypatch.setattr(store, "source_fingerprint", lambda: "엔진 고침")
for mode in DEFAULT_MODES:
assert _ask(False, mode=mode)["stored"] is True
assert wired.count == 5 # 접근 = 셈 0 · 안내 없음
# 입력이 다르면(제약 바꿈 · 사용자 셈 갈래) 옛 지문을 끌어오지 않음
assert _ask(False, mode="gentle", criteria={"min_radius_m": 15.0})["stored"] is False
assert wired.count == 6
asyncio.run(router.run_precompute(uuid4())) # 체인 = 지문 견줌 · 다시 셈
assert wired.count == 11
sources = {
json.loads(p.read_text("utf-8"))["source"] for p in store_dir(wired.root).glob("*.json")
}
assert "엔진 고침" in sources and len(sources) == 2 # 옛 판은 30일 정리 몫
def test_초기_종단_접근은_코드_지문_안_견줌(gather, monkeypatch) -> None: # noqa: F811
first = gather["run"]()
assert "error" not in first, first
def broken(*args, **kwargs):
raise RuntimeError("다시 셈")
monkeypatch.setattr(ip, "initial_longitudinal", broken)
monkeypatch.setattr(ip, "source_fingerprint", lambda: "엔진 고침")
assert gather["run"]("k1", False) == first # 접근 = 그대로 읽음
assert "다시 셈" in gather["run"]("k1", True)["error"] # 체인 = 다시 셈
path = gather["root"] / ip.INITIAL_DIR / "k1.json"
data = json.loads(path.read_text("utf-8"))
path.write_text(json.dumps({k: v for k, v in data.items() if k != "source"}), "utf-8")
assert gather["run"]("k1", False) == first # 코드 지문 없는 옛 파일도 읽음
path.write_text(json.dumps(data), "utf-8")
gather["state"]["stages"][0]["params"]["max_grade_pct"] = 7.5 # 입력이 바뀜
assert gather["run"]("k1", False) == {"error": router.NOT_PREPARED}
@@ -0,0 +1,101 @@
"""PLAN 505-2 5d — 수치지형도 sheet 고르기 = DB 모델 있을 때만 · 없으면 [확인] 409 사유 한 줄.
./venv/Scripts/python.exe -m pytest resources/tester/test_505_2_terrain_sheet.py -q
고르기 목록(`/route/terrain-sources`)은 npz 파일 + DB 완료 모델을 함께 봄 — 모델 없는 sheet 는 숨김
(화면은 고르기가 숨으면 분석 지형을 안 실음). 바꿈(`switch_design_terrain`)이 모델을 못 찾으면
`_recompute` 가 `{error}` → 노선 되돌리고 409(전엔 LookupError 가 저장 통째 되돌림).
"""
from __future__ import annotations
import asyncio
import sys
from contextlib import asynccontextmanager
from pathlib import Path
from uuid import uuid4
import pytest
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import B04_PreProcess.B04_PreProcess_Service as service # noqa: E402
import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402
import B05_Profile.B05_Profile_Router_Terrain as terrain # noqa: E402
from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER # noqa: E402
LAS = {"source_filter": "classification", "method": "dtm", "smooth": True}
class _Pool:
@asynccontextmanager
async def acquire(self):
yield _Connection()
class _Connection:
async def begin(self):
pass
async def rollback(self):
_Connection.rolled = True
async def commit(self):
raise AssertionError("모델 없으면 확정하지 않음")
def _async(value):
async def inner(*args, **kwargs):
return value
return inner
def _model(source_filter: str) -> dict:
return {
"id": 7,
"status": "COMPLETE",
"model_type": "DTM",
"generation_params": {"source_filter": source_filter},
}
@pytest.fixture
def sources(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
models = tmp_path / terrain._MODELS_SUBDIR
models.mkdir(parents=True)
(models / f"dtm_{SHEET_ANALYSIS_FILTER}.npz").write_bytes(b"") # sheet 파일은 있음
monkeypatch.setattr(terrain, "get_db_pool", lambda: _Pool())
monkeypatch.setattr(terrain, "get_project_storage_relative_path", _async("p"))
monkeypatch.setattr(terrain, "get_surface_confirmation_params", _async(dict(LAS)))
monkeypatch.setattr(terrain, "_confirmed_model", _async(None))
monkeypatch.setattr(terrain, "resolve_stored_project_path", lambda stored: tmp_path)
monkeypatch.setattr(terrain, "_design_grid", lambda *args: None)
def run(models_in_db: list[dict]) -> list[str]:
monkeypatch.setattr(service, "list_surface_models", _async(models_in_db))
return asyncio.run(terrain.read_terrain_sources(uuid4()))["sources"]
return run
def test_모델_없는_sheet_고르기_숨김(sources) -> None:
assert sources([_model("classification")]) == ["las"]
def test_모델_있는_sheet_고르기_보임(sources) -> None:
assert sources([_model(SHEET_ANALYSIS_FILTER)]) == ["las", "sheet"]
def test_모델_없으면_재계산_409_사유(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
import common_util.common_util_surface_confirmation as surface
_Connection.rolled = False
monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool())
monkeypatch.setattr(service, "get_surface_confirmation_params", _async(dict(LAS)))
monkeypatch.setattr(service, "list_surface_models", _async([_model("classification")]))
monkeypatch.setattr(surface, "get_surface_confirmation_params", _async(dict(LAS)))
got = asyncio.run(replan._recompute(uuid4(), tmp_path, "p", terrain="sheet"))
assert got == {"error": replan.TERRAIN_MODEL_MISSING} and _Connection.rolled
@@ -0,0 +1,102 @@
"""PLAN 505-4 — 구조물 고른 채 새로 고침 = 평면 편집기 뜸 · 빈 관경 = 빈 선택지(값 안 채움).
글 시험 = 소스 순서 · 화면 시험 = 자기 포트(5179) Playwright 머리 없는 브라우저 · 서버가 떠 있을 때만 · 검증_sub4.
./venv/Scripts/python.exe -m pytest resources/tester/test_505_4_refresh_blank_pipe.py -q
"""
from __future__ import annotations
import json
import urllib.request
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
BASE = "http://localhost:5179"
PROJECT = "c725d747-50ae-46b0-87ce-4d4e56a53cf3" # 검증_sub4(OWNERS.md)
LIST_URL = f"{BASE}/api/projects/{PROJECT}/route/structures"
def _read(*parts: str) -> str:
return ROOT.joinpath(*parts).read_text(encoding="utf-8")
def test_횡단_뜸_콜백은_만드는_도중_안_그림() -> None:
edit = _read("B05_Profile", "B05_Profile_UI_RouteEdit.ts")
# createCrossPreview 가 고른 측점을 만들며 곧바로 부름 — draw 는 아래 선언(closed · view)이 선 뒤에
assert "onShown: () => queueMicrotask(draw)," in edit
assert "onShown: () => draw()" not in edit
assert edit.index("const crossPreview = createCrossPreview(") < edit.index(
"let closed = false;"
)
def test_관경_빈칸은_빈_보기_채움_없음() -> None:
form = _read("A00_Common", "design", "design_structure_form.ts")
assert 'spec?.typeId === "pipe" && c.id === "pipe" && item.id === "diameter"' in form
assert "c.required && !tag && !blankKept(c, item) ? [] : [new Option(blank," in form
fill = form.index("필수 상자 빈 고르기 칸 = 정의 기본값")
assert form.index("if (blankKept(c, item)) continue;", fill) < form.index(
"values[item.id] =", fill
)
def _up() -> bool:
try:
urllib.request.urlopen(BASE, timeout=3)
return True
except Exception:
return False
JS_DIAMETER = """() => [...document.querySelectorAll('select')]
.filter((s) => s.offsetParent && [...s.options].some((o) => o.value === '800'))
.map((s) => ({ value: s.value, first: s.options[0].value, count: s.options.length }))"""
JS_CANVAS = """() => {
const c = document.querySelector('.b05-routeedit__canvas');
const d = c.getContext('2d').getImageData(0, 0, c.width, c.height).data;
let n = 0;
for (let i = 3; i < d.length; i += 4) if (d[i] > 0) n += 1;
return { visible: !!c.offsetParent, painted: n };
}"""
@pytest.mark.skipif(not _up(), reason="5179 서버 없음")
def test_화면_새로_고침_편집기_뜸_빈_관경_빈_선택지() -> None:
sync = pytest.importorskip("playwright.sync_api")
cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
with sync.sync_playwright() as play:
chrome = play.chromium.launch(channel="msedge", headless=True)
try:
ctx = chrome.new_context(viewport={"width": 1600, "height": 1000})
login = {k: cred[k] for k in ("email", "password")}
assert ctx.request.post(f"{BASE}/api/auth/login/request", data=login, timeout=60000).ok
tab = ctx.new_page()
errors: list[str] = []
tab.on("pageerror", lambda e: errors.append(str(e)))
tab.add_init_script(f"localStorage.setItem('frd_current_project_id', '{PROJECT}')")
tab.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
tab.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
tab.wait_for_timeout(8000)
before = ctx.request.get(LIST_URL).text()
# 관경 빈칸 관(19+8) — 빈 선택지 · 값 안 채움
tab.locator(".b05-route__irregular-item", has_text="19+8").first.click()
tab.wait_for_timeout(3000)
if not tab.evaluate(JS_DIAMETER):
tab.get_by_text("구조물 배치", exact=True).click()
tab.wait_for_timeout(1500)
assert tab.evaluate(JS_DIAMETER) == [{"value": "", "first": "", "count": 5}]
assert ctx.request.get(LIST_URL).text() == before # 열기만 — 저장 값 그대로
# 고른 채 새로 고침 — 편집기가 뜸(전엔 `closed` 선언 전 씀으로 멈춤)
errors.clear()
tab.reload(wait_until="domcontentloaded")
tab.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
tab.wait_for_timeout(9000)
assert not [e for e in errors if "before initialization" in e], errors
canvas = tab.evaluate(JS_CANVAS)
assert canvas["visible"] and canvas["painted"] > 0
assert ctx.request.get(LIST_URL).text() == before
finally:
chrome.close()