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」 · 지정 회사 = 자기 조각 · 자기 행만
- 594-2 붙인 선 이름 = 조각 이름표 · zip 폴더 · 좌표표 xlsx 「선」 칸
- 내보내기 = 조각마다 폴더 · 전체 = 선마다 레코드 · 지정 회사 = 확정된 자기 조각만
- 확정 메일 = 업체마다 자기 조각 zip(남의 조각 안 샘)
- 업체 저장 = 조각 id 검사 · 연도는 조각에서 · 같은 업체 두 조각 = 두 행
- 연결 · 가져오기 = 조각 꺾은선 shp · 받은 목록 = 조각 이름 · 길이
- 노선 · 조각 · 연도 = 설계 내용 파일(PLAN 590-1 · `helper_plan_file`)
"""

from __future__ import annotations

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

import geopandas as gpd
import openpyxl
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 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}


LINES = [
    {"id": "main", "vertices": [P1[0], P1[1], P2[1]], "branch": None},
    {"id": "b1", "vertices": P3, "branch": {"line": "main", "vertex": 1}},
]
MARKS = [{"line": "main", "at_m": 250.0}]
PIECES = [
    {"id": "p1", "line": "main", "from_m": 0.0, "to_m": 250.0, "year": 2026},
    {"id": "p3", "line": "b1", "from_m": 0.0, "to_m": 130.0, "year": 2026},
    {"id": "p2", "line": "main", "from_m": 250.0, "to_m": 250.0 + L2, "year": 2027},
]
PLAN = plan_of(LINES, MARKS, PIECES)


def _assignee(aid, piece, year, company, name, email, notified=None):
    """DB 업체 줄 — `year` 는 낡은 사본(읽지 않음 · 연도 = 파일 조각)."""
    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 = {
        "file": use_plan(monkeypatch, PLAN, {2026: "x"}),
        "assignees": [
            _assignee(7, "p1", 2026, 5, "가", "a@ga.kr"),
            _assignee(8, "p3", 2029, 6, "나", "b@na.kr"),  # 낡은 연도 사본(요구 452 ⑤)
        ],
        "mails": [],
        "applied": None,
    }

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

    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_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]
    assert [(a["id"], a["year"]) for a in out["assignees"]] == [(7, 2026), (8, 2026)]
    # 같은 연도 다른 업체 — 자기 조각 · 자기 행만
    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["file"]["confirmed"].clear()  # 확정 전 = 받을 것 없음
    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_confirm_drops_when_file_changes(plan, monkeypatch):
    """파일만 바뀌고 연도 상태 지문이 낡으면 확정으로 안 보임(파일 쓰기 · DB 커밋 틈)."""
    from D03_PlanAnalysis import D03_PlanAnalysis_Store as store

    async def stale(_pid):
        return {2026: {"fingerprint": "낡음", "confirmed_at": "x"}}

    monkeypatch.setattr(store, "year_status", stale)
    out = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))
    assert all(r["confirmed_at"] is None for r in out["pieces"])


def test_link_uses_piece_route(plan):
    unit = asyncio.run(link.piece_route(PLAN_ID, "p3"))
    assert unit["year"] == 2026 and unit["epsg"] == 5186 and unit["confirmed_at"] == "x"
    assert unit["route"]["planned"] == P3 and unit["route"]["length_m"] == 130.0
    files = link.year_files({**unit, "name": "2026 홍천 노선 계획", "region": "일월.용화"})
    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
    gone = asyncio.run(link.piece_route(PLAN_ID, "없음"))
    assert gone["route"] is None and gone["year"] is None


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(plan, monkeypatch):
    """연도 · 길이 · 이름 = 파일 조각 · 확정 안 되고 연결도 없는 줄(2027 · 없는 조각)은 뺌."""
    row = {
        "assignee_id": 8,
        "project_id": PLAN_ID,
        "project_name": "시험",
        "piece_id": "p3",
        "linked_project_id": None,
        "linked_project_name": None,
        "source_fingerprint": None,
    }
    rows = [row, {**row, "assignee_id": 9, "piece_id": "p2"}, {**row, "piece_id": "없음"}]
    monkeypatch.setattr(d01, "get_db_pool", lambda: _Conn(rows))
    out = asyncio.run(d01.assigned_plan_routes(FIRM6))["routes"]
    assert [(r["piece_id"], r["year"], r["piece_label"], r["length_m"]) for r in out] == [
        ("p3", 2026, "갈래 1 0+0~6+10", 130.0),
    ]


def test_line_name_flows_to_labels_zip_xlsx(plan):
    """594-2 붙인 선 이름 = 조각 이름표 · zip 폴더 · 좌표표 xlsx 「선」 칸(빈칸 뺌)."""
    named = [{**LINES[0], "name": "용화 본선"}, LINES[1]]
    plan["file"]["plan"] = plan_of(named, MARKS, PIECES)
    rows = asyncio.run(d04.get_plan_summary(PLAN_ID, FOREST))["pieces"]
    assert [r["label"] for r in rows] == [
        "용화 본선 0+0~12+10",
        "갈래 1 0+0~6+10",
        "용화 본선 12+10~25+19.3",
    ]
    body = asyncio.run(d04.export_plan(PLAN_ID, FOREST)).body
    with zipfile.ZipFile(io.BytesIO(body)) as z:
        assert {n.split("/")[0] for n in z.namelist()} == {
            "2026_용화본선_p1",
            "2026_갈래1_p3",
            "2027_용화본선_p2",
            "전체",
        }
        book = next(n for n in z.namelist() if n.startswith("전체/") and n.endswith(".xlsx"))
        sheet = openpyxl.load_workbook(io.BytesIO(z.read(book))).active
    lines = [row[14] for row in sheet.iter_rows(min_row=3, values_only=True)]
    assert sorted(lines) == ["갈래1", "용화본선", "용화본선"]