Aislo/resources/tester/test_563_2_plan_pieces.py

# -*- coding: utf-8 -*-
"""563-2 예정노선 아래 단계 = 조각 단위(요구 391).

견본 = 같은 연도 조각 둘(본선 p1 · 갈래 p3, 2026) + 본선 p2(2027) · 업체 둘(p1 = 회사 5 · p3 = 회사 6).
- D04 요약 = 조각 줄 · 이름 「본선 0+0~12+10」 · 지정 회사 = 자기 조각 · 자기 행만
- 내보내기 = 조각마다 폴더 · 전체 = 선마다 레코드 · 지정 회사 = 확정된 자기 조각만
- 확정 메일 = 업체마다 자기 조각 zip(남의 조각 안 샘)
- 업체 저장 = 조각 id 검사 · 연도는 조각에서 · 같은 업체 두 조각 = 두 행
- 연결 · 가져오기 = 조각 꺾은선 shp · 받은 목록 = 조각 이름 · 길이
"""

from __future__ import annotations

import asyncio
import io
import json
import sys
import tempfile
import zipfile
from pathlib import Path
from uuid import uuid4

import geopandas as gpd
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 B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata  # noqa: E402
from D01_PlanRegister import D01_PlanRegister_Router as d01  # 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())
P1 = [[210000.0, 380000.0], [210200.0, 380150.0]]  # 본선 0 ~ 250
P2 = [[210200.0, 380150.0], [210300.0, 380400.0]]  # 본선 250 ~ 519.258
P3 = [[210200.0, 380150.0], [210330.0, 380150.0]]  # 갈래 0 ~ 130
L2 = (100.0**2 + 250.0**2) ** 0.5

FOREST = {"role": "SYSTEM_ADMIN", "email": "root@x.kr", "company_id": 1, "user_id": 1}
FIRM5 = {"role": "USER", "email": "a@ga.kr", "company_id": 5, "user_id": 3}
FIRM6 = {"role": "USER", "email": "b@na.kr", "company_id": 6, "user_id": 4}


def _piece(pid, line, name, year, start, end, planned):
    return {
        "piece_id": pid,
        "line_id": line,
        "line_name": name,
        "year": year,
        "from_m": start,
        "to_m": end,
        "length_m": end - start,
        "epsg": 5186,
        "route_json": {"planned": planned, "stations": [start, end]},
    }


PIECES = [
    _piece("p1", "main", "본선", 2026, 0.0, 250.0, P1),
    _piece("p3", "b1", "갈래 1", 2026, 0.0, 130.0, P3),
    _piece("p2", "main", "본선", 2027, 250.0, 250.0 + L2, P2),
]


def _assignee(aid, piece, year, company, name, email, notified=None):
    return {
        "id": aid,
        "year": year,
        "piece_id": piece,
        "company_id": company,
        "company_name": name,
        "email": email,
        "notified_at": notified,
        "linked_project_id": None,
    }


@pytest.fixture
def plan(monkeypatch):
    state = {
        "years": [
            {"year": 2026, "epsg": 5186, "route_json": {"planned": P1}, "confirmed_at": "x"},
            {"year": 2027, "epsg": 5186, "route_json": {"planned": P2}, "confirmed_at": None},
        ],
        "assignees": [
            _assignee(7, "p1", 2026, 5, "가", "a@ga.kr"),
            _assignee(8, "p3", 2026, 6, "나", "b@na.kr"),
        ],
        "mails": [],
        "applied": None,
    }

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

    async def years(_pid):
        return state["years"]

    async def pieces(_pid):
        return PIECES

    async def assignees(_pid):
        return state["assignees"]

    async def noop(*a):
        return None

    async def apply(_pid, d, u, i):
        state["applied"] = (d, u, i)

    async def contact(cid):
        return {"id": cid, "name": f"등록{cid}", "email": f"m{cid}@reg.kr"}

    async def send(*, to_email, subject, body, attachments):
        state["mails"].append((to_email, attachments))
        return True

    monkeypatch.setattr(repo, "get_plan_project", project)
    monkeypatch.setattr(repo, "list_year_routes", years)
    monkeypatch.setattr(repo, "list_pieces", pieces)
    monkeypatch.setattr(repo, "list_assignees", assignees)
    monkeypatch.setattr(repo, "mark_confirmed", noop)
    monkeypatch.setattr(repo, "mark_notified", noop)
    monkeypatch.setattr(repo, "apply_assignee_changes", apply)
    monkeypatch.setattr(repo, "get_company_contact", contact)
    monkeypatch.setattr(d04, "send_completion_mail", send)
    return state


