Merge remote-tracking branch 'origin/sub_laptop_5' into sub_laptop_1
This commit is contained in:
@@ -411,6 +411,9 @@ process.stdout.write(JSON.stringify({
|
||||
stress: m.sliceStress([input.slice]),
|
||||
ticks: m.niceTicks(-3, 47, 5),
|
||||
zAt: [m.zAt([[0, 0], [10, 10]], 2.5), m.zAt([[10, 10], [0, 0]], -5)],
|
||||
stride: [m.labelStride(100, 64), m.labelStride(40, 64), m.labelStride(10, 64), m.labelStride(0.1, 64)],
|
||||
labels: [1, 2, 5].map((k) => m.planLabelKeys(input.rows, 20, k, null)),
|
||||
labelsActive: m.planLabelKeys(input.rows, 20, 5, "e"),
|
||||
}));
|
||||
"""
|
||||
|
||||
@@ -541,6 +544,10 @@ def test_logic_TS():
|
||||
)
|
||||
assert out["ticks"][0] <= -3 and out["ticks"][-1] >= 47 and 0 in out["ticks"]
|
||||
assert out["zAt"] == [2.5, 0]
|
||||
# 평면 띠 측점 글 솎음 — 간격 px × 걸음 ≥ 64 · 본 측점 중 걸음 배수 · 고른 측점은 늘
|
||||
assert out["stride"] == [1, 2, 10, 100]
|
||||
assert out["labels"] == [["a", "b", "c", "d", "e"], ["a", "c", "e"], ["a"]]
|
||||
assert out["labelsActive"] == ["a", "e"]
|
||||
|
||||
|
||||
def test_엔진_지문은_코드_내용():
|
||||
|
||||
@@ -107,7 +107,7 @@ def test_analyze_cached_force_and_result(client, project, monkeypatch):
|
||||
|
||||
monkeypatch.setattr(run, "run_engine", counted)
|
||||
first = client.post(f"/api/projects/{PID}/stab/analyze3d", json=BODY).json()
|
||||
assert first == {"status": "accepted", "key": "0.00_100.00"}
|
||||
assert first == {"status": "accepted", "key": "0.00_100.00_dry"}
|
||||
done = _wait(client)
|
||||
assert done["error"] is None and done["percent"] == 100
|
||||
assert done["summary"]["cells"] == 5 and done["summary"]["verdict"] == "ng"
|
||||
@@ -133,7 +133,7 @@ def test_analyze_cached_force_and_result(client, project, monkeypatch):
|
||||
_wait(client)
|
||||
assert len(calls) == 3
|
||||
|
||||
doc = client.get(f"/api/projects/{PID}/stab/result3d", params={"key": "0.00_100.00"}).json()
|
||||
doc = client.get(f"/api/projects/{PID}/stab/result3d", params={"key": "0.00_100.00_dry"}).json()
|
||||
grid = doc["grid"]
|
||||
assert len(doc["f"]) == len(doc["z"]) == len(doc["v"]) == grid["nx"] * grid["ny"]
|
||||
f = np.array([np.nan if v is None else v for v in doc["f"]])
|
||||
@@ -143,7 +143,7 @@ def test_analyze_cached_force_and_result(client, project, monkeypatch):
|
||||
assert doc["stale"] is False
|
||||
|
||||
over = client.get(f"/api/projects/{PID}/stab/result3d").json()
|
||||
assert [s["key"] for s in over["saved"]] == ["0.00_100.00"] and over["editable"] is True
|
||||
assert [s["key"] for s in over["saved"]] == ["0.00_100.00_dry"] and over["editable"] is True
|
||||
assert len(over["line"]) == len(STATIONS) and over["sections"][0]["end_m"] == 100.0
|
||||
# E02 지반을 바꾸면 저장본이 「옛 것」
|
||||
(project / "E02_StabGround").mkdir()
|
||||
@@ -179,7 +179,9 @@ def test_cancel_and_restart(client, monkeypatch):
|
||||
body = _wait(client)
|
||||
assert body["cancelled"] is True and body["error"] is None
|
||||
assert (
|
||||
client.get(f"/api/projects/{PID}/stab/result3d", params={"key": "0.00_100.00"}).status_code
|
||||
client.get(
|
||||
f"/api/projects/{PID}/stab/result3d", params={"key": "0.00_100.00_dry"}
|
||||
).status_code
|
||||
== 404
|
||||
)
|
||||
# 다시 시작
|
||||
@@ -191,6 +193,39 @@ def test_cancel_and_restart(client, monkeypatch):
|
||||
assert body["cancelled"] is False and body["percent"] == 100
|
||||
|
||||
|
||||
def test_dry_then_wet_both_kept(client, project, monkeypatch):
|
||||
"""건기 → 우기 = 다른 열쇠 · 앞 결과가 덮이지 않음 · 옛 이름(조건 없음) 파일도 읽음."""
|
||||
monkeypatch.setattr(run, "run_engine", _fake_engine)
|
||||
for condition in ("dry", "wet"):
|
||||
body = client.post(
|
||||
f"/api/projects/{PID}/stab/analyze3d", json={**BODY, "condition": condition}
|
||||
).json()
|
||||
assert body == {"status": "accepted", "key": f"0.00_100.00_{condition}"}
|
||||
_wait(client)
|
||||
over = client.get(f"/api/projects/{PID}/stab/result3d").json()
|
||||
assert [(s["key"], s["condition"]) for s in over["saved"]] == [
|
||||
("0.00_100.00_dry", "dry"),
|
||||
("0.00_100.00_wet", "wet"),
|
||||
]
|
||||
folder = run.results_dir(project)
|
||||
# 옛 이름 — 우기 파일을 조건 없는 이름으로 옮기면 우기 열쇠로 읽힘
|
||||
(folder / "0.00_100.00_wet.json").rename(folder / "0.00_100.00.json")
|
||||
over = client.get(f"/api/projects/{PID}/stab/result3d").json()
|
||||
assert [s["key"] for s in over["saved"]] == ["0.00_100.00_dry", "0.00_100.00_wet"]
|
||||
url = f"/api/projects/{PID}/stab/result3d"
|
||||
assert client.get(url, params={"key": "0.00_100.00_wet"}).json()["stale"] is False
|
||||
again = client.post(
|
||||
f"/api/projects/{PID}/stab/analyze3d", json={**BODY, "condition": "wet"}
|
||||
).json()
|
||||
assert again["status"] == "cached"
|
||||
|
||||
|
||||
def test_engine_hash_in_fingerprint(monkeypatch):
|
||||
before = run.fingerprint("a", None, {})
|
||||
monkeypatch.setattr(run, "engine_hash", lambda: "changed")
|
||||
assert run.fingerprint("a", None, {}) != before
|
||||
|
||||
|
||||
def test_bad_requests(client):
|
||||
url = f"/api/projects/{PID}/stab/analyze3d"
|
||||
assert client.post(url, json={"start_m": 100, "end_m": 50}).status_code == 400
|
||||
|
||||
@@ -170,7 +170,7 @@ def project(tmp_path, monkeypatch) -> Path:
|
||||
path.write_text(json.dumps(ground), encoding="utf-8")
|
||||
r3 = root / "E04_Stab3D" / "results"
|
||||
r3.mkdir(parents=True)
|
||||
(r3 / "0.00_200.00.json").write_text(
|
||||
(r3 / "0.00_200.00_wet.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"fingerprint": "3d-1",
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
"""D03 수동 그리기 셈 거울(PLAN 562-1) — 파이썬 `D03_PlanAnalysis_Draw` ↔
|
||||
TS `logic/D03_PlanAnalysis_Draw.ts`.
|
||||
|
||||
같은 높이 함수(`hill` · x > 800 = 지형 밖)로 두 쪽 값이 같은지 + 손셈 값.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_562_1_plan_draw.py -q
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Draw as draw # noqa: E402
|
||||
|
||||
TS_FILE = ROOT / "D03_PlanAnalysis" / "logic" / "D03_PlanAnalysis_Draw.ts"
|
||||
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
|
||||
|
||||
|
||||
def hill(x: float, y: float) -> float:
|
||||
if x > 800:
|
||||
return math.nan
|
||||
return 0.0005 * x * x + 3 * math.sin(y / 40)
|
||||
|
||||
|
||||
def plane(x: float, y: float) -> float:
|
||||
return 0.1 * x
|
||||
|
||||
|
||||
LINE = [[0.0, 0.0], [30.0, 40.0], [30.0, 100.0], [90.0, 180.0]] # 50 + 60 + 100 = 210 m
|
||||
|
||||
# (함수 이름, 인자) — 높이 자리는 "hill" · "plane" 글자로 넘김
|
||||
CASES = [
|
||||
("segmentLength", [[1, 2], [4, 6]]),
|
||||
("groundSlopePct", [[0, 0], [100, 30], "hill"]),
|
||||
("groundSlopePct", [[100, 30], [0, 0], "hill"]),
|
||||
("groundSlopePct", [[700, 0], [900, 0], "hill"]),
|
||||
("groundSlopePct", [[5, 5], [5, 5], "hill"]),
|
||||
("fixedLengthPoint", [[10, 10], [13, 14], 20]),
|
||||
("fixedLengthPoint", [[10, 10], [10, 10], 20]),
|
||||
("fixedSlopePoint", [[0, 0], [50, 0], 9, "hill"]),
|
||||
("fixedSlopePoint", [[0, 0], [50, 0], -9, "hill"]),
|
||||
("fixedSlopePoint", [[100, 30], [130, 70], 12, "hill"]),
|
||||
("fixedSlopePoint", [[0, 0], [50, 0], 9, "hill", 100]),
|
||||
("fixedSlopePoint", [[600, 0], [700, 0], 1, "hill"]),
|
||||
("compassCandidates", [[0, 0], 50, 5, "plane"]),
|
||||
("compassCandidates", [[200, 10], 60, 15, "hill"]),
|
||||
("compassCandidates", [[0, 0], 50, 50, "plane"]),
|
||||
("compassPoint", [[0, 0], [10, 40], 50, 5, "plane"]),
|
||||
("compassPoint", [[0, 0], [10, 40], 50, 50, "plane"]),
|
||||
("lineLength", [LINE]),
|
||||
("pointAtDistance", [LINE, -5]),
|
||||
("pointAtDistance", [LINE, 75]),
|
||||
("pointAtDistance", [LINE, 500]),
|
||||
("distanceAlong", [LINE, [40, 70]]),
|
||||
("distanceAlong", [LINE, [-10, -10]]),
|
||||
("sliceLine", [LINE, 25, 160]),
|
||||
("sliceLine", [LINE, 160, 25]),
|
||||
("stationChainages", [210]),
|
||||
("stationChainages", [40, 20]),
|
||||
("stationChainages", [0]),
|
||||
]
|
||||
|
||||
_RUNNER = """
|
||||
import { readFileSync, writeFileSync } from "node:fs";
|
||||
import * as draw from "./D03_PlanAnalysis/logic/D03_PlanAnalysis_Draw.js";
|
||||
const heights = {
|
||||
hill: (x, y) => (x > 800 ? NaN : 0.0005 * x * x + 3 * Math.sin(y / 40)),
|
||||
plane: (x, y) => 0.1 * x,
|
||||
};
|
||||
const cases = JSON.parse(readFileSync(process.argv[2], "utf8"));
|
||||
const out = cases.map(([name, args]) =>
|
||||
draw[name](...args.map((a) => (typeof a === "string" ? heights[a] : a))),
|
||||
);
|
||||
writeFileSync(process.argv[3], JSON.stringify(out));
|
||||
"""
|
||||
|
||||
_HEIGHTS = {"hill": hill, "plane": plane}
|
||||
_PY = {
|
||||
"segmentLength": draw.segment_length,
|
||||
"groundSlopePct": draw.ground_slope_pct,
|
||||
"fixedLengthPoint": draw.fixed_length_point,
|
||||
"fixedSlopePoint": draw.fixed_slope_point,
|
||||
"compassCandidates": draw.compass_candidates,
|
||||
"compassPoint": draw.compass_point,
|
||||
"lineLength": draw.line_length,
|
||||
"pointAtDistance": draw.point_at_distance,
|
||||
"distanceAlong": draw.distance_along,
|
||||
"sliceLine": draw.slice_line,
|
||||
"stationChainages": draw.station_chainages,
|
||||
}
|
||||
|
||||
|
||||
def _py(name: str, args: list) -> object:
|
||||
return _PY[name](*(_HEIGHTS[a] if isinstance(a, str) else a for a in args))
|
||||
|
||||
|
||||
def _close(a: object, b: object) -> bool:
|
||||
if a is None or b is None:
|
||||
return a is b
|
||||
if isinstance(a, (int, float)):
|
||||
return isinstance(b, (int, float)) and abs(a - b) < 1e-6
|
||||
return len(a) == len(b) and all(_close(x, y) for x, y in zip(a, b)) # type: ignore[arg-type]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ts_out(tmp_path_factory: pytest.TempPathFactory) -> list:
|
||||
tmp = tmp_path_factory.mktemp("draw")
|
||||
out = tmp / "js"
|
||||
subprocess.run( # noqa: S603 — 고정 실행 파일
|
||||
[
|
||||
"node",
|
||||
str(TSC),
|
||||
str(TS_FILE),
|
||||
"--outDir",
|
||||
str(out),
|
||||
"--rootDir",
|
||||
str(ROOT),
|
||||
"--module",
|
||||
"esnext",
|
||||
"--target",
|
||||
"es2022",
|
||||
"--moduleResolution",
|
||||
"bundler",
|
||||
"--ignoreConfig",
|
||||
],
|
||||
cwd=str(ROOT),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
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")
|
||||
job = tmp / "job.json"
|
||||
result = tmp / "out.json"
|
||||
job.write_text(json.dumps(CASES), encoding="utf-8")
|
||||
subprocess.run( # noqa: S603
|
||||
["node", str(out / "runner.mjs"), str(job), str(result)],
|
||||
cwd=str(ROOT),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
return json.loads(result.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_ts_matches_python(ts_out: list) -> None:
|
||||
for (name, args), ts in zip(CASES, ts_out):
|
||||
assert _close(ts, _py(name, args)), (name, args, ts, _py(name, args))
|
||||
|
||||
|
||||
def test_hand_values() -> None:
|
||||
assert draw.segment_length((1, 2), (4, 6)) == 5
|
||||
# 오르막 + · 내리막 − · 지형 밖 · 길이 0
|
||||
up = draw.ground_slope_pct((0, 0), (100, 0), hill)
|
||||
assert up == pytest.approx(5.0)
|
||||
assert draw.ground_slope_pct((100, 0), (0, 0), hill) == pytest.approx(-5.0)
|
||||
assert draw.ground_slope_pct((700, 0), (900, 0), hill) is None
|
||||
assert draw.ground_slope_pct((5, 5), (5, 5), hill) is None
|
||||
assert draw.fixed_length_point((10, 10), (13, 14), 20) == pytest.approx((22, 26))
|
||||
# z = 0.0005 x² · x0 = 0 → 평균 기울기 0.0005 t = 9 % 첫 자리 t = 180
|
||||
# 부호 무관 · 한도 100 m = 없음
|
||||
assert draw.fixed_slope_point((0, 0), (50, 0), 9, hill) == pytest.approx((180, 0), abs=1e-6)
|
||||
assert draw.fixed_slope_point((0, 0), (50, 0), -9, hill) == pytest.approx((180, 0), abs=1e-6)
|
||||
assert draw.fixed_slope_point((0, 0), (50, 0), 9, hill, 100) is None
|
||||
# 지형 밖에 닿음
|
||||
assert draw.fixed_slope_point((600, 0), (700, 0), 1, hill) is None
|
||||
# 평면 10 % · 반지름 50 · 5 % → cos θ = ±0.5 · 60 · 120 · 240 · 300°
|
||||
got = draw.compass_candidates((0, 0), 50, 5, plane)
|
||||
angles = sorted(round(math.degrees(math.atan2(y, x)) % 360, 6) for x, y in got)
|
||||
assert angles == [60, 120, 240, 300]
|
||||
assert draw.compass_candidates((0, 0), 50, 50, plane) == []
|
||||
near = draw.compass_point((0, 0), (10, 40), 50, 5, plane)
|
||||
assert near == pytest.approx((25, 25 * math.sqrt(3)), abs=1e-6)
|
||||
assert draw.compass_point((0, 0), (10, 40), 50, 50, plane) is None
|
||||
|
||||
|
||||
def test_line_values() -> None:
|
||||
assert draw.line_length(LINE) == 210
|
||||
assert draw.point_at_distance(LINE, -5) == (0, 0)
|
||||
assert draw.point_at_distance(LINE, 75) == (30, 65)
|
||||
assert draw.point_at_distance(LINE, 500) == (90, 180)
|
||||
assert draw.distance_along(LINE, (40, 70)) == 80
|
||||
assert draw.distance_along(LINE, (-10, -10)) == 0
|
||||
assert draw.slice_line(LINE, 25, 160) == pytest.approx(
|
||||
[(15, 20), (30, 40), (30, 100), (60, 140)]
|
||||
)
|
||||
assert draw.slice_line(LINE, 160, 25) == draw.slice_line(LINE, 25, 160)
|
||||
assert draw.station_chainages(210) == [0, 20, 40, 60, 80, 100, 120, 140, 160, 180, 200, 210]
|
||||
assert draw.station_chainages(40, 20) == [0, 20, 40]
|
||||
assert draw.station_chainages(0) == [0]
|
||||
@@ -0,0 +1,233 @@
|
||||
# -*- 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")
|
||||
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