Aislo/resources/tester/test_531_d04_export_finalize.py

# -*- coding: utf-8 -*-
"""531-1 · 531-2 D04 예정노선 내보내기 · 메일 첨부 · 업체 지정 · 확정.

- 만든 shp 를 B03 입력 분석(`analyze_input_metadata`)이 읽음 · 용화 shp 와 칸 · .prj · 인코딩 같음
- 메일 조립(SMTP 가짜) — 첨부 이름 · 크기 · 첨부 없는 기존 호출 그대로
- finalize / assignees 권한(산림청 아님 403) · 보낸 업체 건너뜀 · 메일 없음 · 분석 안 저장 409
"""

from __future__ import annotations

import asyncio
import io
import struct
import sys
import zipfile
from email import message_from_bytes
from pathlib import Path
from uuid import uuid4

import pytest
from fastapi import HTTPException
from openpyxl import load_workbook
from pyproj import Transformer

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

import common_util.common_util_email as email_mod  # noqa: E402
from B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata  # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Repository as repo  # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Router as router  # noqa: E402
from D04_PlanConfirm.D04_PlanConfirm_Email import build_completion_mail  # noqa: E402
from D04_PlanConfirm.D04_PlanConfirm_Export import (  # noqa: E402
    YearRoute,
    build_export,
    company_zip,
)

YONGHWA = ROOT / "resources/knowledge/original/실무문서/2026년 임도 용화/예정노선"
YONGHWA_BASE = YONGHWA / "대상지_일월.용화.2.2"  # 이름에 점이 있어 with_suffix 못 씀

# 5186(중부원점) 좌표의 두 해 — 2026 끝 = 2027 시작
Y2026 = [(210000.0, 380000.0), (210100.0, 380050.0), (210250.0, 380040.0)]
Y2027 = [(210250.0, 380040.0), (210400.0, 380120.0)]


def _yh(ext: str) -> Path:
    return YONGHWA / f"대상지_일월.용화.2.2{ext}"


def _bundle(with_xlsx: bool = True):
    return build_export(
        "시험 프로젝트",
        "일월.용화",
        [YearRoute(2027, Y2027, 5186), YearRoute(2026, Y2026, 5186)],
        first_station_m=100.0,
        with_xlsx=with_xlsx,
    )


def _dbf_fields(blob: bytes) -> list[tuple]:
    count = (struct.unpack("<H", blob[8:10])[0] - 33) // 32
    out = []
    for i in range(count):
        d = blob[32 + i * 32 : 64 + i * 32]
        out.append((d[:11].rstrip(b"\0").decode("cp949"), chr(d[11]), d[16], d[17]))
    return out


def _write(tmp_path: Path, files: dict[str, bytes]) -> Path:
    for name, data in files.items():
        (tmp_path / name).write_bytes(data)
    return next(tmp_path / n for n in files if n.endswith(".shp"))


# ── 내보내기 ─────────────────────────────────────────────


def test_year_shapefile_matches_yonghwa_format(tmp_path):
    bundle = _bundle()
    files = bundle.by_year[2026]
    shp_name = next(n for n in files if n.endswith(".shp"))
    stem = shp_name[: -len(".shp")]
    assert stem.startswith("대상지_일월.용화.") and stem.endswith("_2026")

    # 칸: 용화 두 칸(이름 · 형 · 폭 · 소수) 그대로 + 덧칸
    fields = _dbf_fields(files[stem + ".dbf"])
    assert fields[:2] == _dbf_fields((_yh(".dbf")).read_bytes())
    assert fields[2:] == [("연도", "N", 4, 0), ("시점누가", "N", 11, 2), ("종점누가", "N", 11, 2)]
    # .prj 원문 · .cpg · 도형 종류(폴리라인 3)
    assert files[stem + ".prj"] == _yh(".prj").read_bytes()
    assert files[stem + ".cpg"] == _yh(".cpg").read_bytes() == b"949"
    assert struct.unpack("<i", files[stem + ".shp"][32:36])[0] == 3


