Aislo/resources/tester/test_795_4_b09_copy.py

# ruff: noqa: F811 — 임시 층 fixture(world)를 test_m02_labor_ref 에서 받아 씀
"""795-4 프로젝트(B09) 사본 규칙 — 만든 때 그대로 · [단가 갱신] 값만 · [재생성] 따로 ·
시점 섞임 고침(임시 storage · M01 은 메모리 덮개 · 정본 안 고침).

① 구조물 사본 단가 묶음 = M02 일위대가 사본 줄(만든 때) · 뒤에 M01 로직 · 노임이 바뀌어도
   사본 금액 그대로 · M01 에서 지워진 키 = 사본 줄 그대로 ② 사본 없는 옛 문서 = 옛 길(지금 M01)
③ [저장] 새 단가 줄 = 사본 기준 시점 값 · 없는 자원만 지금 값 + 표지 · 옛 책 줄 그대로
④ 연결 사본 `단가.json` 자원 줄 = 기준 시점 원천 ⑤ [단가 갱신] = 값만(사본 재적용 없음) ·
   [재생성] = 따로 누를 때만 ⑥ 미리 보기 꼴.
"""

from __future__ import annotations

import asyncio
import copy
import json
from uuid import uuid4

from B09_Estimation import B09_Estimation_Linked as linked
from B09_Estimation import B09_Estimation_Prices as prices
from B09_Estimation import B09_Estimation_Refresh as refresh
from B09_Estimation import B09_Estimation_Router as router
from M01_MasterData import M01_MasterData_Store as m01
from M02_MasterTemplete import M02_Item_Prices as item_prices
from M02_MasterTemplete import M02_Item_Source as source
from M02_MasterTemplete import M02_Sangeun_Copy as skc
from resources.tester.test_145_2_b09_labor_calc import _project, _snap
from resources.tester.test_795_1_sk_copy import _changed, _fake, _live
from resources.tester.test_795_2_sc_copy import LB, PCT, _double, _labor
from resources.tester.test_m02_labor_ref import world  # noqa: F401 — 같은 임시 층


def _json(value):
    return json.loads(json.dumps(value, ensure_ascii=False, default=str))


def _m01(monkeypatch, rows=(), drop=()):
    """M01 바꿈 흉내 — 색인(`_fake`) + 파일 묶음(단가 묶음이 읽음) 둘 다 · 정본 안 고침."""
    files = m01.loaded()[0]
    fake = _fake(monkeypatch, rows, drop)
    new = {str(r["키"]): r for r in rows}
    out = {}
    for name, data in files.items():
        slot = m01.items_key(data)
        got = [new.get(str(r.get("키")), r) for r in data.get(slot) or []]
        out[name] = {**data, slot: [r for r in got if str(r.get("키")) not in drop]}
    monkeypatch.setattr(m01, "loaded", lambda: (out, fake))


def _amount(labor_rows, bundle):
    got = item_prices.calc(labor_rows, bundle, {}, {})
    return _json(got["반영합"])


def test_bundle_is_m02_copy_then_frozen(world, monkeypatch):
    """① 사본 묶음 = M02 사본 줄 · M01 로직 · 노임 바뀜 · 지워짐에도 금액 그대로."""
    me = world["personal"]
    doc = _labor(me, [{"로직": PCT, "수량": {"손": 2}}])
    assert skc.has(doc)
    rows = source.labor_doc_rows(doc)
    was = _amount(rows, source.price_bundle(rows, me))  # 지금 M01 = 사본 만든 때
    bundle = source.price_bundle(rows, me)
    _m01(monkeypatch, [_changed(_live().logic(PCT)), _double(LB)])
    assert _amount(rows, bundle) == was  # 굳힌 묶음 = M01 안 읽음
    late = source.price_bundle(source.labor_doc_rows(doc), me)  # 구조물 지정이 M01 바뀐 뒤
    assert _amount(rows, late) == was  # M02 사본 = 만든 때
    old = {k: v for k, v in doc.items() if k != skc.KEY}  # ② 사본 없는 옛 문서 = 지금 M01
    old_rows = source.labor_doc_rows(old)
    assert _amount(old_rows, source.price_bundle(old_rows, me)) != was


