297 lines
12 KiB
Python
297 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""532-3 E02 지반 조건 — 기본값 json 꼴 = E03 Types · KDS 표 · 검사 · 측점 적용 거울(파이썬 ↔ TS) ·
|
|
[저장] · [초기화] · 지층 추가 → 저장 → 다시 읽음(임시 storage · DB 대신 측점 가짜).
|
|
"""
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
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 E02_StabGround import E02_StabGround_Model as model # noqa: E402
|
|
from E02_StabGround import E02_StabGround_Router as router # noqa: E402
|
|
from E03_StabSection import E03_StabSection_Types as types # noqa: E402
|
|
|
|
TS = PROJECT_ROOT / "E02_StabGround" / "logic" / "E02_StabGround_Check.ts"
|
|
TYPESCRIPT = PROJECT_ROOT / "config" / "node_modules" / "typescript"
|
|
DEFAULTS = model.load_defaults()
|
|
LIMITS = DEFAULTS["limits"]
|
|
STATIONS = [
|
|
{"chainage_m": 0.0, "ground_type": "soil", "soil_thickness_m": None},
|
|
{"chainage_m": 20.0, "ground_type": "ripping_rock", "soil_thickness_m": 0.5},
|
|
{"chainage_m": 40.0, "ground_type": "blasting_rock", "soil_thickness_m": 0.8},
|
|
{"chainage_m": 60.0, "ground_type": "ripping_rock", "soil_thickness_m": 0.5},
|
|
]
|
|
|
|
|
|
def _ground(**changes):
|
|
ground = model.build_initial(DEFAULTS, copy.deepcopy(STATIONS))
|
|
ground.update(changes)
|
|
return ground
|
|
|
|
|
|
def _bad(mutate):
|
|
ground = _ground()
|
|
mutate(ground)
|
|
return ground
|
|
|
|
|
|
# 거울 시험 판 — 통과 하나 + 틀린 꼴 여럿(NaN 은 JSON 으로 못 넘겨 null · 글 · 참거짓으로)
|
|
CHECK_CASES = [
|
|
_ground(),
|
|
_bad(lambda g: g["materials"][0].update(gamma_t=9.9)),
|
|
_bad(lambda g: g["materials"][0].update(gamma_sat=17.0)),
|
|
_bad(lambda g: g["materials"][1].update(c=None, phi="33")),
|
|
_bad(lambda g: g["materials"][2].update(phi=True, water="sand")),
|
|
_bad(lambda g: g["materials"][3].update(name=" ", id="soft_rock")),
|
|
_bad(lambda g: g["materials"][4].update(name="풍화토")),
|
|
_bad(lambda g: g["materials"][5].update(name="가" * 31)),
|
|
_bad(lambda g: g["materials"].append(7)),
|
|
_bad(lambda g: g.update(materials=[])),
|
|
_bad(lambda g: g["label_map"].update(soil="nope", fill="")),
|
|
_bad(lambda g: g.update(zones=[{"from_m": 50, "to_m": 50, "soil": "x", "rock": None}])),
|
|
_bad(
|
|
lambda g: g.update(
|
|
zones=[
|
|
{
|
|
"from_m": -1,
|
|
"to_m": 10,
|
|
"soil": None,
|
|
"rock": "hard_rock",
|
|
"soil_thickness_m": 51,
|
|
},
|
|
"zone",
|
|
]
|
|
)
|
|
),
|
|
_bad(lambda g: g["safety_factors"]["wet"].update(cut=0.9, soft=3.1)),
|
|
_bad(lambda g: g["safety_factors"].pop("seismic")),
|
|
_bad(lambda g: g.update(road_surcharge_kpa=101)),
|
|
_bad(lambda g: g["groundwater"].update(wet_rock="full")),
|
|
_bad(lambda g: g.update(seismic={"enabled": "yes", "kh": 0.6})),
|
|
_bad(lambda g: g.update(label_map=[], safety_factors=None, groundwater="x")),
|
|
]
|
|
|
|
RESOLVE_CASES = [
|
|
_ground(),
|
|
_ground(
|
|
zones=[
|
|
{"from_m": 0, "to_m": 20, "soil": "colluvium", "rock": None, "soil_thickness_m": None},
|
|
{"from_m": 20, "to_m": 50, "soil": None, "rock": "hard_rock", "soil_thickness_m": 2.5},
|
|
{"from_m": 55, "to_m": 70, "soil": None, "rock": None, "soil_thickness_m": None},
|
|
]
|
|
),
|
|
_ground(
|
|
zones=[{"from_m": 0, "to_m": 10, "soil": None, "rock": "soft_rock", "soil_thickness_m": 1}]
|
|
),
|
|
]
|
|
|
|
SCRIPT = r"""
|
|
const ts = require(process.argv[1]);
|
|
const fs = require("fs");
|
|
const src = fs.readFileSync(process.argv[2], "utf8");
|
|
const opts = { compilerOptions: { module: ts.ModuleKind.CommonJS } };
|
|
const js = ts.transpileModule(src, opts).outputText;
|
|
const box = { exports: {} };
|
|
new Function("exports", "module", js)(box.exports, box);
|
|
const m = box.exports;
|
|
const input = JSON.parse(fs.readFileSync(0, "utf8"));
|
|
process.stdout.write(JSON.stringify({
|
|
check: input.check.map((g) => m.checkGround(g, input.limits)),
|
|
resolve: input.resolve.map((g) => m.resolveStations(g)),
|
|
}));
|
|
"""
|
|
|
|
|
|
def test_기본값_꼴은_Types_그대로():
|
|
assert model.check_ground(DEFAULTS, LIMITS) == []
|
|
assert [m["id"] for m in DEFAULTS["materials"]] == [m["id"] for m in types.DEFAULT_MATERIALS]
|
|
for item in DEFAULTS["materials"]:
|
|
assert types.material(item) == {k: item[k] for k in types.Material.__annotations__}
|
|
for field in model.MATERIAL_FIELDS:
|
|
assert DEFAULTS["sources"][f"materials.{item['id']}.{field}"].startswith("07 ")
|
|
assert DEFAULTS["road_surcharge_kpa"] == types.DEFAULT_ROAD_SURCHARGE_KPA == 10.0
|
|
assert DEFAULTS["groundwater"] == {
|
|
"dry": "none",
|
|
"wet_soil": "surface",
|
|
"wet_rock": "half_height",
|
|
}
|
|
assert DEFAULTS["seismic"] == {"enabled": False, "kh": 0.0}
|
|
# 07 §4 종합 성토 φ 28~30 — 브레인 기본 25 는 범위 밖이라 아래값 28
|
|
assert next(m for m in DEFAULTS["materials"] if m["id"] == "fill")["phi"] == 28.0
|
|
assert types.groundwater({"mode": "none"}) == {"mode": "none"}
|
|
|
|
|
|
def test_기준안전율_KDS_11_70_05():
|
|
assert DEFAULTS["safety_factors"] == {
|
|
"dry": {"cut": 1.5, "fill": 1.5, "soft": 1.3},
|
|
"wet": {"cut": 1.2, "fill": 1.3, "soft": 1.2},
|
|
"seismic": {"cut": 1.1, "fill": 1.1, "soft": 1.1},
|
|
"short_term": {"cut": 1.1, "fill": 1.1, "soft": 1.1},
|
|
}
|
|
|
|
|
|
def test_범위_검사():
|
|
codes = lambda g: {(e["path"], e["code"]) for e in model.check_ground(g, LIMITS)} # noqa: E731
|
|
assert codes(CHECK_CASES[1]) == {("materials.0.gamma_t", "range")}
|
|
assert codes(CHECK_CASES[2]) == {("materials.0.gamma_sat", "sat_lt_t")}
|
|
assert ("materials.4.name", "name_dup") in codes(CHECK_CASES[6])
|
|
assert ("zones.0.to_m", "order") in codes(CHECK_CASES[11])
|
|
assert ("zones.0.soil", "ref") in codes(CHECK_CASES[11])
|
|
assert codes(CHECK_CASES[13]) == {
|
|
("safety_factors.wet.cut", "range"),
|
|
("safety_factors.wet.soft", "range"),
|
|
}
|
|
assert codes(CHECK_CASES[17]) == {("seismic.enabled", "enum"), ("seismic.kh", "range")}
|
|
|
|
|
|
def test_파이썬과_TS_같음():
|
|
payload = json.dumps({"limits": LIMITS, "check": CHECK_CASES, "resolve": RESOLVE_CASES})
|
|
proc = subprocess.run(
|
|
["node", "-e", SCRIPT, str(TYPESCRIPT), str(TS)],
|
|
input=payload,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
timeout=60,
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
out = json.loads(proc.stdout)
|
|
assert out["check"] == [model.check_ground(g, LIMITS) for g in CHECK_CASES]
|
|
assert out["resolve"] == [model.resolve_stations(g) for g in RESOLVE_CASES]
|
|
assert all(out["check"][1:]) # 틀린 판은 모두 걸림
|
|
|
|
|
|
def test_측점_적용():
|
|
rows = model.resolve_stations(RESOLVE_CASES[1])
|
|
assert [(r["upper"], r["lower"], r["soil_thickness_m"]) for r in rows] == [
|
|
("colluvium", None, None), # 토사 측점 · 두께 없음 = 전부 위 지층
|
|
("colluvium", "hard_rock", 2.5), # 20 = 두 구간 끝 · 뒤 구간이 이김
|
|
("weathered_soil", "hard_rock", 2.5),
|
|
("weathered_soil", "weathered_rock", 0.5), # 빈 구간 = B06 그대로
|
|
]
|
|
assert [r["zoned"] for r in rows] == [True, True, True, True]
|
|
assert model.resolve_stations(RESOLVE_CASES[2])[0]["lower"] == "soft_rock"
|
|
|
|
|
|
def _write_copy(root: Path) -> None:
|
|
"""E01 동결 복사본 꼴(copy_manifest.json + db.json cross_sections) — 측점은 여기서만 읽음."""
|
|
source = root / "E01_StabRegister" / "source"
|
|
source.mkdir(parents=True)
|
|
(source / "copy_manifest.json").write_text("{}", encoding="utf-8")
|
|
rows = [
|
|
{"chainage_m": 20.0, "data": json.dumps({"design": {"ground_type": "ripping_rock"}})},
|
|
{"chainage_m": 0.0, "data": {"design": {"ground_type": "soil"}}},
|
|
{
|
|
"chainage_m": 60.0,
|
|
"data": {"design": {"ground_type": "ripping_rock", "rock_boundary_offset_m": -0.5}},
|
|
},
|
|
{
|
|
"chainage_m": 40.0,
|
|
"data": {"design": {"ground_type": "blasting_rock", "rock_boundary_offset_m": -0.8}},
|
|
},
|
|
]
|
|
(source / "db.json").write_text(json.dumps({"cross_sections": rows}), encoding="utf-8")
|
|
|
|
|
|
@pytest.fixture
|
|
def project(tmp_path, monkeypatch):
|
|
async def fake_project(_project_id):
|
|
return tmp_path, {"company_id": 1, "project_kind": "STABILITY"}
|
|
|
|
monkeypatch.setattr(router, "_project", fake_project)
|
|
_write_copy(tmp_path)
|
|
return tmp_path
|
|
|
|
|
|
PID = "9872e9a7-e483-427b-a6b2-72cdbc063bc3"
|
|
SESSION = {"role": "USER", "company_id": 1}
|
|
|
|
|
|
def test_저장_다시읽음_초기화(project):
|
|
from uuid import UUID
|
|
|
|
pid = UUID(PID)
|
|
first = asyncio.run(router.get_ground(pid, SESSION))
|
|
assert first["saved"] is False and first["editable"] is True
|
|
assert first["ground"]["stations"] == STATIONS
|
|
assert model.initial_path(project).is_file() and not model.ground_path(project).is_file()
|
|
|
|
ground = copy.deepcopy(first["ground"])
|
|
ground["materials"].append(
|
|
{
|
|
"id": "user_1",
|
|
"name": " 점토층 ",
|
|
"gamma_t": 16.5,
|
|
"gamma_sat": 17.5,
|
|
"c": 20,
|
|
"phi": 18,
|
|
"water": "soil",
|
|
"extra": 1,
|
|
}
|
|
)
|
|
ground["materials"][0]["c"] = 12.0
|
|
ground["zones"] = [
|
|
{"from_m": 0, "to_m": 30, "soil": "user_1", "rock": None, "soil_thickness_m": 1.2}
|
|
]
|
|
ground["safety_factors"]["wet"]["cut"] = 1.3
|
|
ground["stations"] = [] # 브라우저 값은 안 받음
|
|
asyncio.run(router.put_ground(pid, router.GroundRequest(ground=ground)))
|
|
|
|
stored = json.loads(model.ground_path(project).read_text(encoding="utf-8"))
|
|
assert stored["materials"][-1] == {
|
|
"id": "user_1",
|
|
"name": "점토층",
|
|
"gamma_t": 16.5,
|
|
"gamma_sat": 17.5,
|
|
"c": 20,
|
|
"phi": 18,
|
|
"water": "soil",
|
|
}
|
|
assert stored["materials"][0]["c"] == 12.0
|
|
assert stored["safety_factors"]["wet"]["cut"] == 1.3
|
|
assert stored["stations"] == STATIONS
|
|
assert set(stored) == set(model.GROUND_KEYS)
|
|
for item in stored["materials"]: # 엔진이 바로 읽음
|
|
types.material(item)
|
|
|
|
again = asyncio.run(router.get_ground(pid, SESSION))
|
|
assert again["saved"] is True and again["ground"] == stored
|
|
|
|
bad = copy.deepcopy(stored)
|
|
bad["materials"][0]["phi"] = 75
|
|
with pytest.raises(HTTPException) as caught:
|
|
asyncio.run(router.put_ground(pid, router.GroundRequest(ground=bad)))
|
|
assert caught.value.status_code == 422
|
|
assert caught.value.detail["errors"] == [{"path": "materials.0.phi", "code": "range"}]
|
|
assert json.loads(model.ground_path(project).read_text(encoding="utf-8")) == stored
|
|
|
|
reset = asyncio.run(router.reset_ground(pid))
|
|
initial = json.loads(model.initial_path(project).read_text(encoding="utf-8"))
|
|
assert reset["ground"] == initial == model.build_initial(DEFAULTS, STATIONS)
|
|
assert json.loads(model.ground_path(project).read_text(encoding="utf-8")) == initial
|
|
|
|
|
|
def test_복사본_없으면_측점_없음(tmp_path):
|
|
assert router._b06_stations(tmp_path) == []
|
|
_write_copy(tmp_path)
|
|
assert router._b06_stations(tmp_path) == STATIONS # 누가 차례 · 암 경계 없으면 −0.5
|
|
|
|
|
|
def test_다른_회사는_보기만(project):
|
|
from uuid import UUID
|
|
|
|
body = asyncio.run(router.get_ground(UUID(PID), {"role": "USER", "company_id": 2}))
|
|
assert body["editable"] is False
|
|
admin = asyncio.run(router.get_ground(UUID(PID), {"role": "SYSTEM_ADMIN", "company_id": None}))
|
|
assert admin["editable"] is True
|