Aislo/resources/tester/test_530_2_plan_analyze.py

"""D 예정노선 2단계 노선 정본(PLAN 562-2) — 손 그은 선 · 갈래 · 마크 · 조각 → 서버가 다시 셈 ·
연도 행 · 옛 연도 노선을 선으로 · GET · PUT(산림청만 · 틀린 꼴 거절 · 업체 지정된 조각 지움 막음).

옛 추천 엔진 시험(합성 지형 갈래 넷 · 저장본 · 나란히 셈)은 엔진과 함께 걷음.

./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_analyze.py -q
"""

import json

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from common_util.common_util_auth import require_company
from D03_PlanAnalysis import D03_PlanAnalysis_Network as net

PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
MAIN = [[0.0, 0.0], [30.0, 40.0], [30.0, 100.0], [90.0, 180.0]]  # 50 + 60 + 100 = 210 m
BRANCH = [[30.0, 100.0], [30.0, 150.0]]  # 본선 점 2 에서 50 m


def _network(**over) -> dict:
    body = {
        "epsg": 5186,
        "lines": [
            {"id": "main", "vertices": MAIN, "branch": None},
            {"id": "b1", "vertices": BRANCH, "branch": {"line": "main", "vertex": 2}},
        ],
        "marks": [{"line": "main", "at_m": 75.0}],
        "pieces": [
            {"id": "p1", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2026},
            {"id": "p2", "line": "main", "from_m": 75.004, "to_m": 210.0, "year": 2027},
            {"id": "p3", "line": "b1", "from_m": 0, "to_m": 50.0, "year": 2026},
        ],
    }
    body.update(over)
    return body


def test_derive_pieces_lengths_stations() -> None:
    got = net.derive(_network(), 2026)
    assert [(x["id"], x["length_m"]) for x in got["lines"]] == [("main", 210), ("b1", 50)]
    assert got["lines"][0]["stations"][-2:] == [200, 210] and got["lines"][1]["stations"] == [
        0,
        20,
        40,
        50,
    ]
    pieces = {p["id"]: p for p in got["pieces"]}
    assert [p["id"] for p in got["pieces"]] == ["p1", "p2", "p3"]  # 선 차례 · 앞에서부터
    # 경계 = 마크 그대로(보낸 75.004 아님)
    assert pieces["p2"]["from_m"] == 75.0 and pieces["p2"]["length_m"] == 135.0
    assert pieces["p1"]["planned"] == [[0, 0], [30, 40], [30, 65]]
    assert pieces["p1"]["stations"] == [0, 20, 40, 60, 75]
    assert pieces["p2"]["stations"][:2] == [75, 80]
    assert pieces["p3"]["planned"] == BRANCH  # 갈래 측점 = 갈래마다 No.0


@pytest.mark.parametrize(
    "over",
    [
        {"marks": [{"line": "main", "at_m": 210.0}]},  # 마크가 끝
        {"marks": [{"line": "zz", "at_m": 10.0}]},
        {"marks": [{"line": "main", "at_m": 75.0}, {"line": "main", "at_m": 75.0005}]},
        {"pieces": [{"id": "p1", "line": "main", "from_m": 0, "to_m": 80.0, "year": 2026}]},
        {"pieces": [{"id": "p1", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2025}]},
        {
            "pieces": [
                {"id": "p1", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2026},
                {"id": "p9", "line": "main", "from_m": 0, "to_m": 75.0, "year": 2027},
            ]
        },
        {
            "lines": [
                {"id": "b1", "vertices": BRANCH, "branch": {"line": "main", "vertex": 2}},
                {"id": "main", "vertices": MAIN, "branch": None},
            ],
            "marks": [],
            "pieces": [],
        },  # 부모가 뒤
        {
            "lines": [
                {"id": "main", "vertices": MAIN, "branch": None},
                {"id": "b1", "vertices": BRANCH, "branch": {"line": "main", "vertex": 1}},
            ],
        },  # 갈래 첫 점 ≠ 부모 점
        {"lines": [{"id": "main", "vertices": [[0, 0], [float("nan"), 1]], "branch": None}]},
    ],
)
def test_derive_rejects(over) -> None:
    with pytest.raises(net.NetworkError):
        net.derive(_network(**over), 2026)