def _names(response) -> list[str]:
    return sorted(zipfile.ZipFile(io.BytesIO(response.body)).namelist())


def test_summary_piece_rows_and_firm_view(plan):
    out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))
    rows = out["pieces"]
    assert [(r["piece_id"], r["year"], r["label"]) for r in rows] == [
        ("p1", 2026, "본선 0+0~12+10"),
        ("p3", 2026, "갈래 1 0+0~6+10"),
        ("p2", 2027, "본선 12+10~25+19.3"),
    ]
    assert rows[1]["length_m"] == 130.0 and rows[1]["start"]["x"] == P3[0][0]
    assert rows[0]["confirmed_at"] == "x" and rows[2]["confirmed_at"] is None
    assert [y["year"] for y in out["years"]] == [2026, 2027]
    # 같은 연도 다른 업체 — 자기 조각 · 자기 행만
    for session, piece, aid in ((FIRM5, "p1", 7), (FIRM6, "p3", 8)):
        mine = asyncio.run(d04.get_plan_summary(PLAN_ID, session))
        assert [r["piece_id"] for r in mine["pieces"]] == [piece]
        assert [a["id"] for a in mine["assignees"]] == [aid]
        assert [y["year"] for y in mine["years"]] == [2026]
    assert "b@na.kr" not in str(asyncio.run(d04.get_plan_summary(PLAN_ID, FIRM5)))


def test_export_piece_folders_and_firm_only_own(plan):
    names = _names(asyncio.run(d04.export_plan(PLAN_ID, FOREST)))
    assert {n.split("/")[0] for n in names} == {
        "2026_본선_p1",
        "2026_갈래1_p3",
        "2027_본선_p2",
        "전체",
    }
    assert sum(n.endswith(".shp") for n in names) == 4
    with zipfile.ZipFile(io.BytesIO(asyncio.run(d04.export_plan(PLAN_ID, FOREST)).body)) as z:
        with tempfile.TemporaryDirectory() as tmp:
            for n in z.namelist():
                if n.startswith("전체/") and not n.endswith(".xlsx"):
                    (Path(tmp) / n.split("/")[1]).write_bytes(z.read(n))
            whole = gpd.read_file(next(Path(tmp).glob("*.shp")))
    # 전체 = 선마다 한 레코드 — 본선(p1 + p2 이음) · 갈래
    assert len(whole) == 2
    assert sorted(round(g.length, 1) for g in whole.to_crs(5186).geometry) == [
        130.0,
        round(250.0 + L2, 1),
    ]
    firm = _names(asyncio.run(d04.export_plan(PLAN_ID, FIRM6)))
    assert {n.split("/")[0] for n in firm} == {"2026_갈래1_p3"}
    plan["years"][0]["confirmed_at"] = None  # 확정 전 = 받을 것 없음
    with pytest.raises(HTTPException) as err:
        asyncio.run(d04.export_plan(PLAN_ID, FIRM6))
    assert err.value.status_code == 403


def test_finalize_mails_own_piece_zip(plan):
    out = asyncio.run(d04.finalize_plan(PLAN_ID, d04.FinalizeRequest(years=[2026]), FOREST))
    assert [(n["piece_id"], n["state"]) for n in out["notifications"]] == [
        ("p1", "sent"),
        ("p3", "sent"),
    ]
    for (to, attachments), piece in zip(plan["mails"], ("본선_p1", "갈래1_p3"), strict=True):
        name, blob, _ = attachments[0]
        assert name == f"시험_예정노선_2026_{piece}.zip"
        shp = [n for n in zipfile.ZipFile(io.BytesIO(blob)).namelist() if n.endswith(".shp")]
        assert len(shp) == 1 and shp[0].endswith(f"_2026_{piece}.shp")


