Aislo/resources/tester/test_810_1_estx.py

"""810-1 지킴 — EST 원파일 파서(`M02_San_Estx`) · 원파일 여섯(실무 셋 + 경쟁사 셋 · 읽기만).

(a) 거울 — node `tools/est-test/import-estx.cjs` 결과와 표 이름 · 행 수 · 모든 칸 값 1:1(다름 0)
(b) 블록 줄합(저장 결과만 · 셈 없음) = 제목 표 저장 금액(TC NC GC JC SC) 100 %
(c) 엑셀 짝 대조는 812-1 엑셀 읽기기 골든(`test_812_1_xlsx_golden` · 옛 엑셀 엔진 걷음)
원파일 · node 없으면 skip(다른 PC).
"""

from __future__ import annotations

import base64
import json
import math
import shutil
import subprocess
import sys
from functools import lru_cache
from pathlib import Path

import pytest

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

from M02_MasterTemplete import M02_San_Estx as E  # noqa: E402

K = ROOT / "resources" / "knowledge" / "original"
P = K / "경쟁사 프로그램" / "EST Plus 2"
FILES = {
    "거창": K / "실무문서/01.산불거창(거창북상병곡)/01.내역서/00.25년_산불_거창(거창,북상,병곡,산8외) _단가변경.estx",  # noqa: E501
    "장수": K / "실무문서/02.산불 장수/01.내역서/00.단가변경_26년_산불_무주(장수,장수,덕산,산54-1).estx",  # noqa: E501
    "진안": K / "실무문서/03.산불진안/01.내역서/00.단가변경_26년_산불_무주(진안,백운,백암,산1-2).estx",  # noqa: E501
    "임도용역": P / "example/00.내역_2024년 임도시설 실시설계용역(12.18).estx",  # noqa: E501
    "통영": P / "example/01.내역서_통영섬숲복원사업(사량도) 실시설계(0830).estx",  # noqa: E501
    "main": P / "다산소프트/2026-07-main.estx",  # noqa: E501
}  # fmt: skip
#: 파일 → (산근 블록, 산근 줄, 일위 블록)
WANT = {
    "거창": (61, 2469 - 61, 14),
    "장수": (65, None, 3),
    "진안": (74, None, 14),
    "임도용역": (73, None, 3),
    "통영": (7, None, 27),
    "main": (1569, None, 1468),
}
NODE = r"""
const fs = require("fs");
const { importEstx } = require(process.argv[1] + "/tools/est-test/import-estx.cjs");
const r = importEstx(fs.readFileSync(process.argv[2]));
process.stdout.write(JSON.stringify(r.tables));
"""


@lru_cache(maxsize=2)
def data(tag: str) -> bytes:
    path = FILES[tag]
    if not path.exists():
        pytest.skip(f"원파일 없음 {path.name}")
    return path.read_bytes()


@lru_cache(maxsize=2)
def parsed(tag: str) -> dict:
    return E.read(data(tag))


def _norm(v):
    if isinstance(v, bytes):
        return {"binary": base64.b64encode(v).decode()}
    if isinstance(v, dict) and isinstance(v.get("binary"), bytes):
        return {"binary": base64.b64encode(v["binary"]).decode(), "kind": v["kind"]}
    return v


@pytest.mark.parametrize("tag", list(FILES))
def test_mirror_node(tag: str) -> None:
    if not shutil.which("node"):
        pytest.skip("node 없음")
    raw = data(tag)
    py = E.tables(raw)
    out = subprocess.run(
        ["node", "-e", NODE, str(ROOT), str(FILES[tag])], capture_output=True, check=True
    ).stdout
    nd = json.loads(out)
    nd.pop("Wonga", None)  # node 덧붙임(원가 서식 풀이) — 원 표 아님
    assert sorted(py) == sorted(nd)
    diff, c1 = [], 0
    for name, rows in nd.items():
        assert len(py[name]) == len(rows), name
        for i, (a, b) in enumerate(zip(py[name], rows, strict=True)):
            raw_a = a.get("_originalTextBytes") or {}
            raw_b = b.pop("_originalTextBytes", None) or {}
            for k, v in raw_b.items():  # node 가 못 푼 글 = 원 바이트 대조(파이썬 cp949 는 풂)
                mine = raw_a.get(k) or a[k].encode("cp949")
                assert base64.b64encode(mine).decode() == v, (name, i, k)
            a = {k: _norm(v) for k, v in a.items() if k != "_originalTextBytes" and k not in raw_b}
            b = {k: v for k, v in b.items() if k not in raw_b}
            if a != b:
                keys = [k for k in b if a.get(k) != b[k]]
                if all(_c1_same(a.get(k), b[k]) for k in keys):
                    c1 += 1
                    continue
                diff.append((name, i, keys))
    assert diff == []
    assert c1 == C1.get(tag, 0)


#: node TextDecoder(euc-kr) 가 CP949 확장 한글(「뀡」 등)을 C1 제어문자로 푼 행 수 — 원 바이트 같음
C1 = {"main": 5}


def _c1_same(mine, theirs) -> bool:
    """node 글의 C1 제어문자(0x80~0x9F) = 원 바이트 · 나머지 = CP949 → 파이썬 글 바이트와 같음."""
    if not (isinstance(mine, str) and isinstance(theirs, str)):
        return False
    raw = b"".join(c.encode("latin1") if "€" <= c <= "Ÿ" else c.encode("cp949") for c in theirs)
    return raw == mine.encode("cp949")


def _same(got: float, want: float | None) -> bool:
    w = want or 0
    return abs(math.floor(round(got, 6)) - w) < 0.5 or abs(round(got) - w) < 0.5


@pytest.mark.parametrize("tag", list(FILES))
def test_saved_amounts(tag: str) -> None:
    d = parsed(tag)
    n_cost, n_lines, n_head = WANT[tag]
    assert (len(d["산근"]), len(d["일위"])) == (n_cost, n_head)
    if n_lines is not None:
        assert sum(len(b["줄"]) for b in d["산근"]) == n_lines
    bad = [
        (key, b["코드"])
        for key in ("산근", "일위")
        for b in d[key]
        if not all(_same(b["줄합"][c], (b["저장"] or {}).get(c)) for c in "LMSR")
    ]
    assert bad == []
    assert all(not b["표지"] for key in ("산근", "일위") for b in d[key])


def test_structure_marks() -> None:
    """거창 첫 산근 — 기계 세 줄 · 연결 단가 · 변수 참조 · 합계 · 비목."""
    b = parsed("거창")["산근"][0]
    assert (b["코드"], b["이름"], b["저장"]["T"]) == ("D0000000003", "측구터파기", 5323)
    m = [ln for ln in b["줄"] if ln["기계"]]
    assert [ln["비목"] for ln in m[:4]] == ["L", "M", "S", "T"]
    assert len({ln["기계"] for ln in m[:4]}) == 1
    assert m[0]["자원"]["코드"] == "E0000000000" and m[0]["자원"]["종류"] == "기계"
    assert (m[0]["연결단가"], m[0]["변수"], m[0]["결과"]) == ([59020], [23.39], [2523.3])
    labor = next(ln for ln in b["줄"] if ln["연결"] == 6)
    assert (labor["자원"]["이름"], labor["품명"]) == ("보통인부", "보통인부")
    assert [ln["비목"] for ln in b["줄"] if ln["합계"]] == ["L", "M", "S", "T"]
    sub = [ln for d in [parsed("거창")] for b in d["산근"] for ln in b["줄"] if ln["산근"]]
    assert sub and all(ln["자원"]["종류"] == "산근" for ln in sub)