def test_year_rows_group_pieces() -> None:
    rows = net.year_rows(net.derive(_network(), 2026)["pieces"])
    assert list(rows) == [2026, 2027]
    first = rows[2026]["route"]
    assert first["mode"] == "manual" and [p["id"] for p in first["pieces"]] == ["p1", "p3"]
    assert first["planned"] == [[0, 0], [30, 40], [30, 65]] and first["length_m"] == 125.0
    assert rows[2026]["start"] == [0, 0] and rows[2026]["end"] == [30, 65]
    assert rows[2027]["route"]["start_chainage_m"] == 75.0


def test_legacy_rows_become_one_line() -> None:
    rows = [
        {"year": 2026, "epsg": 5186, "route_json": json.dumps({"planned": [[0, 0], [30, 40]]})},
        {"year": 2027, "epsg": 5186, "route_json": None},
        {"year": 2028, "epsg": 5186, "route_json": {"planned": [[30, 40], [30, 100]]}},
    ]
    got = net.legacy_network(rows)
    assert got["lines"][0]["vertices"] == [[0, 0], [30, 40], [30, 100]]
    assert got["marks"] == [{"line": "main", "at_m": 50.0}]
    assert [(p["year"], p["from_m"], p["to_m"]) for p in got["pieces"]] == [
        (2026, 0, 50),
        (2028, 50, 110),
    ]
    assert net.derive(got)["pieces"][1]["planned"] == [[30, 40], [30, 100]]
    assert net.legacy_network([{"year": 2026, "epsg": 5186, "route_json": None}]) is None


class _Cursor:
    def __init__(self, db):
        self.db = db
        self.last = ""

    async def __aenter__(self):
        return self

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

    async def execute(self, sql, args=None):
        self.last = sql
        self.db.log.append((sql, args))

    async def fetchone(self):
        return self._rows()[0] if self._rows() else None

    async def fetchall(self):
        return self._rows()

    def _rows(self):
        for key, rows in self.db.answers.items():
            if key in self.last:
                return rows
        return []


class _Connection:
    def __init__(self, db):
        self.db = db

    async def __aenter__(self):
        return self

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

    async def begin(self):
        pass

    async def commit(self):
        self.db.log.append(("COMMIT", None))

    async def rollback(self):
        self.db.log.append(("ROLLBACK", None))

    def cursor(self, *args):
        return _Cursor(self.db)


class _Db:
    def __init__(self):
        self.log: list = []
        self.answers: dict = {"FROM projects": [{"project_year": 2026, "plan_base_json": "{}"}]}

    def acquire(self):
        return _Connection(self)


@pytest.fixture
def client(monkeypatch):
    db = _Db()
    monkeypatch.setattr(net, "get_db_pool", lambda: db)
    base = {"epsg": 5186, "points": [{"x": 0.0, "y": 0.0}, {"x": 90.0, "y": 180.0}]}

    async def load_base(project_id):
        return base

    monkeypatch.setattr(net.plan_base, "load_base", load_base)
    app = FastAPI()
    app.include_router(net.router)
    session = {"role": "USER", "email": "a@forest.go.kr"}
    app.dependency_overrides[require_company] = lambda: session
    test = TestClient(app)
    test.db = db
    test.session = session
    test.base = base
    return test


URL = f"/api/projects/{PID}/plan/network"


