Aislo/resources/tester/test_531_3_plan_link.py

# -*- coding: utf-8 -*-
"""531-3 예정노선 → 임도 연결 · 업데이트 있음 · [내보내기] 받기.

- 연결 → B03 입력 `input/shp/` 파일 5 개 · input_files 행 5 · B03 분석 통과 · 원본 칸 · 연결 표시
- 입력이 다 찼고 지울 결과가 있으면 확인 409 · 파일 · 행 안 남김
- 지정 안 받은 회사 403 · 확정 안 된 연도 409 · 남의 회사 임도 403
- 지문 — 노선 바뀌면 update_available True · 같은 값으로 되돌리면 False
- 받은 예정노선 목록 · plan-source 가 지문으로 판정
- [내보내기] — 산림청 = 전 연도 + 전체 · 지정 회사 = 확정된 자기 연도 · 없으면 403
"""

from __future__ import annotations

import asyncio
import io
import json
import sys
import zipfile
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 B03_FileInput.B03_FileInput_Engine_Analyze import analyze_input_metadata  # noqa: E402
from B03_FileInput.B03_FileInput_Router_Helpers import OutputsWouldBeDiscarded  # noqa: E402
from common_util.common_util_route_geometry import find_planned_route_file  # 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
from D04_PlanConfirm.D04_PlanConfirm_Export import SHP_SUFFIXES  # noqa: E402

PLAN_ID = str(uuid4())
ROAD_ID = str(uuid4())
# 5186 좌표 두 해 — 2026 끝 = 2027 시작
P2026 = [[210000.0, 380000.0], [210100.0, 380050.0], [210250.0, 380040.0]]
P2027 = [[210250.0, 380040.0], [210400.0, 380120.0]]


def _route(points, start=0.0):
    length = sum(
        ((b[0] - a[0]) ** 2 + (b[1] - a[1]) ** 2) ** 0.5
        for a, b in zip(points, points[1:], strict=False)
    )
    return {
        "mode": "balanced",
        "planned": points,
        "length_m": length,
        "start_chainage_m": start,
        "end_chainage_m": start + length,
    }


def _assignee(**kw):
    row = {
        "id": 7,
        "project_id": PLAN_ID,
        "year": 2026,
        "company_id": 5,
        "linked_project_id": None,
        "name": "시험 예정노선",
        "region": "일월.용화",
        "road_type": "work",
        "design_speed_kph": 20,
        "route_start_m": 0.0,
        "epsg": 5186,
        "route_json": json.dumps(_route(P2026)),
        "confirmed_at": "2026-10-25 12:00:00",
    }
    row.update(kw)
    return row


SESSION = {"user_id": 3, "company_id": 5, "role": "USER", "email": "a@corp.com"}


class FakeCursor:
    def __init__(self, log, rows=None):
        self.log, self.rows = log, rows or []

    async def execute(self, sql, args=None):
        self.log.append((" ".join(sql.split()), args))

    async def fetchall(self):
        return self.rows

    async def fetchone(self):
        return None

    async def __aenter__(self):
        return self

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


class FakeConn:
    def __init__(self, log, rows=None):
        self.log, self.rows = log, rows

    def acquire(self):
        return self

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

    async def begin(self):
        self.log.append(("begin", None))

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

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

    async def __aenter__(self):
        return self

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


@pytest.fixture
def env(monkeypatch, tmp_path):
    """DB 가짜 — input_files 행 · 입력 유형 · 완료 판정 · 연결 기록을 쌓음."""
    state = {"rows": [], "types": set(), "complete": None, "links": [], "log": []}
    conn = FakeConn(state["log"])
    monkeypatch.setattr(link, "get_db_pool", lambda: conn)

    async def false(*a, **k):
        return False

    async def storage(*a, **k):
        return "1/3/x"

    async def create_input_file(connection, **kw):
        state["rows"].append(kw)
        return len(state["rows"])

    async def noop(*a, **k):
        return 0

    async def readiness(*a, **k):
        types = set(state["types"])
        for row in state["rows"]:
            ft = row["file_type"]
            types.add("route_prj" if ft == "prj" and "/input/shp/" in row["relative_path"] else ft)
        return types, None, None

    async def complete(connection, pid, las_free, confirm):
        state["complete"] = confirm
        if not confirm:
            raise OutputsWouldBeDiscarded(["노선 · 종단 설계"])
        return 99

    async def record(project_id, assignee, fingerprint, speed):
        state["links"].append((project_id, assignee["id"], fingerprint, speed))

    monkeypatch.setattr(link, "is_analysis_running", false)
    monkeypatch.setattr(link, "get_project_storage_relative_path", storage)
    monkeypatch.setattr(link, "resolve_stored_project_path", lambda s: str(tmp_path))
    monkeypatch.setattr(link, "create_input_file", create_input_file)
    monkeypatch.setattr(link, "supersede_previous_input_files", noop)
    monkeypatch.setattr(link, "get_project_input_readiness", readiness)
    monkeypatch.setattr(link, "_complete_file_input_if_ready", complete)
    monkeypatch.setattr(link, "record_link", record)
    monkeypatch.setattr(link, "_schedule_background_task", lambda coro, **k: coro.close())
    state["root"] = tmp_path
    return state