def test_deleted_m01_key_kept_in_copy(world, monkeypatch):
    """① M01 에서 지워진 로직 키 = 사본 줄 그대로(`사본_남은줄.json`) · 같은 금액."""
    me = world["personal"]
    doc = _labor(me, [{"로직": PCT, "수량": {"손": 1}}])
    rows = source.labor_doc_rows(doc)
    was = _amount(rows, source.price_bundle(rows, me))
    _m01(monkeypatch, drop=[PCT])
    late = source.price_bundle(rows, me)
    assert [r["키"] for r in late[item_prices.COPY_FILE]["줄"]] == [PCT]
    assert _amount(rows, late) == was


def test_copy_rows_order_and_sk_origin():
    """사본 줄 모음 — 앞 문서 먼저 · 굳힌 SK 는 `원본` 사본."""
    a = {skc.KEY: {"로직": [{"키": "GC1", "v": 1}], "단가": [{"키": "LB1", "v": 1}]}}
    b = {"원본": {skc.KEY: {"로직": [{"키": "GC1", "v": 2}, {"키": "GC2"}], "단가": []}}}
    got = source.copy_rows([a, b, {"x": 1}, None])
    assert [(r["키"], r.get("v")) for r in got] == [("GC1", 1), ("LB1", 1), ("GC2", None)]
    assert item_prices.collect(["GF000219"], over=[]) == item_prices.collect(["GF000219"])


#: 841-1 단가 책 = 자재만 — 원목 · 미송각재
WOOD, BAR = "MT000401", "MT000338"


def _row(key, price):
    row = copy.deepcopy(prices._row(_live(), key))
    row["물가자료"] = price
    return item_prices.to_json(row)


def _price(book_row):
    return book_row["prices"]["물가자료"]


def test_book_new_rows_use_copy_basis():
    """③ [저장] 새 줄 = 사본 기준 시점 값 · 없는 자원 = 지금 값 + 표지 · 옛 책 줄 그대로."""
    frozen = {WOOD: _row(WOOD, 111)}
    first = _json(prices.extend_book(None, [WOOD], frozen))
    assert _price(first["rows"][WOOD]) == 111 and first["now_keys"] == []
    both = _json(prices.extend_book(first, [WOOD, BAR], frozen))
    assert both["rows"][WOOD] == first["rows"][WOOD]
    assert _price(both["rows"][BAR]) == float(prices._row(_live(), BAR)["물가자료"])
    assert both["now_keys"] == [BAR]
    inputs = prices.build_inputs(None, [WOOD, BAR], None, frozen)
    assert [_price(r) for r in inputs["materials"]] == [111, _price(both["rows"][BAR])]
    assert inputs["now"] == [BAR] and prices.build_inputs(both, [WOOD, BAR])["now"] == [BAR]
    old = _json(prices.refresh_book([WOOD, BAR]))  # 옛 책(사본 규칙 전) = 줄 · 값 그대로
    again = _json(prices.extend_book(old, [WOOD, BAR], frozen))
    assert again["rows"] == old["rows"] and again["now_keys"] == []
    fresh = _json(prices.refresh_book([WOOD, BAR], both))  # [단가 갱신] = 지금 값 · 표지 없음
    assert _price(fresh["rows"][WOOD]) != 111 and fresh["now_keys"] == []


