From 8f7bb8e7e68b955b5121d20358fe52d09fae9799 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 27 Oct 2026 19:57:52 +0900 Subject: [PATCH] =?UTF-8?q?feat(564-1):=20E02=20=ED=95=9C=20=EB=B2=88?= =?UTF-8?q?=EC=97=90=20=ED=95=B4=EC=84=9D=20=EC=B2=B4=EC=9D=B8=20=E2=80=94?= =?UTF-8?q?=20=EC=A7=80=EB=B0=98=20=EC=A0=80=EC=9E=A5=20=E2=86=92=202D=20?= =?UTF-8?q?=E2=86=92=203D=20=EA=B1=B4=EA=B8=B0=20=C2=B7=20=EC=9A=B0?= =?UTF-8?q?=EA=B8=B0=20=E2=86=92=20=EC=95=94=EB=B0=98=20=E2=86=92=20?= =?UTF-8?q?=EA=B4=91=EC=97=AD=20=E2=86=92=20=EB=B3=B4=EA=B3=A0=EC=84=9C=20?= =?UTF-8?q?=C2=B7=20=EC=A7=84=ED=96=89=20=C2=B7=20=EB=81=8A=EA=B8=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 새 E02_StabGround_Chain.py — 페이지 해석 함수 그대로 부름 · 지문 같은 단계 건너뜀 · 단계 실패 격리 · 페이지 _JOBS 꽂음 · 3D 진행률 = 구간 × 조건 수로 나눔 - E02 Router 길 셋(POST /stab/run · GET /stab/run/progress · POST /stab/run/cancel) · PUT 과 같은 저장 함수 - 3D 열쇠 = section_key(start, end, condition) 꼴(564-2 뒤 실제 연결) · 단계 상태 적기 자리 = 565-1 뒤 - 새 시험 test_564_1_e02_chain.py(차례 · 비율 · 건너뜀 · 끊기 · 실패 격리 · 3D 조각 · 길) Co-Authored-By: Claude Opus 5.5 --- E02_StabGround/E02_StabGround_Chain.py | 288 +++++++++++++++++++++++ E02_StabGround/E02_StabGround_Router.py | 43 +++- resources/tester/test_564_1_e02_chain.py | 235 ++++++++++++++++++ 3 files changed, 560 insertions(+), 6 deletions(-) create mode 100644 E02_StabGround/E02_StabGround_Chain.py create mode 100644 resources/tester/test_564_1_e02_chain.py diff --git a/E02_StabGround/E02_StabGround_Chain.py b/E02_StabGround/E02_StabGround_Chain.py new file mode 100644 index 000000000..9c75c7f8c --- /dev/null +++ b/E02_StabGround/E02_StabGround_Chain.py @@ -0,0 +1,288 @@ +"""E02 [해석 시작] 체인 (요구 390 1 · PLAN 564-1). + +차례 = 2D → 3D(모든 구간 · 건기 · 우기) → 암반 → 광역 → 보고서. 계산 자리 = ③ 서버 단독. +각 페이지 해석 함수를 그대로 부름(그 파일 안 고침) · 단계마다 입력 지문이 저장본과 같으면 건너뜀 +· 결과는 각 페이지 저장 자리에 영구 저장 · 한 단계가 실패해도 다음 단계로. +도는 동안 그 페이지 진행 표(`_JOBS`)에 같은 일감을 꽂음 → E03 · E04 · E07 화면이 「해석 중」 +· 그 페이지 [해석] 은 409. [끊기] = 다음 확인 때 멈춤(3D 는 엔진 안 · 그 밖은 단계 사이). + +진행 꼴 = {running, percent, stage, stage_name, steps:[{key, name, state, seconds}], + message, error, cancelled} · state = wait · run · done · skip · error · cancel +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from collections.abc import Awaitable, Callable +from pathlib import Path +from typing import Any +from uuid import UUID + +logger = logging.getLogger(__name__) + +#: 단계 — (열쇠, 이름, 비율 몫) · 몫 = 1.2 km 실측 시간 비(2D 1~3 분 · 3D 건기 · 우기 약 25 분) +STEPS: tuple[tuple[str, str, float], ...] = ( + ("2d", "2D 측점 해석", 10.0), + ("3d", "3D 구간 해석", 84.0), + ("rock", "암반 해석", 2.0), + ("area", "광역 위험", 2.0), + ("report", "보고서", 2.0), +) +CONDITIONS_3D = ("dry", "wet") + +Report = Callable[[float, str], None] # (단계 안 비율 0~1, 글) +Stopped = Callable[[], bool] +Step = Callable[[Report, Stopped], Awaitable[str]] # → "done" | "skip" + +_JOBS: dict[str, dict[str, Any]] = {} +_STOPS: dict[str, bool] = {} +_TASKS: set[asyncio.Task] = set() + + +def idle() -> dict[str, Any]: + return { + "running": False, + "percent": 0, + "stage": None, + "stage_name": None, + "steps": [{"key": k, "name": n, "state": "wait", "seconds": None} for k, n, _ in STEPS], + "message": None, + "error": None, + "cancelled": False, + } + + +def _mark_stage(project_id: str, key: str, state: str) -> None: + """대시보드 단계 상태 — ponytail: 지금은 안 적음 · 565-1 함수가 들어오면 여기서 부름.""" + + +async def run_chain( + job: dict[str, Any], steps: dict[str, Step], stopped: Stopped, project_id: str = "" +) -> None: + """STEPS 차례로 돎 — job 을 제자리에서 고침(진행 GET 이 그대로 읽음).""" + total = sum(w for _, _, w in STEPS) + base = 0.0 + errors: list[str] = [] + for index, (key, name, weight) in enumerate(STEPS): + row = job["steps"][index] + if stopped(): + job.update(cancelled=True, message="끊음") + row["state"] = "cancel" + break + job.update(stage=key, stage_name=name, message=name) + row["state"] = "run" + _mark_stage(project_id, key, "run") + + def report(frac: float, message: str, base: float = base, weight: float = weight) -> None: + frac = min(1.0, max(0.0, float(frac))) + job.update(percent=min(99, int(100 * (base + weight * frac) / total)), message=message) + + started = time.perf_counter() + try: + row["state"] = await steps[key](report, stopped) + except InterruptedError: + row["state"] = "cancel" + job.update(cancelled=True, message="끊음") + break + except Exception as error: # noqa: BLE001 — 단계 실패 격리 · 까닭은 그대로 올림 + logger.exception("E02 체인 %s 실패: project_id=%s", key, project_id) + row["state"] = "error" + errors.append(f"{name}: {getattr(error, 'detail', None) or error}") + finally: + row["seconds"] = round(time.perf_counter() - started, 1) + _mark_stage(project_id, key, row["state"]) + base += weight + job["percent"] = min(99, int(100 * base / total)) + job["error"] = " · ".join(errors) or None + if not job["cancelled"]: + job.update(percent=100, message="끝") + job.update(stage=None, stage_name=None) + + +# --------------------------------------------------------------------------- +# 실제 단계 — 페이지 함수 부르기(늦은 import · E03 · E06 라우터가 E02 라우터를 import 함) +# --------------------------------------------------------------------------- + + +def _plug(jobs: dict[str, dict[str, Any]], pid: str, extra: dict[str, Any]) -> dict[str, Any]: + """그 페이지 진행 표에 체인 일감을 꽂음 — 페이지 화면 「해석 중」 · 페이지 [해석] 409.""" + job = {"running": True, "percent": 0, "message": "한 번에 해석", "error": None, **extra} + jobs[pid] = job + return job + + +def build_steps(project_id: UUID, project_root: Path) -> dict[str, Step]: + from E01_StabRegister import E01_StabRegister_Source as source + from E03_StabSection import E03_StabSection_Router as e03 + from E04_Stab3D import E04_Stab3D_Router as e04 + from E04_Stab3D import E04_Stab3D_Run as run3d + from E05_StabReport import E05_StabReport_Build as e05_build + from E05_StabReport import E05_StabReport_Router as e05 + from E05_StabReport import E05_StabReport_Store as e05_store + from E06_StabRock import E06_StabRock_Router as e06 + from E07_StabArea import E07_StabArea_Router as e07 + + pid = str(project_id) + root = project_root + + async def step_2d(report: Report, _stopped: Stopped) -> str: + plan = await asyncio.to_thread(e03.plan_run, root, e03._ground(root)) + if e03.read_derived(root, e03.KIND_SUMMARY, plan["fp"]): + return "skip" + page = _plug(e03._JOBS, pid, {"computed": 0}) + + def relay(percent: int, message: str) -> None: + page.update(percent=int(percent), message=message) + report(percent / 100, f"2D {message}") + + try: + page["computed"] = await asyncio.to_thread(e03.run_all, root, plan, relay) + page.update(percent=100, message="끝") + finally: + page["running"] = False + return "done" + + def search_settings() -> dict[str, Any]: + """3D 탐색 설정 = 가장 최근 저장본 설정 · 없으면 기본값(구간 · 조건 빼고).""" + files = sorted(run3d.results_dir(root).glob("*.json"), key=lambda p: p.stat().st_mtime_ns) + saved = run3d.load(root, files[-1].stem) if files else None + fields = e04.SearchSettings.model_fields + values = {k: v for k, v in ((saved or {}).get("settings") or {}).items() if k in fields} + return e04.SearchSettings(**values).model_dump() + + async def step_3d(report: Report, stopped: Stopped) -> str: + src = e04._source_root(root) + stations = await asyncio.to_thread(source.stations, src) + sections = run3d.auto_sections(stations) + search = await asyncio.to_thread(search_settings) + units = [(s, c) for s in sections for c in CONDITIONS_3D] + computed = 0 + for index, (section, condition) in enumerate(units): + if stopped(): + raise InterruptedError + settings = { + "start_m": round(section["start_m"], 2), + "end_m": round(section["end_m"], 2), + "condition": condition, + **search, + } + key = run3d.section_key(settings["start_m"], settings["end_m"], condition) + label = f"3D {index + 1} / {len(units)} 구간" + finger = await asyncio.to_thread(e04._current_fingerprint, root, src, settings) + saved = await asyncio.to_thread(run3d.load, root, key) + if saved and saved["fingerprint"] == finger: + report((index + 1) / len(units), label) + continue + page = _plug(e04._JOBS, pid, {**e04._IDLE, "running": True, "key": key}) + e04._STOPS[pid] = False + + def relay(percent: int, message: str, index: int = index, page=page) -> None: + page.update(percent=max(0, min(99, int(percent))), message=message) + report((index + percent / 100) / len(units), label) + + try: + page["summary"] = await asyncio.to_thread( + run3d.analyze, + root, + src, + settings, + finger, + relay, + lambda: stopped() or e04._STOPS.get(pid, False), + ) + page.update(percent=100, message="끝") + except InterruptedError: + page.update(cancelled=True, message="끊음") + raise + finally: + page["running"] = False + computed += 1 + return "done" if computed else "skip" + + async def step_rock(_report: Report, _stopped: Stopped) -> str: + out = await asyncio.to_thread(e06._run, root) + return "skip" if out["status"] == "fresh" else "done" + + async def step_area(report: Report, _stopped: Stopped) -> str: + src = e07._source_root(root) + saved = await asyncio.to_thread(e07.load, root) + base = (saved or {}).get("settings") or e07.defaults(root) + settings = e07.AreaRequest(**base).model_dump(exclude={"force"}) + finger = e07.fingerprint(source.manifest(src)["source_fingerprint"], settings) + if saved and saved["fingerprint"] == finger: + return "skip" + page = _plug(e07._JOBS, pid, {}) + + def relay(percent: int, message: str) -> None: + page.update(percent=max(0, min(99, int(percent))), message=message) + report(percent / 100, f"광역 {message}") + + try: + await asyncio.to_thread(e07.run, root, src, settings, finger, relay) + page.update(percent=100, message="끝") + finally: + page["running"] = False + return "done" + + async def step_report(_report: Report, _stopped: Stopped) -> str: + async with e05._lock(project_id): + book, meta = e05_store.load(root) + fresh = await asyncio.to_thread(e05_build.analysis_fingerprint, root) + if book is not None and meta and meta.get("fingerprint") == fresh: + return "skip" + await asyncio.to_thread(e05_store.refresh, root) + return "done" + + return { + "2d": step_2d, + "3d": step_3d, + "rock": step_rock, + "area": step_area, + "report": step_report, + } + + +def busy(project_id: UUID) -> bool: + """체인 · 페이지 해석 하나라도 도는지 — 한 프로젝트에 체인 하나만.""" + from E03_StabSection import E03_StabSection_Router as e03 + from E04_Stab3D import E04_Stab3D_Router as e04 + from E07_StabArea import E07_StabArea_Router as e07 + + pid = str(project_id) + return any( + (jobs.get(pid) or {}).get("running") for jobs in (_JOBS, e03._JOBS, e04._JOBS, e07._JOBS) + ) + + +def start(project_id: UUID, project_root: Path) -> dict[str, Any]: + """뒤에서 체인 시작 — 부르는 쪽이 busy 를 먼저 봄.""" + pid = str(project_id) + job = {**idle(), "running": True, "message": "시작"} + _JOBS[pid] = job + _STOPS[pid] = False + steps = build_steps(project_id, project_root) + + async def work() -> None: + try: + await run_chain(job, steps, lambda: _STOPS.get(pid, False), pid) + finally: + job["running"] = False + + task = asyncio.create_task(work(), name=f"E02 chain {pid}") + _TASKS.add(task) + task.add_done_callback(_TASKS.discard) + return job + + +def progress(project_id: UUID) -> dict[str, Any]: + return dict(_JOBS.get(str(project_id)) or idle()) + + +def cancel(project_id: UUID) -> bool: + pid = str(project_id) + running = bool((_JOBS.get(pid) or {}).get("running")) + if running: + _STOPS[pid] = True + return running diff --git a/E02_StabGround/E02_StabGround_Router.py b/E02_StabGround/E02_StabGround_Router.py index aee593430..66fe01cc6 100644 --- a/E02_StabGround/E02_StabGround_Router.py +++ b/E02_StabGround/E02_StabGround_Router.py @@ -5,6 +5,8 @@ - `PUT …/stab/ground` — [저장] 검사 뒤 작업본 `E02_StabGround/ground.json` · 측점(B06 암 경계)은 서버 값 - `POST …/stab/ground/reset` — [초기화] 초기값을 작업본에 덮어씀(재계산 아님) +- `POST …/stab/run` — [해석 시작] 지반 저장 → 한 번에 해석 체인(`E02_StabGround_Chain` · PLAN 564-1) + · `GET …/stab/run/progress` · `POST …/stab/run/cancel` 초기값 = E 생성 직후 마스터 기본값 + 복사본 B06 측점(`ensure_initial`) · 그 밖에는 쓰지 않음. 공유 회사(보기만)는 `require_project_access` 가 PUT · POST 를 막음. @@ -29,6 +31,7 @@ from common_util.common_util_storage import resolve_stored_project_path from config.config_db import get_db_pool from config.config_system_design import STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M from E01_StabRegister.E01_StabRegister_Source import db_snapshot, missing_stations, source_dir +from E02_StabGround import E02_StabGround_Chain as chain from E02_StabGround import E02_StabGround_Model as model router = APIRouter(prefix="/api/projects", tags=["E02 Stab Ground"]) @@ -130,20 +133,25 @@ async def get_ground( } -@router.put("/{project_id}/stab/ground") -async def put_ground(project_id: UUID, payload: GroundRequest) -> dict[str, Any]: - root, _ = await _project(project_id) +def _save(root: Path, payload: dict[str, Any]) -> dict[str, Any]: + """검사 뒤 작업본 쓰기 — 틀리면 422.""" defaults = model.load_defaults() - errors = model.check_ground(payload.ground, defaults["limits"]) + errors = model.check_ground(payload, defaults["limits"]) if errors: raise HTTPException(status_code=422, detail={"errors": errors}) - ground = model.clean(payload.ground) + ground = model.clean(payload) # 측점(B06 암 경계)은 서버 값 — 브라우저 값을 받아 적지 않음 base = _read_working(root) or ensure_initial(root) ground["stations"] = base.get("stations", []) ground["version"] = defaults["version"] atomic_write_json(model.ground_path(root), ground) - return {"ground": ground} + return ground + + +@router.put("/{project_id}/stab/ground") +async def put_ground(project_id: UUID, payload: GroundRequest) -> dict[str, Any]: + root, _ = await _project(project_id) + return {"ground": _save(root, payload.ground)} @router.post("/{project_id}/stab/ground/reset") @@ -152,3 +160,26 @@ async def reset_ground(project_id: UUID) -> dict[str, Any]: initial = ensure_initial(root) atomic_write_json(model.ground_path(root), initial) return {"ground": model.upgrade(initial, model.load_defaults())} + + +@router.post("/{project_id}/stab/run") +async def run_chain(project_id: UUID, payload: GroundRequest) -> dict[str, Any]: + """[해석 시작] — 지반 저장 → 체인(뒤에서) · 진행은 `GET …/stab/run/progress`.""" + root, _ = await _project(project_id) + if chain.busy(project_id): + raise HTTPException(status_code=409, detail="이미 해석 중입니다.") + ground = _save(root, payload.ground) + return {"ground": ground, **chain.start(project_id, root)} + + +@router.get("/{project_id}/stab/run/progress") +async def run_progress(project_id: UUID) -> dict[str, Any]: + await _project(project_id) + return chain.progress(project_id) + + +@router.post("/{project_id}/stab/run/cancel") +async def run_cancel(project_id: UUID) -> dict[str, Any]: + """[끊기] — 다음 확인 때 멈춤 · 끝난 단계 저장본은 그대로.""" + await _project(project_id) + return {"stopping": chain.cancel(project_id)} diff --git a/resources/tester/test_564_1_e02_chain.py b/resources/tester/test_564_1_e02_chain.py new file mode 100644 index 000000000..3f97b87e1 --- /dev/null +++ b/resources/tester/test_564_1_e02_chain.py @@ -0,0 +1,235 @@ +# -*- coding: utf-8 -*- +"""564-1 E02 [해석 시작] 체인 — 가짜 단계로 차례 · 비율 · 건너뜀 · 끊기 · 실패 격리 · +3D 단계 = 구간 × 건기 · 우기 · 지문 같은 조각 건너뜀 · 구간 수로 나눈 진행률 · +길 셋(지반 저장 → 시작 · 두 번째 409 · 끊기). +""" + +import asyncio +import copy +import sys +from pathlib import Path +from uuid import UUID + +import pytest +from fastapi import HTTPException + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from E01_StabRegister import E01_StabRegister_Source as source # noqa: E402 +from E02_StabGround import E02_StabGround_Chain as chain # noqa: E402 +from E02_StabGround import E02_StabGround_Model as model # noqa: E402 +from E02_StabGround import E02_StabGround_Router as router # noqa: E402 +from E04_Stab3D import E04_Stab3D_Router as e04 # noqa: E402 +from E04_Stab3D import E04_Stab3D_Run as run3d # noqa: E402 + +PID = UUID("9872e9a7-e483-427b-a6b2-72cdbc063bc3") +KEYS = [k for k, _, _ in chain.STEPS] +TOTAL = sum(w for _, _, w in chain.STEPS) + + +def _fake(log, states=None, fail=(), interrupt=(), stop_after=None, flag=None): + """가짜 단계 — 부른 차례 · 단계 안 반쯤 진행 · 정한 상태 · 실패 · 끊김.""" + states = states or {} + + def make(key): + async def step(report, stopped): + log.append(key) + report(0.5, f"{key} 반") + if key in fail: + raise RuntimeError(f"{key} 망가짐") + if key in interrupt: + raise InterruptedError + if key == stop_after: + flag[0] = True + return states.get(key, "done") + + return step + + return {k: make(k) for k in KEYS} + + +class _Job(dict): + """update 마다 진행률을 적어 둠(뒤로 안 가는지 봄).""" + + def __init__(self, *a, **kw): + super().__init__(*a, **kw) + self.seen = [] + + def update(self, *a, **kw): + super().update(*a, **kw) + self.seen.append(self["percent"]) + + +def _run(steps, stopped=lambda: False): + job = _Job({**chain.idle(), "running": True}) + asyncio.run(chain.run_chain(job, steps, stopped)) + return job, job.seen + + +def test_차례_비율_건너뜀(): + log = [] + job, seen = _run(_fake(log, states={"rock": "skip", "report": "skip"})) + assert log == KEYS + assert [s["state"] for s in job["steps"]] == ["done", "done", "skip", "done", "skip"] + assert all(s["seconds"] is not None for s in job["steps"]) + assert job["percent"] == 100 and job["message"] == "끝" and job["error"] is None + assert job["stage"] is None and not job["cancelled"] + assert seen == sorted(seen) # 진행률은 뒤로 안 감 + # 3D 반쯤 = 2D 몫 + 3D 몫 반(몫 = STEPS 비율) + half_3d = int(100 * (chain.STEPS[0][2] + chain.STEPS[1][2] / 2) / TOTAL) + assert half_3d in seen + + +def test_끊기_다음_단계_안_돎(): + log, flag = [], [False] + job, _ = _run(_fake(log, stop_after="3d", flag=flag), stopped=lambda: flag[0]) + assert log == ["2d", "3d"] + assert [s["state"] for s in job["steps"]] == ["done", "done", "cancel", "wait", "wait"] + assert job["cancelled"] and job["percent"] < 100 + + +def test_단계_안에서_끊김(): + log = [] + job, _ = _run(_fake(log, interrupt={"3d"})) + assert log == ["2d", "3d"] + assert [s["state"] for s in job["steps"]][:3] == ["done", "cancel", "wait"] + assert job["cancelled"] and job["message"] == "끊음" + + +def test_실패는_그_단계만(): + log = [] + job, _ = _run(_fake(log, fail={"2d", "area"})) + assert log == KEYS + assert [s["state"] for s in job["steps"]] == ["error", "done", "done", "error", "done"] + assert "2D 측점 해석: 2d 망가짐" in job["error"] and "광역 위험" in job["error"] + assert job["percent"] == 100 + + +# --------------------------------------------------------------------------- +# 3D 단계 — 구간 × 조건 · 지문 같으면 건너뜀 · 구간 수로 나눈 진행률 · E04 [끊기] +# --------------------------------------------------------------------------- + +STATIONS = [{"chainage_m": float(c)} for c in range(0, 320, 20)] # 0~300 m → 구간 셋 + + +@pytest.fixture +def fake_3d(tmp_path, monkeypatch): + calls = [] + fresh = {"0.00_100.00_dry"} # 이 조각만 저장본 지문이 지금과 같음 + monkeypatch.setattr(e04, "_source_root", lambda root: root) + monkeypatch.setattr(source, "stations", lambda src: STATIONS) + monkeypatch.setattr(e04, "_current_fingerprint", lambda *a: "fp") + # 564-2 열쇠 꼴(조건 들어감) — 그 전 판에서도 시험이 돌게 가짜로 + monkeypatch.setattr(run3d, "section_key", lambda s, e, c="dry": f"{s:.2f}_{e:.2f}_{c}") + monkeypatch.setattr( + run3d, "load", lambda root, key: {"fingerprint": "fp"} if key in fresh else None + ) + + def analyze(root, src, settings, finger, on_progress, should_stop): + calls.append((settings["start_m"], settings["end_m"], settings["condition"])) + assert (e04._JOBS.get(str(PID)) or {}).get("running") # E04 화면 「해석 중」 + on_progress(50, "탐색") + if should_stop(): + raise InterruptedError + return {"min_fs": 1.5} + + monkeypatch.setattr(run3d, "analyze", analyze) + e04._JOBS.pop(str(PID), None) + return tmp_path, calls + + +def test_3D_구간_조건_건너뜀_진행률(fake_3d): + root, calls = fake_3d + step = chain.build_steps(PID, root)["3d"] + seen = [] + state = asyncio.run(step(lambda f, m: seen.append((round(f, 4), m)), lambda: False)) + sections = run3d.auto_sections(STATIONS) + assert len(sections) == 3 + units = len(sections) * len(chain.CONDITIONS_3D) + assert state == "done" and len(calls) == units - 1 # 지문 같은 조각 하나 건너뜀 + assert calls[0] == (0.0, 100.0, "wet") and {c for *_, c in calls} == {"dry", "wet"} + assert seen[0] == (round(1 / units, 4), f"3D 1 / {units} 구간") # 건너뛴 조각 = 바로 한 칸 + assert (round(1.5 / units, 4), f"3D 2 / {units} 구간") in seen # 둘째 조각 반쯤 + assert [f for f, _ in seen] == sorted(f for f, _ in seen) + assert not e04._JOBS[str(PID)]["running"] + + +def test_3D_E04_끊기(fake_3d): + root, calls = fake_3d + step = chain.build_steps(PID, root)["3d"] + + def report(_f, _m): + e04._STOPS[str(PID)] = True # E04 화면 [끊기] + + with pytest.raises(InterruptedError): + asyncio.run(step(report, lambda: False)) + assert len(calls) == 1 and e04._JOBS[str(PID)]["cancelled"] + + +def test_3D_모두_최신이면_건너뜀(fake_3d, monkeypatch): + root, calls = fake_3d + monkeypatch.setattr(run3d, "load", lambda root, key: {"fingerprint": "fp"}) + step = chain.build_steps(PID, root)["3d"] + assert asyncio.run(step(lambda f, m: None, lambda: False)) == "skip" and not calls + + +# --------------------------------------------------------------------------- +# 길 — [해석 시작] = 지반 저장 → 시작 · 도는 중 두 번째 409 · 끊기 +# --------------------------------------------------------------------------- + + +def test_길_저장_시작_409_끊기(tmp_path, monkeypatch): + async def fake_project(_project_id): + return tmp_path, {"company_id": 1, "project_kind": "STABILITY"} + + monkeypatch.setattr(router, "_project", fake_project) + gate = {} + + def fake_steps(_pid, _root): + async def slow(report, stopped): + await gate["open"].wait() + return "done" + + async def quick(report, stopped): + return "skip" + + return {k: (slow if k == "2d" else quick) for k in KEYS} + + monkeypatch.setattr(chain, "build_steps", fake_steps) + ground = model.build_initial(model.load_defaults(), []) + ground["materials"][0]["c"] = 7.0 + + async def scenario(): + gate["open"] = asyncio.Event() + first = await router.run_chain(PID, router.GroundRequest(ground=copy.deepcopy(ground))) + assert first["running"] and first["ground"]["materials"][0]["c"] == 7.0 + assert model.ground_path(tmp_path).is_file() # 지반 저장 먼저 + await asyncio.sleep(0) + with pytest.raises(HTTPException) as busy: + await router.run_chain(PID, router.GroundRequest(ground=copy.deepcopy(ground))) + assert busy.value.status_code == 409 + assert (await router.run_progress(PID))["stage"] == "2d" + assert (await router.run_cancel(PID))["stopping"] is True + gate["open"].set() + for _ in range(50): + await asyncio.sleep(0) + return await router.run_progress(PID) + + done = asyncio.run(scenario()) + assert not done["running"] and done["cancelled"] + assert [s["state"] for s in done["steps"]][:2] == ["done", "cancel"] + + +def test_틀린_지반은_시작_안_함(tmp_path, monkeypatch): + async def fake_project(_project_id): + return tmp_path, {"company_id": 1, "project_kind": "STABILITY"} + + monkeypatch.setattr(router, "_project", fake_project) + chain._JOBS.pop(str(PID), None) + ground = model.build_initial(model.load_defaults(), []) + ground["materials"][0]["gamma_t"] = 9.9 + with pytest.raises(HTTPException) as bad: + asyncio.run(router.run_chain(PID, router.GroundRequest(ground=ground))) + assert bad.value.status_code == 422 and not chain._JOBS.get(str(PID))