def _use(monkeypatch, assignee, road=None):
    async def load(_id):
        return assignee

    async def road_project(_pid):
        return road

    monkeypatch.setattr(link, "load_assignee", load)
    monkeypatch.setattr(link, "load_road_project", road_project)


def _link(mode="existing", **kw):
    payload = link.LinkRequest(mode=mode, project_id=kw.pop("project_id", ROAD_ID), **kw)
    return asyncio.run(link.link_plan_route(7, payload, SESSION))


ROAD = {"id": ROAD_ID, "company_id": 5, "project_kind": "ROAD"}


def test_link_writes_route_set_like_upload(env, monkeypatch):
    _use(monkeypatch, _assignee(), ROAD)
    result = _link()
    shp_dir = env["root"] / "B03_FileInput/input/shp"
    names = sorted(p.name for p in shp_dir.iterdir())
    assert len(names) == 5 and {Path(n).suffix for n in names} == set(SHP_SUFFIXES)
    assert all(n.startswith("대상지_일월.용화.") and "_2026." in n for n in names)
    assert len(env["rows"]) == 5
    assert {r["relative_path"] for r in env["rows"]} == {f"{link.ROUTE_DIR}/{n}" for n in names}
    assert env["complete"] is None  # 지형 없음 → 완료 판정 안 함 · WF1 없음
    assert result["analysis_started"] is False and result["project_id"] == ROAD_ID
    shp = find_planned_route_file(env["root"] / "B03_FileInput/input")
    meta = analyze_input_metadata(shp)
    assert meta["purpose"] == "planned_route" and meta["epsg"] == 5179
    assert meta["point_count"] == len(P2026)
    stage_meta = json.loads((env["root"] / "B03_FileInput/metadata.json").read_text("utf-8"))
    assert len(stage_meta["files"]) == 5
    fp = link.route_fingerprint(_assignee()["route_json"])
    assert env["links"] == [(ROAD_ID, 7, fp, 20)] and len(fp) == 40
    assert any("SUPERSEDED" in sql for sql, _ in env["log"] if isinstance(sql, str))


def test_complete_set_asks_before_discarding(env, monkeypatch):
    _use(monkeypatch, _assignee(), ROAD)
    env["types"] = {"las", "prj", "tfw"}
    response = _link()
    assert response.status_code == 409 and json.loads(response.body)["confirm"] == "replace_outputs"
    assert not (env["root"] / "B03_FileInput/input/shp").exists()
    assert ("rollback", None) in env["log"] and env["links"] == []
    env["rows"].clear()
    result = _link(confirm_replace=True)
    assert env["complete"] is True and result["analysis_started"] is True
    assert len(list((env["root"] / "B03_FileInput/input/shp").iterdir())) == 5


def test_other_company_and_unconfirmed_and_foreign_project(env, monkeypatch):
    _use(monkeypatch, _assignee(company_id=9), ROAD)
    with pytest.raises(HTTPException) as err:
        _link()
    assert err.value.status_code == 403
    _use(monkeypatch, _assignee(confirmed_at=None), ROAD)
    with pytest.raises(HTTPException) as err:
        _link()
    assert err.value.status_code == 409
    _use(monkeypatch, _assignee(), {**ROAD, "company_id": 9})
    with pytest.raises(HTTPException) as err:
        _link()
    assert err.value.status_code == 403
    _use(monkeypatch, _assignee(), {**ROAD, "project_kind": "PLAN"})
    with pytest.raises(HTTPException) as err:
        _link()
    assert err.value.status_code == 400
    assert env["rows"] == []


