feat(564-1): E02 한 번에 해석 체인 — 지반 저장 → 2D → 3D 건기 · 우기 → 암반 → 광역 → 보고서 · 진행 · 끊기
- 새 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 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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))
|
||||
Reference in New Issue
Block a user