Aislo/resources/tester/test_344_4_design_pipeline.py

"""PLAN 344-4 — 실시간 계산 묶음(`A00_Common/design/logic/design_pipeline.ts`) · 가게 칸 · 노선 초안 캐시 칸.

· 늦은 응답 버림(브라우저 단계 · 뒤 단계 끊김) · 같은 입력 다시 안 셈 · 범위 전달 · 범위 합침
· 고리 없음(단계 · 알림 안 입력 쓰기 막음 · follow 만 한 번 · 노선 · 횡단은 못 고침)
· 단계 시간 기록(window.__aisloPipeline) · 결과 칸 메모리 우선 · 노선 초안 즉시 · clearDrafts
짝: `helper_344_4_design_pipeline.cjs`(실제 TS 를 Node 로 돌림).
"""

from __future__ import annotations

import json
import os
import subprocess

import pytest

HERE = os.path.dirname(__file__)
ROOT = os.path.join(HERE, "..", "..")

pytestmark = pytest.mark.skipif(
    not os.path.isdir(os.path.join(ROOT, "config", "node_modules")),
    reason="프론트엔드 의존성 미설치",
)

BROWSER = [
    "route",
    "follow",
    "stations",
    "ground",
    "planline",
    "profile",
    "cross",
    "footprint",
]
AFTER = ["drainage", "corridor"]


@pytest.fixture(scope="module")
def r() -> dict:
    proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
        ["node", os.path.join(HERE, "helper_344_4_design_pipeline.cjs")],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=60,
    )
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def _range(sources, from_m, to_m, keys=()):
    return {"sources": list(sources), "fromM": from_m, "toM": to_m, "keys": list(keys)}


def test_처음은_전_단계_한_번_전_구간(r):
    first = r["basic"]["first"]
    assert first["outcome"] == "done" and first["skipped"] == []
    assert first["allOnce"] == {name: 1 for name in BROWSER + AFTER}
    assert first["firstRange"] is None
    assert r["basic"]["events"] == ["gen1", "route", "follow"]


def test_같은_입력_다시_안_셈(r):
    b = r["basic"]
    assert b["same"] == {"result": None, "generation": 1}  # 같은 값 = 차례 번호도 안 오름
    # 횡단만 바뀌면 노선 ~ 종단 · 배수유역은 건너뜀
    assert b["crossRan"] == ["cross", "footprint", "corridor"]
    assert b["crossSkipped"] == [
        "route",
        "follow",
        "stations",
        "ground",
        "planline",
        "profile",
        "drainage",
    ]
    # 계획고만 바뀌면 측점 · 지반 · 첫 계획선 · 배수유역 건너뜀
    assert b["gradeRan"] == ["profile", "cross", "footprint", "corridor"]
    # 노선이 바뀌면 전부
    assert b["routeRan"] == BROWSER + AFTER


def test_범위_전달(r):
    b = r["basic"]
    assert b["crossRange"] == _range(["cross"], 40, 40, ["0+040"])  # 횡단 = 그 측점
    assert b["gradeRange"] == _range(["grade"], 80, 120)  # 계획고 = 이웃 꼭짓점 사이
    assert b["routeRange"] == _range(["route"], 120, "Infinity")  # 노선 = 바뀐 접선점부터 끝


def test_늦은_응답_버림_브라우저_단계(r):
    late = r["lateBrowser"]
    assert late["a"] == "dropped" and late["b"] == "done"
    assert late["groundGen"] == 2
    assert not [
        c
        for c in late["commits"]
        if c.endswith("@1") and not c.startswith(("route", "follow", "stations"))
    ]


def test_늦은_응답_버림_뒤_단계_끊김(r):
    late = r["lateAfter"]
    assert late["after1Aborted"] is True  # 새 조작 = 옛 요청 끊음
    assert late["commits"] == [2] and late["afterGen"] == 2 and late["generation"] == 2


def test_못_끝낸_단계는_범위를_합쳐_셈(r):
    m = r["merge"]
    assert m["first"] == "dropped"
    assert m["profiles"] == [_range(["grade"], 40, 60), _range(["grade"], 200, 220)]
    assert m["crosses"] == [_range(["grade"], 40, 60), _range(["grade"], 40, 220)]
    assert r["mergeFn"] == _range(["grade", "cross"], 10, 60, ["a"])


def test_고리_없음(r):
    loop = r["loop"]
    assert loop["stageWrite"] is None and loop["notifyWrite"] is None
    assert loop["blocked"] == 3  # 단계 안 1 · 알림 안 2(처음 · 계획고)
    assert loop["afterStart"] == {"generation": 1, "follow": 1, "stations": 1}
    # follow 만 입력을 한 번 고침 — 관은 따라감 · 노선 · 횡단 고침은 버림
    assert loop["stationsPipes"] == [{"id": "p1", "followed": 1}]
    assert loop["stationsRoute"] == {"rev": 1} and loop["stationsCross"] == {}
    assert "follow" not in loop["gradeRan"] and "stations" not in loop["gradeRan"]
    assert loop["warnings"] == 5


def test_단계_시간_기록(r):
    t = r["timing"]
    assert t["size"] == t["limit"] == 20
    assert t["last"]["outcome"] == "done" and t["last"]["sources"] == ["grade"]
    assert t["last"]["stageNames"] == [
        "profile",
        "cross",
        "footprint",
        "corridor",
    ]
    assert t["last"]["totalIsNumber"] is True


def test_가게_결과_칸_메모리_우선(r):
    s = r["store"]
    assert s["beforeFlush"] is None and s["memoryRead"] is True  # 끌기마다 세션에 안 씀
    assert s["afterDelay"] == [{"key": "0", "chainageM": 0}]  # 멈춘 뒤 한 번
    assert s["generationSession"] == []  # 차례 번호는 메모리만
    assert s["cleared"] == {"stations": None, "profile": None}  # clearResults = 메모리도 버림
    assert s["footprintRevived"] is None  # 지운 뒤 미뤄 둔 쓰기가 되살리지 않음
    assert s["published"] == {
        "generation": 1,
        "stations": [{"key": "0", "chainageM": 0}],
        "profileGen": 1,
    }


def test_노선_초안_칸(r):
    s = r["store"]
    assert s["draftNow"] == {"rev": 3, "nodes": [[0, 0], [10, 0]]}  # 초안은 세션 즉시
    assert s["hasDraft"] is True
    assert s["draftCleared"] == {"session": None, "store": None}  # clearDrafts 가 함께 비움
    assert s["registry"]["draft"] == {"bucket": "draft", "scope": "project"}
    for name in ("stations", "profile", "footprint"):
        assert s["registry"][name] == {"bucket": "result", "scope": "project"}