def test_get_saved_legacy_or_empty(client) -> None:
    assert client.get(URL).json()["network"] is None
    client.db.answers["FROM plan_route_years"] = [
        {"year": 2026, "epsg": 5186, "route_json": '{"planned": [[0, 0], [30, 40]]}'}
    ]
    got = client.get(URL).json()
    assert got["legacy"] is True and got["derived"]["pieces"][0]["length_m"] == 50
    client.db.answers["FROM plan_networks"] = [{"network_json": json.dumps(_network())}]
    got = client.get(URL).json()
    assert got["legacy"] is False and len(got["derived"]["pieces"]) == 3
    assert got["base"]["points"][1] == {"x": 90.0, "y": 180.0} and got["year_start"] == 2026


def test_put_stores_network_pieces_years(client) -> None:
    got = client.put(URL, json=_network())
    assert got.status_code == 200 and got.json()["derived"]["pieces"][1]["from_m"] == 75.0
    assert got.json()["year_start"] == 2026
    log = client.db.log
    assert log[-1][0] == "COMMIT"
    pieces = [a for s, a in log if s.startswith("INSERT INTO plan_route_pieces")]
    assert [(a[1], a[2], a[3]) for a in pieces] == [
        ("p1", "main", 2026),
        ("p2", "main", 2027),
        ("p3", "b1", 2026),
    ]
    years = [a for s, a in log if s.startswith("INSERT INTO plan_route_years")]
    assert [a[1] for a in years] == [2026, 2027]
    assert json.loads(years[0][7])["pieces"][1]["id"] == "p3"
    assert ("UPDATE projects SET project_year_end = %s WHERE id = %s", (2027, PID)) in log
    stale = [a for s, a in log if "year NOT IN" in s]
    assert stale == [(PID, 2026, 2027)]
    assert not [s for s, _ in log if "plan_base_json = %s" in s]  # 정본 칸 있음 = 안 굳힘


def test_put_freezes_folded_base_first(client) -> None:
    client.db.answers["FROM projects"] = [{"project_year": 2026, "plan_base_json": None}]
    assert client.put(URL, json=_network()).status_code == 200
    sqls = [(s, a) for s, a in client.db.log if s.startswith("UPDATE projects SET plan_base_json")]
    assert len(sqls) == 1 and json.loads(sqls[0][1][0]) == client.base
    first_year = next(i for i, (s, _) in enumerate(client.db.log) if "plan_route_years" in s)
    assert client.db.log.index(sqls[0]) < first_year


def test_put_reset_clears_year_routes(client) -> None:
    got = client.put(URL, json={"epsg": 5186, "lines": [], "marks": [], "pieces": []})
    assert got.status_code == 200 and got.json()["derived"] == {"lines": [], "pieces": []}
    sqls = [s for s, _ in client.db.log]
    assert any(s.startswith("UPDATE plan_route_years SET route_json = NULL") for s in sqls)
    assert not any("project_year_end" in s for s in sqls)


def test_put_rejects(client) -> None:
    bad = _network(marks=[{"line": "main", "at_m": 500.0}])
    assert client.put(URL, json=bad).status_code == 400
    assert (
        client.put(URL, json=_network(lines=[{"id": "a b", "vertices": MAIN}])).status_code == 422
    )
    # 업체가 지정된 조각 · 옛 연도를 지우면 막음
    client.db.answers["piece_id IS NOT NULL"] = [("p2",)]
    one = _network(
        lines=[{"id": "main", "vertices": MAIN}],
        marks=[],
        pieces=[{"id": "p1", "line": "main", "from_m": 0, "to_m": 210, "year": 2026}],
    )
    assert client.put(URL, json=one).status_code == 409
    client.db.answers["piece_id IS NOT NULL"] = []
    client.db.answers["piece_id IS NULL"] = [(2027,)]
    assert client.put(URL, json=one).status_code == 409
    assert not [s for s, _ in client.db.log if s.startswith("INSERT")]
    assert client.put(URL, json=_network(epsg=5187)).status_code == 400  # 기준 지정과 다른 좌표계
    client.session["email"] = "a@gmail.com"
    assert client.put(URL, json=_network()).status_code == 403