def test_b03_analysis_accepts_exported_shapefile(tmp_path):
    bundle = _bundle()
    shp = _write(tmp_path, bundle.by_year[2026])
    meta = analyze_input_metadata(shp)
    ref = analyze_input_metadata(_yh(".shp"))

    assert meta["shape_type"] == ref["shape_type"]
    assert meta["part_count"] == 1 and meta["point_count"] == len(Y2026)
    assert meta["encoding"] == ref["encoding"] == "cp949"
    assert meta["crs_input"] == ref["crs_input"]  # .prj 원문 같음 → 같은 좌표계
    assert meta["missing_members"] == []
    assert meta["route_name"] == meta["attributes"]["대상지"]
    assert meta["route_name"].startswith("일월.용화.")
    # 5186 → 5179 로 바뀐 시점 좌표
    sx, sy = Transformer.from_crs(5186, 5179, always_xy=True).transform(*Y2026[0])
    assert meta["start_point"][:2] == pytest.approx([sx, sy], abs=1e-6)
    assert float(meta["attributes"]["연도"]) == 2026
    assert float(meta["attributes"]["시점누가"]) == pytest.approx(100.0)


def test_stations_chain_and_total(tmp_path):
    bundle = _bundle()
    l26, l27 = bundle.lengths_m[2026], bundle.lengths_m[2027]
    shp27 = _write(tmp_path, bundle.by_year[2027])
    attrs = analyze_input_metadata(shp27)["attributes"]
    assert float(attrs["시점누가"]) == pytest.approx(100.0 + l26, abs=0.01)  # 직전 종점 = 시점
    assert float(attrs["종점누가"]) == pytest.approx(100.0 + l26 + l27, abs=0.01)

    total_tmp = tmp_path / "all"
    total_tmp.mkdir()
    meta = analyze_input_metadata(_write(total_tmp, bundle.all))
    assert meta["point_count"] == len(Y2026) + len(Y2027) - 1  # 겹친 경계점 하나 뺌
    assert bundle.lengths_m[0] == pytest.approx(l26 + l27, abs=0.01)
    assert meta["attributes"].get("연도") in (
        None,
        "",
        "****",
    )  # 전체는 연도 칸 비움(dBase 널 표기)


def test_explicit_stations_win_and_short_route_rejected():
    routes = [YearRoute(2026, Y2026, 5186, start_station_m=500.0, end_station_m=900.0)]
    rows = build_export("p", "r", routes, with_xlsx=False)
    meta = {}
    import tempfile

    with tempfile.TemporaryDirectory() as tmp:
        meta = analyze_input_metadata(_write(Path(tmp), rows.by_year[2026]))
    assert float(meta["attributes"]["시점누가"]) == 500.0
    assert float(meta["attributes"]["종점누가"]) == 900.0
    with pytest.raises(ValueError):
        build_export("p", "r", [YearRoute(2026, [(0.0, 0.0)], 5179)])


def test_zip_and_coordinate_table():
    bundle = _bundle()
    name, blob = company_zip("시험 프로젝트", 2026, bundle.by_year[2026])
    assert name == "시험 프로젝트_예정노선_2026.zip"
    with zipfile.ZipFile(io.BytesIO(blob)) as archive:
        names = archive.namelist()
    assert sum(n.endswith(s) for n in names for s in (".shp", ".shx", ".dbf", ".prj", ".cpg")) == 5
    assert "좌표표_시험 프로젝트.xlsx" in names

    sheet = load_workbook(io.BytesIO(bundle.all["좌표표_시험 프로젝트.xlsx"])).active
    rows = list(sheet.iter_rows(min_row=3, values_only=True))
    assert [r[0] for r in rows] == [2026, 2027]
    head = [c.value for c in next(sheet.iter_rows(min_row=2, max_row=2))]
    assert head[9:11] == ["시점 측점", "종점 측점"] and head[11:13] == [
        "시점누가(m)",
        "종점누가(m)",
    ]
    # 100 m → 5+0 · m 칸은 그대로
    assert rows[0][9] == "5+0" and rows[0][11] == pytest.approx(100.0)
    # 업체용 좌표표는 그 연도만
    one = load_workbook(io.BytesIO(bundle.by_year[2026]["좌표표_시험 프로젝트.xlsx"])).active
    assert [r[0] for r in one.iter_rows(min_row=3, values_only=True)] == [2026]


