Aislo/resources/tester/test_638_sets.py

# -*- coding: utf-8 -*-
"""638-1 예정노선 노선 세트(연도-번호 · 요구 547 · 548) — 서버.

- 조각 `set` · 옛 조각(set 없음) = 그 해 선 차례 · 거리 차례로 최댓값 + 1
- 같은 선 이웃 같은 세트 = 하나(앞 조각 id · 사이 마크 걷음)
- 세트 셈 `derived.sets[]` — 덩이 수 · 갈래 · 길이 · 조각
- 업체 줄 열쇠 = 세트 · 옛 조각 id = 든 세트 · 사라진 세트 409 · 옛 열쇠 → 세트 열쇠
- D04 확정 막음(덩이 2 이상 · 갈래) · 연결 = 세트 한 줄(아니면 거절) · 파일 판 2
- 거울 사례(`fixtures/d03_638_set_cases.json`) = TS `logic/D03_PlanAnalysis_Group.ts` 와 같은 답
"""

from __future__ import annotations

import asyncio
import json
import sys
from pathlib import Path
from uuid import uuid4

import pytest
from fastapi import HTTPException

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

from helper_plan_file import plan_of, use_plan  # noqa: E402

from D03_PlanAnalysis import D03_PlanAnalysis_Network as net  # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store  # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Link as link  # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Repository as repo  # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Router as d04  # noqa: E402

PLAN_ID = str(uuid4())
FOREST = {"role": "SYSTEM_ADMIN", "email": "root@x.kr", "company_id": 1, "user_id": 1}
MAIN = {"id": "main", "vertices": [[0.0, 0.0], [100.0, 0.0], [200.0, 0.0], [300.0, 0.0]]}


def _piece(pid, a, b, year=2027, no=None, line="main"):
    out = {"id": pid, "line": line, "from_m": a, "to_m": b, "year": year}
    return {**out, "set": no} if no else out


def _sets(plan):
    return [
        (s["key"], s["parts"], s["branch"], round(s["length_m"], 3), s["pieces"])
        for s in net.derive(plan)["sets"]
    ]


def test_old_pieces_numbered_by_line_then_distance():
    plan = plan_of(
        [MAIN],
        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
        [_piece("p3", 200.0, 300.0), _piece("p1", 0.0, 100.0, 2028), _piece("p2", 100.0, 200.0)],
    )
    assert _sets(plan) == [
        ("2027-1", 1, False, 100.0, ["p2"]),
        ("2027-2", 1, False, 100.0, ["p3"]),
        ("2028-1", 1, False, 100.0, ["p1"]),
    ]
    # 있는 번호 뒤로(새 조각 = 그 해 최댓값 + 1)
    plan["pieces"][1]["year"] = 2027
    plan["pieces"][2]["set"] = 5
    keys = {p["id"]: p["set_key"] for p in net.derive(plan)["pieces"]}
    assert keys == {"p1": "2027-6", "p2": "2027-5", "p3": "2027-7"}


def test_same_set_neighbours_merge_and_drop_mark():
    plan = plan_of(
        [MAIN],
        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
        [
            _piece("a", 0.0, 100.0, no=1),
            _piece("b", 100.0, 200.0, no=1),
            _piece("c", 200, 300, no=2),
        ],
    )
    flat = net.normalize(plan)
    assert flat["marks"] == [{"line": "main", "at_m": 200.0}]
    assert [(p["id"], p["from_m"], p["to_m"], p["set"]) for p in flat["pieces"]] == [
        ("a", 0.0, 200.0, 1),
        ("c", 200, 300, 2),
    ]
    assert _sets(plan) == [("2027-1", 1, False, 200.0, ["a"]), ("2027-2", 1, False, 100.0, ["c"])]
    assert net.normalize(flat) == flat  # 두 번 해도 같음
    # 다른 연도 같은 번호 = 안 합침
    plan["pieces"][1]["year"] = 2028
    assert len(net.normalize(plan)["pieces"]) == 3