def test_new_mode_creates_road_project(env, monkeypatch):
    _use(monkeypatch, _assignee())
    made = {}

    async def create_project(**kw):
        made.update(kw)
        return {"project_id": ROAD_ID}

    monkeypatch.setattr(link, "create_project", create_project)
    result = _link(mode="new", project_id=None)
    assert result["project_id"] == ROAD_ID
    assert made["name"] == "시험 예정노선 2026" and made["region"] == "일월.용화"
    assert made["road_type"] == "work" and made["project_year"] == 2026
    assert made["company_id"] == 5 and len(env["rows"]) == 5


def _stage2_writes(log):
    return [
        args
        for sql, args in log
        if isinstance(sql, str) and "UPDATE project_workflow_stages" in sql
    ]


def test_new_mode_seeds_b05_design_speed(env, monkeypatch):
    """531-4 — 새 임도 = 예정노선 설계속도가 단계 2 params options 에 · B05 가 그 값을 씀."""
    from B05_Profile.B05_Profile_PageSettings import grade_options_of

    _use(monkeypatch, _assignee(design_speed_kph=30, road_type="main"))

    async def create_project(**kw):
        return {"project_id": ROAD_ID}

    monkeypatch.setattr(link, "create_project", create_project)
    _link(mode="new", project_id=None)
    (args,) = _stage2_writes(env["log"])
    params = json.loads(args[0])
    assert args[1:] == (ROAD_ID, 2) and params == {"options": {"design_speed_kph": 30}}
    # 자동설계 체인이 grade_class trunk(기본)로 params 를 다시 써도 이 값을 이어 넘김
    chained = {"options": {"grade_class": "trunk", "design_speed_kph": 30}}
    assert grade_options_of("main", chained).design_speed_kph == 30
    chain = (ROOT / "B03_FileInput/B03_FileInput_Service_Chain.py").read_text("utf-8")
    assert "'$.options.design_speed_kph'" in chain and "design_speed_kph=int(seeded_speed)" in chain


def test_existing_and_odd_speed_leave_stage_params(env, monkeypatch):
    """기존 프로젝트는 사용자 값 보존 · 20/30/40 밖 값은 안 넣음."""
    _use(monkeypatch, _assignee(design_speed_kph=30), ROAD)
    _link()
    assert _stage2_writes(env["log"]) == []

    async def create_project(**kw):
        return {"project_id": ROAD_ID}

    monkeypatch.setattr(link, "create_project", create_project)
    _use(monkeypatch, _assignee(design_speed_kph=25))
    _link(mode="new", project_id=None)
    assert _stage2_writes(env["log"]) == []


def test_fingerprint_update_and_revert():
    linked = link.route_fingerprint(_route(P2026))
    moved = [p[:] for p in P2026]
    moved[1][0] += 0.5
    assert link.update_available(json.dumps(_route(P2026)), linked) is False
    assert link.update_available(json.dumps(_route(moved)), linked) is True
    moved[1][0] -= 0.5  # 같은 값으로 되돌림
    assert link.update_available(json.dumps(_route(moved)), linked) is False
    assert link.update_available(None, linked) is False  # 지점 바뀌어 노선 빔 → 알림 없음
    assert link.update_available(_route(P2026), None) is False  # 연결 안 함


def test_assigned_list_flags_update(monkeypatch):
    linked = link.route_fingerprint(_route(P2026))
    moved = [[x + 1.0, y] for x, y in P2026]
    rows = [
        {
            "assignee_id": 7,
            "project_id": PLAN_ID,
            "project_name": "시험",
            "year": 2026,
            "linked_project_id": ROAD_ID,
            "linked_project_name": "임도",
            "source_fingerprint": linked,
            "route_json": json.dumps(_route(moved)),
        },
        {
            "assignee_id": 8,
            "project_id": PLAN_ID,
            "project_name": "시험",
            "year": 2027,
            "linked_project_id": None,
            "linked_project_name": None,
            "source_fingerprint": None,
            "route_json": json.dumps(_route(P2027)),
        },
    ]
    log = []
    monkeypatch.setattr(d01, "get_db_pool", lambda: FakeConn(log, rows))
    out = asyncio.run(d01.assigned_plan_routes(SESSION))["routes"]
    assert [r["update_available"] for r in out] == [True, False]
    assert "route_json" not in out[0] and out[0]["assignee_id"] == 7
    assert "confirmed_at IS NOT NULL" in log[0][0]