# ── 메일 ─────────────────────────────────────────────────


class _FakeSmtp:
    sent: list = []

    def __init__(self, *args, **kwargs):
        pass

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def starttls(self):
        pass

    def login(self, *args):
        pass

    def send_message(self, message):
        _FakeSmtp.sent.append(message)


def test_mail_attachment_assembly_and_old_calls(monkeypatch):
    monkeypatch.setattr(email_mod.smtplib, "SMTP", _FakeSmtp)
    _FakeSmtp.sent = []
    payload = b"PK\x03\x04" + b"x" * 5000

    email_mod._send_sync(
        "a@b.kr", "제목", "<p>본문</p>", [("시험_예정노선_2026.zip", payload, "application/zip")]
    )
    email_mod._send_sync("a@b.kr", "제목", "<p>본문</p>")  # 기존 호출 그대로
    with_file, plain = (message_from_bytes(m.as_bytes()) for m in _FakeSmtp.sent)

    parts = [p for p in with_file.walk() if p.get_filename()]
    assert [p.get_filename() for p in parts] == ["시험_예정노선_2026.zip"]
    assert parts[0].get_content_type() == "application/zip"
    assert parts[0].get_payload(decode=True) == payload
    assert not [p for p in plain.walk() if p.get_filename()]
    assert plain.get_content_type() == "multipart/alternative"


def test_send_email_disabled_ok_with_attachments(monkeypatch):
    monkeypatch.setattr(email_mod, "EMAIL_NOTIFICATION_ENABLED", False)
    assert asyncio.run(email_mod.send_email("a@b.kr", "s", "<p>h</p>")) is True
    assert asyncio.run(
        email_mod.send_email("a@b.kr", "s", "<p>h</p>", [("a.zip", b"1", "application/zip")])
    )


def test_completion_mail_body():
    subject, body = build_completion_mail(
        project_name="용화 <예정>",
        region="일월.용화",
        year=2026,
        length_m=2135.9,
        recipient="가나다건설",
        attachment_names=["용화_예정노선_2026.zip"],
    )
    assert "용화" in subject and "2026" in subject
    assert "2026년" in body and "2.136 km" in body and "용화_예정노선_2026.zip" in body
    assert "&lt;예정&gt;" in body  # 이스케이프


# ── API ──────────────────────────────────────────────────

FOREST = {"user_id": 1, "email": "a@korea.go.kr", "role": "MEMBER", "company_id": 7}
OUTSIDER = {"user_id": 2, "email": "b@company.co.kr", "role": "MASTER", "company_id": 8}


def _route_row(year: int, points, start: float, end: float):
    return {
        "year": year,
        "epsg": 5186,
        "route_json": {
            "planned": [list(p) + [0.0] for p in points],
            "start_chainage_m": start,
            "end_chainage_m": end,
        },
        "fingerprint": "f" * 40,
        "confirmed_at": None,
    }


@pytest.fixture
def fake_db(monkeypatch):
    state = {
        "years": [_route_row(2026, Y2026, 0.0, 330.0), _route_row(2027, Y2027, 330.0, 520.0)],
        "assignees": [
            {
                "id": 1,
                "year": 2026,
                "company_id": None,
                "company_name": "가나다건설",
                "email": "ga@x.kr",
                "notified_at": None,
                "linked_project_id": None,
            },
            {
                "id": 2,
                "year": 2026,
                "company_id": 9,
                "company_name": "라마",
                "email": "ra@x.kr",
                "notified_at": "2026-10-01",
                "linked_project_id": None,
            },
            {
                "id": 3,
                "year": 2027,
                "company_id": None,
                "company_name": "메일없음",
                "email": None,
                "notified_at": None,
                "linked_project_id": None,
            },
        ],
        "confirmed": [],
        "notified": [],
        "mails": [],
        "applied": None,
    }

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

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

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

    async def confirmed(_pid, ys):
        state["confirmed"] = ys

    async def notified(aid):
        state["notified"].append(aid)

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

    async def contact(cid):
        return {"id": cid, "name": "등록사", "email": "master@reg.kr"} if cid == 9 else None

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

    monkeypatch.setattr(repo, "get_plan_project", project)
    monkeypatch.setattr(repo, "list_year_routes", years)
    monkeypatch.setattr(repo, "list_assignees", assignees)
    monkeypatch.setattr(repo, "mark_confirmed", confirmed)
    monkeypatch.setattr(repo, "mark_notified", notified)
    monkeypatch.setattr(repo, "apply_assignee_changes", apply)
    monkeypatch.setattr(repo, "get_company_contact", contact)
    monkeypatch.setattr(router, "send_completion_mail", send)
    return state


