Aislo/resources/tester/test_573_1_clone_de.py

"""573-1 복제 도구 D · E 갈래 — 임시 storage 사본에서(요구 404 · 405).

- D(예정노선) — plan_route_years · plan_route_assignees · 지표면 · 지형층 행 수 같음 ·
  설계 내용 = 폴더 파일(plan_network.json) 그대로 따라감 · 자동증가 id 새로 · 지형층 → 새 지표면 ·
  업체 연결 끊김 · 지형 models_plan · _derived 파일 그대로
- E(안정성평가) — 원본 칸(source_project_id 등) 그대로 · 단계 행 여섯 상태 그대로 ·
  E05 보고서 · _derived 파일 수 같음
- 둘 다 — 종류 판별 · 이름 그대로 · 파일 안 옛 UUID 없음 · 원본 행 · 파일 그대로 ·
  --dry-run 은 쓰지 않음 · 만든 복제본은 끝에 hard_delete_project(임시 사본 폴더째)
"""

import asyncio
import hashlib
import importlib.util
import json
import os
import shutil
import sys
from contextlib import contextmanager
from pathlib import Path

import aiomysql
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent))

from helper_temp_storage import _LINKED, temp_project_storage  # noqa: E402

from common_util import common_util_project_copy as cp  # noqa: E402
from common_util.common_util_storage import resolve_stored_project_path  # noqa: E402
from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER  # noqa: E402

#: 가이드 예정노선 · 가이드 안정성평가(읽기만)
PLAN_ID = "87d47800-9936-4a56-9850-276fbcb34cac"
STAB_ID = "088e173d-a8ab-4c51-badb-f5a95afc43bb"
ROOT = Path(__file__).resolve().parents[2]
#: 복제 안 뜨는 옛 D 표 — 지움 SQL 025 적용 전까지 남음(설계 내용 = 폴더 파일)
RETIRED = {"plan_networks", "plan_route_pieces"}
TABLES = (
    "surface_models",
    "project_members",
    "project_workflow_stages",
    *cp._PLAN_ALL,
)


async def _query(sql, args=()):
    connection = await aiomysql.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        db=DB_NAME,
        charset="utf8mb4",
        autocommit=True,
    )
    try:
        async with connection.cursor(aiomysql.DictCursor) as cursor:
            await cursor.execute(sql, args)
            return await cursor.fetchall()
    finally:
        connection.close()


def _q(sql, args=()):
    return asyncio.run(_query(sql, args))


def _project(project_id):
    rows = _q("SELECT * FROM projects WHERE id = %s", (project_id,))
    return rows[0] if rows else None


def _counts(project_id):
    return {
        table: _q(f"SELECT COUNT(*) AS n FROM {table} WHERE project_id = %s", (project_id,))[0]["n"]
        for table in TABLES
    }


def _hashes(root: Path) -> dict[str, str]:
    return {
        path.relative_to(root).as_posix(): hashlib.sha1(path.read_bytes()).hexdigest()
        for path in root.rglob("*")
        if path.is_file() and path.suffix == ".json"
    }


def _files(root: Path) -> set[str]:
    """정션 자리(시험 복사가 건너뜀) 빼고."""
    names = {p.relative_to(root).as_posix() for p in root.rglob("*") if p.is_file()}
    return {n for n in names if not n.startswith(tuple(f"{link}/" for link in _LINKED))}


def _cli():
    spec = importlib.util.spec_from_file_location(
        "tools_clone_project", ROOT / "db_management" / "tools_clone_project.py"
    )
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _copytree(source: Path, target: Path) -> None:
    """robocopy 대신 — 시험 사본의 정션(B03 · 전처리 점군)은 따라가지 않음."""
    shutil.copytree(
        source,
        target,
        ignore=lambda folder, names: [n for n in names if os.path.isjunction(Path(folder, n))],
    )


async def _hard_delete(project_id, actor):
    from B01_Dashboard.B01_Dashboard_Repository import soft_delete_project
    from common_util.common_util_project_delete import hard_delete_project
    from config.config_db import close_db_pool, init_db_pool

    await init_db_pool()
    try:
        # 영구 삭제는 휴지통 것만(592) — 먼저 휴지통으로
        await soft_delete_project(project_id, actor)
        return await hard_delete_project(project_id, actor)
    finally:
        await close_db_pool()