def test_parts_branch_and_join():
    # 같은 세트가 다른 세트를 사이에 두고 끊김 → 2 덩이
    gap = plan_of(
        [MAIN],
        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
        [_piece("a", 0, 100, no=1), _piece("b", 100, 200, no=2), _piece("c", 200, 300, no=1)],
    )
    assert _sets(gap)[0] == ("2027-1", 2, False, 200.0, ["a", "c"])
    # 본선 가운데 꼭짓점에서 갈래 · 같은 세트 → 갈래
    fork = plan_of(
        [
            MAIN,
            {
                "id": "b",
                "vertices": [[100.0, 0.0], [100.0, 50.0]],
                "branch": {"line": "main", "vertex": 1},
            },
        ],
        [],
        [_piece("m", 0, 300, no=1), _piece("g", 0, 50, no=1, line="b")],
    )
    assert _sets(fork) == [("2027-1", 1, True, 350.0, ["m", "g"])]
    # 끝끼리 닿은 다른 선(뒤집힌 방향) = 한 덩이 · 이은 한 줄
    tail = {"id": "t", "vertices": [[400.0, 0.0], [300.0, 0.0]]}
    chain = plan_of(
        [MAIN, tail], [], [_piece("m", 0, 300, no=1), _piece("t", 0, 100, no=1, line="t")]
    )
    assert _sets(chain) == [("2027-1", 1, False, 400.0, ["m", "t"])]
    _, units = store.plan_parts(chain)
    assert units[0]["planned"] == [[0, 0], [100, 0], [200, 0], [300, 0], [400, 0]]
    assert (units[0]["from_m"], units[0]["to_m"]) == (0.0, 400.0)
    # 끊긴 세트 · 갈래 세트 = 이은 줄 없음
    assert store.plan_parts(gap)[1][0]["planned"] == []
    assert store.plan_parts(fork)[1][0]["planned"] == []


def test_find_unit_old_piece_key():
    plan = plan_of(
        [MAIN], [{"line": "main", "at_m": 100.0}], [_piece("p1", 0, 100), _piece("p2", 100, 300)]
    )
    _, units = store.plan_parts(plan)
    assert [u["id"] for u in units] == ["2027-1", "2027-2"]
    assert store.find_unit(units, "2027-2")["pieces"] == ["p2"]
    assert store.find_unit(units, "p2")["id"] == "2027-2"  # 옛 업체 열쇠
    assert store.find_unit(units, "없음") is None
    assert net.old_keys(plan) == {"p1": "2027-1", "p2": "2027-2"}


def test_file_version_two(tmp_path):
    assert store.VERSION == 2
    plan = plan_of([MAIN], [], [_piece("p1", 0, 300, no=1)])
    store.save_plan(tmp_path, plan)
    saved = json.loads(store.plan_path(tmp_path).read_text("utf-8"))
    assert saved["version"] == 2 and saved["pieces"][0]["set"] == 1
    # 판 1(set 없음) = 읽힘 · 보기에 번호
    store.plan_path(tmp_path).write_text(
        json.dumps({**saved, "version": 1, "pieces": [_piece("p1", 0, 300)]}), "utf-8"
    )
    old = store.read_plan(tmp_path)
    view = net._view(old, {"site": None, "year_start": 2027})
    assert (
        view["network"]["pieces"][0]["set"] == 1 and view["derived"]["sets"][0]["key"] == "2027-1"
    )
    assert net.Piece(**_piece("p1", 0, 1)).set is None  # 옛 화면 몸 = 받음
    with pytest.raises(ValueError):
        net.Piece(**_piece("p1", 0, 1), set=0)


class _Cursor:
    def __init__(self, held):
        self.held, self.log, self.rows = held, [], []

    async def __aenter__(self):
        return self

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

    async def execute(self, sql, args=()):
        self.log.append((sql, args))
        self.rows = [(k,) for k in self.held] if sql.startswith("SELECT DISTINCT") else []

    async def fetchall(self):
        return self.rows