def test_plan_source(monkeypatch):
    linked = link.route_fingerprint(_route(P2026))
    row = {
        "source_fingerprint": linked,
        "plan_name": "시험",
        "assignee_id": 7,
        "year": 2026,
        "route_json": json.dumps(_route(P2026)),
    }

    async def one(sql, args):
        return row

    monkeypatch.setattr(link, "_fetch_one", one)
    out = asyncio.run(link.plan_source(ROAD_ID))
    assert out["linked"] and out["update_available"] is False and out["year"] == 2026
    row["route_json"] = json.dumps(_route([[x, y + 2.0] for x, y in P2026]))
    assert asyncio.run(link.plan_source(ROAD_ID))["update_available"] is True

    async def none(sql, args):
        return None

    monkeypatch.setattr(link, "_fetch_one", none)
    assert asyncio.run(link.plan_source(ROAD_ID)) == {"status": "success", "linked": False}


@pytest.fixture
def plan_repo(monkeypatch):
    async def project(_pid):
        return {
            "id": PLAN_ID,
            "name": "시험 예정노선",
            "region": "일월.용화",
            "project_kind": "PLAN",
            "route_start_m": 0.0,
        }

    async def years(_pid):
        return [
            {"year": 2026, "epsg": 5186, "route_json": _route(P2026), "confirmed_at": "x"},
            {"year": 2027, "epsg": 5186, "route_json": _route(P2027), "confirmed_at": None},
        ]

    async def assignees(_pid):
        return [
            {"id": 7, "year": 2026, "company_id": 5},
            {"id": 8, "year": 2027, "company_id": 5},
        ]

    monkeypatch.setattr(repo, "get_plan_project", project)
    monkeypatch.setattr(repo, "list_year_routes", years)
    monkeypatch.setattr(repo, "list_assignees", assignees)


def _zip_names(response):
    return sorted(zipfile.ZipFile(io.BytesIO(response.body)).namelist())


def test_export_forest_gets_all(plan_repo):
    forest = {"role": "USER", "email": "kim@forest.go.kr", "company_id": 1}
    names = _zip_names(asyncio.run(d04.export_plan(PLAN_ID, forest)))
    assert {n.split("/")[0] for n in names} == {"2026", "2027", "전체"}
    assert sum(n.endswith(".shp") for n in names) == 3


def test_export_shared_gets_own_confirmed_year(plan_repo):
    names = _zip_names(asyncio.run(d04.export_plan(PLAN_ID, SESSION)))
    assert {n.split("/")[0] for n in names} == {"2026"}  # 2027 = 확정 전 · 전체 없음
    with pytest.raises(HTTPException) as err:
        asyncio.run(d04.export_plan(PLAN_ID, {**SESSION, "company_id": 9}))
    assert err.value.status_code == 403


def test_summary_year_rows(plan_repo):
    out = asyncio.run(d04.get_plan_summary(PLAN_ID))
    first = out["years"][0]
    assert first["saved"] and first["start"]["x"] == P2026[0][0] and first["confirmed_at"] == "x"
    assert 120 < first["start"]["lon"] < 135 and first["end_chainage_m"] == pytest.approx(
        first["length_m"]
    )
    assert out["years"][1]["confirmed_at"] is None and len(out["assignees"]) == 2


def test_put_assignees_years_clears_empty_year(plan_repo, monkeypatch):
    """업체를 다 뺀 연도 — `years` 로 보내면 그 연도 행을 지움 · 안 보낸 연도는 그대로."""
    applied = {}

    async def apply(_pid, delete_ids, updates, inserts):
        applied.update(delete=delete_ids, updates=updates, inserts=inserts)

    monkeypatch.setattr(repo, "apply_assignee_changes", apply)
    forest = {"role": "SYSTEM_ADMIN", "email": "a@b.c", "company_id": 1}
    request = d04.AssigneesRequest(assignees=[], years=[2026])
    asyncio.run(d04.put_plan_assignees(PLAN_ID, request, forest))
    assert applied == {"delete": [7], "updates": [], "inserts": []}
    with pytest.raises(HTTPException) as err:
        asyncio.run(
            d04.put_plan_assignees(
                PLAN_ID, d04.AssigneesRequest(assignees=[], years=[2030]), forest
            )
        )
    assert err.value.status_code == 400