Aislo/resources/tester/test_532_3_e02_ground.py

# -*- 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"


@pytest.fixture
def project(tmp_path, monkeypatch):
    async def fake_project(_project_id):
        return tmp_path, {"company_id": 1, "project_kind": "STABILITY", "source_project_id": None}

    async def fake_stations(_connection, _ids):
        return copy.deepcopy(STATIONS)

    class FakePool:
        def acquire(self):
            class Ctx:
                async def __aenter__(self):
                    return None

                async def __aexit__(self, *_):
                    return False

            return Ctx()

    monkeypatch.setattr(router, "_project", fake_project)
    monkeypatch.setattr(router, "_b06_stations", fake_stations)
    monkeypatch.setattr(router, "get_db_pool", lambda: FakePool())
    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_다른_회사는_보기만(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