412 lines
16 KiB
Python
412 lines
16 KiB
Python
# -*- 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]
|
|
# 업체용 좌표표는 그 연도만
|
|
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 "<예정>" 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] # 빠진 업체 지움
|