Aislo/resources/tester/test_506_3_b07_save_path.py

# -*- coding: utf-8 -*-
"""506-3 가 · 나(서버) — B07 저장 길 · 풀린 장 찌꺼기.

못박는 것
  1-A 도면 GET · 저장 PUT = `API_ANALYSIS_TIMEOUT_MS` · 저장 중 미리받기 쉼 · 끊긴 저장 = 목록 `confirmed` 로 판정.
  2   CAD 자동백업 칸 = `b07:{프로젝트}:{도면}` · 도각 편집 = `b07-frame:{프로젝트}` · [초기화]가 같은 열쇠로 지움.
  3-1 [초기화] 물음 글 = 프로젝트 도각은 그대로 · [도각 편집] → [기본 도각 복원] 안내.
  3-2 풀린 장(`confirmed` 거짓) = 저장본 json · 수량표 걷음 · 도면번호 남김.
  5-1 B07 쓰기 길 = `mark_stage(save)`(옛 `start_stage` 없음).
  7-3 `start_stage` 가 `DEV_UNLOCK:` 표식을 남김.
"""

from __future__ import annotations

import asyncio
import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = Path(HERE).parents[1]
B07 = ROOT / "B07_DesignDetail"
sys.path.insert(0, str(ROOT))


def _read(name: str) -> str:
    return (B07 / name).read_text(encoding="utf-8").replace("\r\n", "\n")


@pytest.fixture(scope="module")
def out():
    proc = subprocess.run(
        ["node", os.path.join(HERE, "helper_506_3_b07_save.cjs")],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=60,
    )
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_1a_한도():
    api = _read("B07_DesignDetail_Api_Fetch.ts")
    get = api[api.index("export function fetchDesignDrawing(") :]
    get = get[: get.index("\n}\n")]
    assert "API_ANALYSIS_TIMEOUT_MS" in get
    put = api[api.index("export function confirmDesignDrawing(") :]
    put = put[: put.index("\n}\n")]
    assert "API_ANALYSIS_TIMEOUT_MS" in put
    assert "timeoutMs: number = API_TIMEOUT_MS" in api


def test_1a_미리받기_쉼():
    page = _read("B07_DesignDetail_UI_Page.ts")
    assert "while (loadInFlight || isSaving())" in page


def test_1a_저장_판정(out):
    assert out["normal"]["savingSeen"] is True
    assert out["normal"]["savingAfter"] is False
    assert out["normal"]["calls"] == [
        "overlay+",
        "confirm:cross_s1",
        "overlay-",
        "toast:success",
        "saved",
    ]
    # 끊김 + 서버 저장됨 = 저장했습니다
    assert out["abortSaved"]["calls"] == [
        "overlay+",
        "confirm:cross_s1",
        "list",
        "overlay-",
        "toast:success",
        "saved",
    ]
    # 끊김 + 서버 저장 안 됨 = 머묾
    assert out["abortUnsaved"]["calls"] == [
        "overlay+",
        "confirm:cross_s1",
        "list",
        "toast:error",
        "overlay-",
    ]
    assert out["abortUnsaved"]["savingAfter"] is False
    # 끊김 아닌 오류 = 목록 안 읽음
    assert out["otherError"] == ["overlay+", "confirm:cross_s1", "toast:error", "overlay-"]


def test_2_자동백업_열쇠(out):
    assert out["scope"] == "b07:p1:cross_s1"
    assert "drop:OPEN_WEB_CAD__RECOVERY__b07:p1:cross_s1" in out["reset"]["calls"]
    page = _read("B07_DesignDetail_UI_Page.ts")
    assert "recoveryScopeOf(projectId, drawing.id)" in page
    assert "{ guide, recoveryScope }" in page
    assert "recoveryScope: `b07-frame:${projectId}`" in page


def test_3_1_초기화_글(out):
    asked = out["reset"]["asked"]
    assert "프로젝트 도각은 그대로" in asked
    assert "[도각 편집] → [기본 도각 복원]" in asked
    frame = (ROOT / "ui_template" / "cad_host" / "cad_host_frame_edit.ts").read_text(
        encoding="utf-8"
    )
    assert 'label: "기본 도각 복원"' in frame


def _project(tmp_path: Path) -> Path:
    from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import _design_root, _write_manifest

    stage = _design_root(tmp_path)
    (stage / "drawings").mkdir(parents=True)
    for drawing_id in ("cross_s1", "plan_a_1"):
        (stage / "drawings" / f"{drawing_id}.json").write_text("{}", encoding="utf-8")
    _write_manifest(
        tmp_path,
        {
            "drawings": {
                "cross_s1": {
                    "kind": "cross",
                    "confirmed": True,
                    "number": 7,
                    "file": "drawings/cross_s1.json",
                    "quantity_table": {"a": 1},
                    "quantity_tables": {"1": {}},
                },
                "plan_a_1": {"kind": "plan", "confirmed": True, "file": "drawings/plan_a_1.json"},
            }
        },
    )
    return stage


def test_3_2_풂_찌꺼기(tmp_path):
    from B07_DesignDetail.B07_DesignDetail_Router_Support import _invalidate_drawing
    from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import _read_manifest

    stage = _project(tmp_path)
    _invalidate_drawing(tmp_path, "cross_s1")
    entry = _read_manifest(tmp_path)["drawings"]["cross_s1"]
    assert entry == {"kind": "cross", "confirmed": False, "number": 7}
    assert not (stage / "drawings" / "cross_s1.json").exists()
    # 안 푼 장은 그대로
    assert (stage / "drawings" / "plan_a_1.json").exists()
    assert _read_manifest(tmp_path)["drawings"]["plan_a_1"]["file"] == "drawings/plan_a_1.json"


def test_3_2_초기화_찌꺼기(tmp_path):
    from B07_DesignDetail.B07_DesignDetail_Router_Reset import reset_design_files

    stage = _project(tmp_path)
    assert reset_design_files(tmp_path)["released"] == 2
    assert list((stage / "drawings").iterdir()) == []


def test_5_1_새_단계_규칙():
    router = _read("B07_DesignDetail_Router.py")
    scale = _read("B07_DesignDetail_Router_Support_Scale.py")
    assert "start_stage" not in router
    assert "start_stage" not in scale
    assert router.count('await mark_stage(str(project_id), 4, "save", cursor)') == 3
    assert scale.count('await mark_stage(str(project_id), 4, "save", cursor)') == 1


class _Cursor:
    def __init__(self) -> None:
        self.sql: list[str] = []

    async def execute(self, sql: str, args: tuple = ()) -> None:
        # pymysql 꼴 — 인자가 있으면 `%` 서식(`%%` = 글자 %)
        self.sql.append(sql % tuple(repr(a) for a in args) if args else sql)


def test_7_3_잠금풀기_표식_남김():
    from common_util.common_util_workflow_state import start_stage

    cursor = _Cursor()
    asyncio.run(start_stage(cursor, "p1", 4))
    first = cursor.sql[0]
    assert "message = CASE WHEN message LIKE 'DEV_UNLOCK:%' THEN message ELSE NULL END" in first
    assert "message = NULL" not in first