def test_frozen_rows_from_linked_copy(tmp_path):
    """④ 연결 사본 `단가.json` 자원 줄 = 기준 시점 원천(자원 표만)."""
    root = _project(tmp_path / "proj", [(10.0, "SS000052")])
    _snap(root, "pipe@10.0", "SS000052", [{"로직": "GF000219", "수량": {"손": 1}}], {})
    got = linked.frozen_rows(root)
    assert got and all(k[:2] in skc.RES for k in got)
    bundle = json.loads(
        (linked.structures_dir(root) / "pipe@10.0" / "단가.json").read_text("utf-8")
    )
    inside = {r["키"]: r for d in bundle.values() if isinstance(d, dict) for r in d.get("줄") or []}
    assert all(inside[k] == row for k, row in got.items())


def _router(monkeypatch, tmp_path, calls):
    async def root(_):
        return tmp_path

    async def demand(*_):
        return [LB]

    async def payload(*_a, **_k):
        return {"user": {}, "notes": [], "sheets": {}}

    monkeypatch.setattr(router, "_project_root", root)
    monkeypatch.setattr(router, "_demand", demand)
    monkeypatch.setattr(router, "payload", payload)
    monkeypatch.setattr(router.store, "read_book", lambda _r: None)
    monkeypatch.setattr(router.store, "write_book", lambda _r, b: calls.append(("book", b)))
    monkeypatch.setattr(
        router.refresh,
        "reapply_changed",
        lambda *_a: calls.append("regen") or {"다시": ["A"], "막힘": []},
    )


def test_price_refresh_default_regen_separate(monkeypatch, tmp_path):
    """⑤ [단가 갱신] = 값만(사본 재적용 없음) · [재생성] = job 줄 때만."""
    calls = []
    _router(monkeypatch, tmp_path, calls)
    out = asyncio.run(router.refresh_prices(uuid4(), {"user": {}}, {}))
    assert "regen" not in calls and calls[-1][0] == "book"
    assert out["notes"][0].startswith("단가 갱신 — ")
    out = asyncio.run(router.refresh_prices(uuid4(), {"user": {}, "job": refresh.REGEN}, {}))
    assert calls.count("regen") == 1
    assert out["notes"][0].startswith("재생성 — 구조물 사본 1 다시 가져옴")


def test_price_preview_shape():
    """⑥ 미리 보기 — 바뀐 자원(사본 자원 줄 먼저 · 자재 단가 책) · 총괄 전 → 후."""
    old = _json(prices.refresh_book([WOOD, BAR]))
    old["rows"][WOOD] = {**old["rows"][WOOD], "prices": {"물가자료": 1}}
    new = _json(prices.refresh_book([WOOD, BAR], old))
    sheets = lambda g: {"total": {"subtotals": {"grand": g}}}  # noqa: E731
    copies = [{"키": LB, "이름": "사본 노임", "전": 1, "후": 2}]
    got = refresh.price_preview(old, new, sheets(10), sheets(20), 3, copies)
    assert got["할일"] == "단가갱신" and (got["전"], got["후"], got["고친칸"]) == (10, 20, 3)
    assert [(r["키"], r["전"]) for r in got["바뀐자원"]] == [(LB, 1), (WOOD, {"물가자료": 1})]
    assert got["바뀐자원"][1]["후"] == new["rows"][WOOD]["prices"]


def test_migrate_view_b09_copy(tmp_path):
    """옮기기 도구 B09 대조(보기만) — 구조물 사본 = 「그대로」 · 전 금액 · 파일 안 바꿈."""
    from M02_MasterTemplete import M02_Copy_Migrate as mig

    root = _project(tmp_path / "proj", [(10.0, "SS000052")])
    _snap(root, "pipe@10.0", "SS000052", [{"로직": "GF000219", "수량": {"손": 1}}], {})
    folder = linked.structures_dir(root) / "pipe@10.0"
    before = {p.name: p.read_bytes() for p in folder.iterdir()}
    got = mig.b09_check("구조물사본", folder)
    assert got["상태"] == "그대로" and got["전"]["기준합"]["계"] > 0
    assert {p.name: p.read_bytes() for p in folder.iterdir()} == before