Aislo/resources/tester/test_272_2_station_sync.py

"""272-2 — 구조물 측점을 지금 목록 전체로 다시 맞춤(요구 167 · 측점 겹침 근본).

서버는 관 · 구조물 목록 **전체**로 종단 정본의 구조물 측점을 다시 맞춘다 — 빈 목록이어도 옛 구조물
측점을 지우고, 구조물이 떠난 격자 자리는 일반 측점으로 되돌리며(횡단 파일 글까지), 같은 m 면
측점 하나(구조물이 위 · `grid`). 종단 정본에 없는 측점의 횡단 행도 지운다(B08 이 옛 자리를 안 셈).
"""

from __future__ import annotations

import asyncio
import json
import math
import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from B05_Profile.B05_Profile_Engine_Sections import cross_filename  # noqa: E402
from B05_Profile.B05_Profile_Repository import sync_cross_section_rows  # noqa: E402
from B05_Profile.B05_Profile_Router_Confirm import _merge_irregular_into_longitudinal  # noqa: E402

REL = "B06_Section/longitudinal/longitudinal.json"
PIPE = "파형강관 D800"


def _write(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")


def _cross(root: Path, chainage: float) -> Path:
    return root / "B06_Section/cross_sections" / cross_filename(chainage)


def _setup(root: Path) -> None:
    """격자 0~100(20 m) · 옛 관 측점 95.38 · 옛 판 「regular + 글」 60 m."""
    grid = [
        {"chainage_m": float(c), "kind": "bp" if c == 0 else "regular"} for c in range(0, 101, 20)
    ]
    grid[3]["structure"] = "옛 글"
    stations = sorted(
        [*grid, {"chainage_m": 95.38, "kind": "irregular", "structure": PIPE}],
        key=lambda s: s["chainage_m"],
    )
    _write(root / REL, {"stations": stations})
    for station in stations:
        _write(_cross(root, station["chainage_m"]), {**station, "samples": []})


def _stations(root: Path) -> list[dict]:
    return json.loads((root / REL).read_text(encoding="utf-8"))["stations"]


def _at(root: Path, chainage: float) -> dict:
    return next(s for s in _stations(root) if s["chainage_m"] == chainage)


def test_관이_격자_위로_가면_측점_하나_구조물이_위(tmp_path: Path) -> None:
    _setup(tmp_path)
    moved = {"chainage_m": 100.0, "kind": "irregular", "grid": True, "structure": PIPE}
    _merge_irregular_into_longitudinal(tmp_path, REL, [moved])
    chainages = [s["chainage_m"] for s in _stations(tmp_path)]
    assert 95.38 not in chainages and chainages.count(100.0) == 1
    assert _at(tmp_path, 100.0) == moved
    # 옛 판 「regular + 글」 도 글을 지움 — 횡단 파일 글까지
    assert _at(tmp_path, 60.0) == {"chainage_m": 60.0, "kind": "regular"}
    cross = json.loads(_cross(tmp_path, 60.0).read_text(encoding="utf-8"))
    assert cross["kind"] == "regular" and "structure" not in cross


def test_구조물이_떠나면_일반_측점이_돌아옴_빈_목록도(tmp_path: Path) -> None:
    _setup(tmp_path)
    on_grid = {"chainage_m": 100.0, "kind": "irregular", "grid": True, "structure": PIPE}
    _merge_irregular_into_longitudinal(tmp_path, REL, [on_grid])
    _write(_cross(tmp_path, 100.0), {**on_grid, "samples": []})  # 측점 만들기가 쓴 파일 흉내
    _merge_irregular_into_longitudinal(tmp_path, REL, [])
    stations = _stations(tmp_path)
    assert [s["chainage_m"] for s in stations] == [0.0, 20.0, 40.0, 60.0, 80.0, 100.0]
    assert all(s["kind"] in ("bp", "regular") and "structure" not in s for s in stations)
    assert all("grid" not in s for s in stations)
    cross = json.loads(_cross(tmp_path, 100.0).read_text(encoding="utf-8"))
    assert cross["kind"] == "regular" and "structure" not in cross and "grid" not in cross


def test_시점_위_구조물이_떠나도_시점은_시점(tmp_path: Path) -> None:
    _setup(tmp_path)
    at_bp = {"chainage_m": 0.0, "kind": "bp", "grid": True, "structure": PIPE}
    _merge_irregular_into_longitudinal(tmp_path, REL, [at_bp])
    assert _at(tmp_path, 0.0) == at_bp
    _merge_irregular_into_longitudinal(tmp_path, REL, [])
    assert _at(tmp_path, 0.0) == {"chainage_m": 0.0, "kind": "bp"}


def test_측점만_옮겨도_계획선_측점_줄을_다시_셈(tmp_path: Path) -> None:
    """편집 없는 관 측점 581.9 → 582 — 계획선 측점 줄(표 · ▲▼)이 옛 측점을 들고 남지 않음."""
    source = PROJECT_ROOT / "resources/tester/fixtures/b05_192_1_moved_pipe_longitudinal.json"
    path = tmp_path / REL
    _write(path, json.loads(source.read_text(encoding="utf-8")))
    data = json.loads(path.read_text(encoding="utf-8"))
    moved = [
        {**s, "chainage_m": 582.0} if s["chainage_m"] == 581.9 else s
        for s in data["stations"]
        if s.get("kind") == "irregular"
    ]
    assert _merge_irregular_into_longitudinal(tmp_path, REL, moved) is not None
    rows = [row["chainage_m"] for row in _stations_rows(path)]
    assert 582.0 in rows and 581.9 not in rows
    assert rows == [s["chainage_m"] for s in _stations(tmp_path)]
    # 그대로 한 번 더 — 줄이 측점과 같으면 안 건드림
    assert _merge_irregular_into_longitudinal(tmp_path, REL, moved) is None


def test_옛_소수_측점을_정수로_옮기면_꼭짓점_편집도_따라감(tmp_path: Path) -> None:
    """484.213 · 581.902 · 676.591 · 769.28 → 정수 m — 꼭짓점이 높이 그대로 옮겨 감(지우면 계획선이
    그 자리에서 바뀜 · 남기면 측점 없는 꼭짓점이 1000% 넘게 꺾음 · 브라우저 실측)."""
    source = PROJECT_ROOT / "resources/tester/fixtures/b05_192_1_moved_pipe_longitudinal.json"
    data = json.loads(source.read_text(encoding="utf-8"))
    _write(tmp_path / REL, data)
    snapped = [
        {**s, "chainage_m": float(math.floor(s["chainage_m"] + 0.5))}
        for s in data["stations"]
        if s.get("kind") == "irregular"
    ]
    _merge_irregular_into_longitudinal(tmp_path, REL, snapped)
    after = json.loads((tmp_path / REL).read_text(encoding="utf-8"))["profile_alignment"]
    nodes = {n["chainage_m"]: n["elevation_m"] for n in after["base_pvi"]}
    old_nodes = {n["chainage_m"]: n["elevation_m"] for n in data["profile_alignment"]["base_pvi"]}
    for old, new in ((484.213, 484.0), (581.902, 582.0), (676.591, 677.0), (769.28, 769.0)):
        assert old not in nodes and nodes[new] == old_nodes[old]
    offsets = after["edits"]["station_offsets"]
    assert "769.280" not in offsets and "769.000" in offsets


def _stations_rows(path: Path) -> list[dict]:
    return json.loads(path.read_text(encoding="utf-8"))["profile_alignment"]["stations"]


class _Cursor:
    def __init__(self, rows: list[tuple[int, float]]) -> None:
        self.rows = rows
        self.moved: list[tuple[float, int]] = []
        self.deleted: list[int] | None = None

    async def __aenter__(self) -> "_Cursor":
        return self

    async def __aexit__(self, *_a) -> bool:
        return False

    async def execute(self, sql: str, args=None) -> None:
        if sql.startswith("UPDATE"):
            self.moved.append(tuple(args))
        elif sql.startswith("DELETE"):
            self.deleted = list(args)

    async def fetchall(self) -> list[tuple[int, float]]:
        return self.rows


class _Connection:
    def __init__(self, rows: list[tuple[int, float]]) -> None:
        self.cursor_ = _Cursor(rows)

    def cursor(self) -> _Cursor:
        return self.cursor_


def _sync(rows: list[tuple[int, float]], stations: list[float]) -> tuple[int, _Cursor]:
    connection = _Connection(rows)
    return asyncio.run(sync_cross_section_rows(connection, 7, stations)), connection.cursor_


def test_옛_소수_측점_행은_새_정수_측점으로_옮김() -> None:
    """95.38 → 95 — 사용자 설계 값이 든 행을 그대로 옮김(면적은 저장 뒤 서버 재계산)."""
    count, cursor = _sync([(1, 0.0), (2, 95.38), (3, 100.004)], [0.0, 20.0, 95.0, 100.0])
    assert count == 1 and cursor.moved == [(95.0, 2)] and cursor.deleted is None


def test_가까운_새_측점이_없으면_지움() -> None:
    count, cursor = _sync([(1, 0.0), (2, 95.38), (5, 140.0)], [0.0, 20.0, 100.0])
    assert count == 2 and cursor.moved == [] and cursor.deleted == [2, 5]


def test_한_측점에_둘이_다투면_가까운_행만_옮김() -> None:
    count, cursor = _sync([(2, 95.38), (6, 94.7), (7, 100.0)], [95.0, 100.0])
    assert count == 2 and cursor.moved == [(95.0, 6)] and cursor.deleted == [2]


def test_측점_목록이_비면_아무것도_안_건드림() -> None:
    count, cursor = _sync([(1, 0.0), (2, 95.38)], [])
    assert count == 0 and cursor.moved == [] and cursor.deleted is None