def test_finalize_requires_forest_service(fake_db):
    with pytest.raises(HTTPException) as err:
        asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(), OUTSIDER))
    assert err.value.status_code == 403
    with pytest.raises(HTTPException) as err:
        asyncio.run(
            router.put_plan_assignees(uuid4(), router.AssigneesRequest(assignees=[]), OUTSIDER)
        )
    assert err.value.status_code == 403
    assert fake_db["confirmed"] == [] and fake_db["mails"] == []


def test_finalize_sends_zip_per_assignee(fake_db):
    out = asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(), FOREST))
    assert out["confirmed_years"] == [2026, 2027]
    states = {(n["year"], n["company_name"]): n["state"] for n in out["notifications"]}
    assert states == {
        (2026, "가나다건설"): "sent",
        (2026, "라마"): "already_sent",  # 이미 보냄 → 건너뜀
        (2027, "메일없음"): "no_email",
    }
    assert fake_db["notified"] == [1]
    ((to, subject, attachments),) = fake_db["mails"]
    assert to == "ga@x.kr" and "2026" in subject
    name, blob, mime = attachments[0]
    assert name == "시험_예정노선_2026.zip" and mime == "application/zip" and len(blob) > 1000
    with zipfile.ZipFile(io.BytesIO(blob)) as archive:
        assert any(n.endswith("_2026.shp") for n in archive.namelist())
        assert not any("_2027" in n for n in archive.namelist())  # 다른 연도는 안 샘

    # 다시 보내기
    fake_db["mails"].clear()
    again = asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(resend=True), FOREST))
    assert sum(n["state"] == "sent" for n in again["notifications"]) == 2


def test_finalize_unsaved_year_409(fake_db):
    fake_db["years"][1]["route_json"] = None
    with pytest.raises(HTTPException) as err:
        asyncio.run(router.finalize_plan(uuid4(), router.FinalizeRequest(years=[2027]), FOREST))
    assert err.value.status_code == 409
    assert fake_db["confirmed"] == []


def test_put_assignees_validation_and_diff(fake_db):
    def put(items):
        return asyncio.run(
            router.put_plan_assignees(uuid4(), router.AssigneesRequest(assignees=items), FOREST)
        )

    for bad in (
        router.AssigneeIn(year=2026, company_name="수기", email="메일아님"),
        router.AssigneeIn(year=2026, email="x@y.kr"),
        router.AssigneeIn(year=2030, company_name="없는해", email="x@y.kr"),
        router.AssigneeIn(year=2026, company_id=99),
    ):
        with pytest.raises(HTTPException) as err:
            put([bad])
        assert err.value.status_code == 400
    assert fake_db["applied"] is None

    # 2026: 가나다 유지(메일 바꿈) · 라마(등록사 9) 유지 · 새 수기 추가 / 2027 은 안 보냈으니 그대로
    put(
        [
            router.AssigneeIn(year=2026, company_name="가나다건설", email="GA@x.kr"),
            router.AssigneeIn(year=2026, company_id=9),
            router.AssigneeIn(year=2026, company_name="새업체", email="new@x.kr"),
        ]
    )
    deleted, updated, inserted = fake_db["applied"]
    assert deleted == []  # 2027 의 id 3 은 건드리지 않음
    assert {u["id"] for u in updated} == {1, 2}
    assert [i["company_name"] for i in inserted] == ["새업체"]
    assert next(u for u in updated if u["id"] == 2)["email"] == "master@reg.kr"

    put([router.AssigneeIn(year=2026, company_name="새업체", email="new@x.kr")])
    assert sorted(fake_db["applied"][0]) == [1, 2]  # 빠진 업체 지움