@contextmanager
def _clone(source_id, name, monkeypatch):
    """임시 storage 사본에서 복제 → (원본 사본 뿌리, 새 id) · 끝에 지움."""
    source = _project(source_id)
    if not source:
        pytest.skip("원본 프로젝트가 DB 에 없음")
    shared = Path(resolve_stored_project_path(source["storage_path"]))
    with temp_project_storage(shared) as root:
        monkeypatch.setattr(cp, "_copy_folder", _copytree)
        before_rows = _counts(source_id)
        before_files = _hashes(root)
        new_id = None
        try:
            # dry-run 은 쓰지 않음
            assert asyncio.run(_cli().main([source_id, "--dry-run"])) is None
            assert not (root.parent / name).exists()
            new_id = asyncio.run(_cli().main([source_id, name]))
            yield source, root, new_id
            # 원본 그대로
            assert _counts(source_id) == before_rows
            assert _hashes(root) == before_files
        finally:
            if new_id:
                assert asyncio.run(_hard_delete(new_id, source["user_id"]))
                assert _project(new_id) is None
                assert not (root.parent / new_id).exists()


def _common(source, root, new_id):
    new = _project(new_id)
    assert new_id != source["id"] and new["name"].startswith("시험_573_1")
    assert new["project_kind"] == source["project_kind"]
    assert new["storage_path"].endswith(new_id)
    assert _counts(new_id) == _counts(source["id"])
    new_root = root.parent / new_id
    manifest = json.loads((new_root / cp.MANIFEST_NAME).read_text(encoding="utf-8"))
    assert manifest["kind"] == cp.KIND_LABELS[source["project_kind"]]
    assert set(manifest["skipped_tables"]) <= RETIRED
    assert _files(new_root) - {cp.MANIFEST_NAME} == _files(root)
    for path in new_root.rglob("*.json"):
        if path.name != cp.MANIFEST_NAME:
            assert source["id"] not in path.read_text(encoding="utf-8"), path
    return new, new_root


def test_clone_plan(monkeypatch):
    with _clone(PLAN_ID, "시험_573_1_D", monkeypatch) as (source, root, new_id):
        new, new_root = _common(source, root, new_id)
        plan = Path("D03_PlanAnalysis") / "plan_network.json"
        assert (new_root / plan).read_bytes() == (root / plan).read_bytes()
        assert (new_root / "B04_PreProcess" / "models_plan").is_dir()
        assert (new_root / "_derived").is_dir()

        def rows(table, project_id, order):
            return _q(
                f"SELECT * FROM {table} WHERE project_id = %s ORDER BY {order}", (project_id,)
            )

        old, now = (
            rows("plan_route_years", PLAN_ID, "year"),
            rows("plan_route_years", new_id, "year"),
        )
        drop = ("project_id", "updated_at")
        assert [{k: v for k, v in r.items() if k not in drop} for r in now] == [
            {k: v for k, v in r.items() if k not in drop} for r in old
        ]
        old_a = rows("plan_route_assignees", PLAN_ID, "id")
        new_a = rows("plan_route_assignees", new_id, "id")
        assert not {r["id"] for r in old_a} & {r["id"] for r in new_a}
        assert [(r["piece_id"], r["company_name"]) for r in new_a] == [
            (r["piece_id"], r["company_name"]) for r in old_a
        ]
        assert all(r["linked_project_id"] is None for r in new_a)

        old_m = _q("SELECT id FROM surface_models WHERE project_id = %s", (PLAN_ID,))
        new_m = _q("SELECT id FROM surface_models WHERE project_id = %s", (new_id,))
        assert {r["id"] for r in new_m}.isdisjoint({r["id"] for r in old_m})
        layers = _q(
            "SELECT t.surface_model_id FROM terrain_layers t JOIN surface_models s"
            " ON s.id = t.surface_model_id WHERE s.project_id = %s",
            (new_id,),
        )
        assert len(layers) == 1


def test_clone_stability(monkeypatch):
    with _clone(STAB_ID, "시험_573_1_E", monkeypatch) as (source, root, new_id):
        new, new_root = _common(source, root, new_id)
        for column in ("source_project_id", "source_copied_at", "source_fingerprint"):
            assert new[column] == source[column]

        def stages(project_id):
            return _q(
                "SELECT stage_no, stage_key, state FROM project_workflow_stages"
                " WHERE project_id = %s ORDER BY stage_no",
                (project_id,),
            )

        assert len(stages(new_id)) == 6 and stages(new_id) == stages(STAB_ID)
        assert (new_root / "E05_StabReport" / "report.json").is_file()
        inner = json.loads(
            (new_root / cp.STABILITY_SOURCE_DIR / cp.MANIFEST_NAME).read_text(encoding="utf-8")
        )
        assert inner["project_id"] == new_id
        assert inner["source_project_id"] == source["source_project_id"]