def test_put_assignees_by_piece(plan):
    def put(items, years=None):
        request = d04.AssigneesRequest(assignees=items, years=years)
        return asyncio.run(d04.put_plan_assignees(PLAN_ID, request, FOREST))

    with pytest.raises(HTTPException) as err:
        put([d04.AssigneeIn(year=2026, piece_id="없음", company_id=5)])
    assert err.value.status_code == 400
    # 같은 업체(회사 5)를 같은 연도 두 조각에 — 두 행 · 연도는 조각에서(보낸 2099 무시)
    put(
        [
            d04.AssigneeIn(year=2026, piece_id="p1", company_id=5),
            d04.AssigneeIn(year=2099, piece_id="p3", company_id=5),
        ],
        years=[2026],
    )
    deleted, updates, inserts = plan["applied"]
    assert deleted == [8] and [u["id"] for u in updates] == [7]
    assert [(i["piece_id"], i["year"], i["company_id"]) for i in inserts] == [("p3", 2026, 5)]


def test_legacy_year_assignee_sits_on_first_piece(plan):
    """조각이 생기기 전 연도 지정(piece_id 없음) — 그 연도 첫 조각에 붙어 보임."""
    plan["assignees"] = [_assignee(9, None, 2026, 5, "가", "a@ga.kr")]
    mine = asyncio.run(d04.get_plan_summary(PLAN_ID, FIRM5))
    assert [r["piece_id"] for r in mine["pieces"]] == ["p1"]


def test_link_uses_piece_route():
    row = {
        "id": 7,
        "year": 2026,
        "piece_id": "p3",
        "piece_epsg": 5186,
        "piece_json": json.dumps({"planned": P3, "stations": [0.0, 130.0]}),
        "from_m": 0.0,
        "to_m": 130.0,
        "piece_length_m": 130.0,
        "epsg": 5186,
        "route_json": json.dumps({"planned": P1, "length_m": 380.0}),
        "name": "2026 홍천 노선 계획",
        "region": "일월.용화",
    }
    unit = link.unit_row(row)
    assert unit["route_json"]["planned"] == P3 and unit["route_json"]["length_m"] == 130.0
    assert "piece_json" not in unit and "from_m" not in unit
    assert link.route_fingerprint(unit["route_json"]) != link.route_fingerprint(row["route_json"])
    files = link.year_files(unit)
    with tempfile.TemporaryDirectory() as tmp:
        for name, data in files.items():
            (Path(tmp) / name).write_bytes(data)
        meta = analyze_input_metadata(next(Path(tmp).glob("*.shp")))
    assert meta["point_count"] == 2 and float(meta["attributes"]["종점누가"]) == 130.0
    # 옛 연도 지정 = 연도 노선 그대로
    legacy = link.unit_row({**row, "piece_id": None})
    assert legacy["route_json"] == row["route_json"] and legacy["epsg"] == 5186


class _Cursor:
    def __init__(self, rows):
        self.rows = rows

    async def execute(self, sql, args=None):
        self.sql = sql

    async def fetchall(self):
        return self.rows

    async def __aenter__(self):
        return self

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


class _Conn:
    def __init__(self, rows):
        self.rows = rows

    def acquire(self):
        return self

    def cursor(self, *a):
        return _Cursor(self.rows)

    async def __aenter__(self):
        return self

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


def test_assigned_list_piece_label_length(monkeypatch):
    rows = [
        {
            "assignee_id": 8,
            "project_id": PLAN_ID,
            "project_name": "시험",
            "year": 2026,
            "piece_id": "p3",
            "piece_length_m": 130.0,
            "linked_project_id": None,
            "linked_project_name": None,
            "source_fingerprint": None,
            "route_json": json.dumps({"planned": P3}),
            "confirmed_at": "x",
        },
        {
            "assignee_id": 9,
            "project_id": PLAN_ID,
            "project_name": "옛",
            "year": 2025,
            "piece_id": None,
            "piece_length_m": None,
            "linked_project_id": None,
            "linked_project_name": None,
            "source_fingerprint": None,
            "route_json": json.dumps({"planned": P1, "length_m": 250.0}),
            "confirmed_at": "x",
        },
    ]

    async def pieces(_pid):
        return PIECES

    monkeypatch.setattr(d01, "get_db_pool", lambda: _Conn(rows))
    monkeypatch.setattr(repo, "list_pieces", pieces)
    out = asyncio.run(d01.assigned_plan_routes(FIRM6))["routes"]
    assert [(r["piece_id"], r["piece_label"], r["length_m"]) for r in out] == [
        ("p3", "갈래 1 0+0~6+10", 130.0),
        (None, None, 250.0),
    ]