class _Conn:
    def __init__(self, held):
        self.c = _Cursor(held)

    def cursor(self):
        return self.c


def test_store_status_lost_set_and_old_key_rewrite():
    conn = _Conn({"p1", "2027-2"})
    years = {2027: {"fingerprint": "f"}}
    asyncio.run(net._store_status(conn, PLAN_ID, {"2027-1", "2027-2"}, years, {"p1": "2027-1"}))
    moved = [a for s, a in conn.c.log if s.startswith("UPDATE plan_route_assignees")]
    assert moved == [("2027-1", PLAN_ID, "p1")]
    conn = _Conn({"p1", "2027-9"})
    with pytest.raises(HTTPException) as err:
        asyncio.run(net._store_status(conn, PLAN_ID, {"2027-1"}, years, {"p1": "2027-1"}))
    assert err.value.status_code == 409 and "2027-9" in err.value.detail


@pytest.fixture
def plan(monkeypatch):
    gap = plan_of(
        [MAIN],
        [{"line": "main", "at_m": 100.0}, {"line": "main", "at_m": 200.0}],
        [_piece("a", 0, 100, no=1), _piece("b", 100, 200, no=2), _piece("c", 200, 300, no=1)],
    )
    state = use_plan(monkeypatch, gap)

    async def project(_pid):
        return {"id": PLAN_ID, "name": "시험", "region": "용화", "project_kind": "PLAN"}

    async def none(*a):
        return []

    monkeypatch.setattr(repo, "get_plan_project", project)
    monkeypatch.setattr(repo, "list_assignees", none)
    monkeypatch.setattr(repo, "mark_confirmed", none)
    return state


def test_d04_units_and_finalize_blocked(plan):
    out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))
    assert [(r["piece_id"], r["parts"], r["saved"]) for r in out["pieces"]] == [
        ("2027-1", 2, False),
        ("2027-2", 1, True),
    ]
    with pytest.raises(HTTPException) as err:
        asyncio.run(d04.finalize_plan(PLAN_ID, d04.FinalizeRequest(years=[2027]), FOREST))
    assert err.value.status_code == 409 and "2027-1 2 덩이" in err.value.detail
    assert "예정노선에서" in err.value.detail


def test_link_needs_one_line(plan, monkeypatch):
    route = asyncio.run(link.piece_route(PLAN_ID, "a"))  # 옛 조각 id → 세트
    assert (route["set_key"], route["parts"], route["route"]["planned"]) == ("2027-1", 2, [])

    async def load(_id):
        return {"id": 7, "company_id": 5, "project_id": PLAN_ID, "piece_id": "a", **route}

    monkeypatch.setattr(link, "load_assignee", load)
    payload = link.LinkRequest(mode="new")
    with pytest.raises(HTTPException) as err:
        asyncio.run(link.link_plan_route(7, payload, {"company_id": 5, "user_id": 3}))
    assert (
        err.value.status_code == 409
        and "2027-1" in err.value.detail
        and "2 덩이" in err.value.detail
    )
    one = asyncio.run(link.piece_route(PLAN_ID, "2027-2"))
    assert one["route"]["planned"] == [[100.0, 0.0], [200.0, 0.0]] and one["parts"] == 1


CASES = ROOT / "resources" / "tester" / "fixtures" / "d03_638_set_cases.json"


@pytest.mark.skipif(not CASES.is_file(), reason="거울 사례 파일 = sub6 638-2 가 만듦")
def test_mirror_cases_match_ts():
    cases = json.loads(CASES.read_text("utf-8"))
    assert cases["cases"]
    for case in cases["cases"]:
        net.derive(case["network"], cases["year_start"])  # 틀린 사례 없음
        got = net.normalize(case["network"])["pieces"]
        keys = ("id", "line", "from_m", "to_m", "year", "set")
        assert [{k: p[k] for k in keys} for p in got] == case["expect"], case["name"]