Aislo/resources/tester/test_532_1_project_copy.py

"""528-4 · 532-1 공용 프로젝트 복사 — 임시 storage 사본에서(요구 345 · 351).

- stability — 필요한 것만(LAS · 전처리 점군 · B07~B09 · 초기값 없음 · 지형 하나) · 지문 40자 ·
  복사본만으로 측점 수 · 설계선 = DB · 원본을 고쳐도 복사본 불변 · 고친 원본 = 다른 지문
- full(CLI 껍데기) — 새 id · 표마다 행 수 같음 · 지표면 id 다시 이음 · 파일 안 UUID 치환 ·
  원본 그대로(만든 행은 끝에 지움 · 연쇄 삭제)
"""

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

import aiomysql
import pytest

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

from helper_temp_storage import 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
from E01_StabRegister import E01_StabRegister_Source as src  # noqa: E402

#: 검증_sub1 — B06 까지 설계된 임도(읽기만)
SOURCE_ID = "916ea58c-0e35-4982-bf55-d8558eeb78b5"
ROOT = Path(__file__).resolve().parents[2]


async def _connect():
    return await aiomysql.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        db=DB_NAME,
        charset="utf8mb4",
        autocommit=True,
    )


async def _query(sql, args=()):
    connection = await _connect()
    try:
        async with connection.cursor(aiomysql.DictCursor) as cursor:
            await cursor.execute(sql, args)
            return await cursor.fetchall()
    finally:
        connection.close()


@pytest.fixture(scope="module")
def source_root():
    rows = asyncio.run(_query("SELECT storage_path FROM projects WHERE id = %s", (SOURCE_ID,)))
    if not rows:
        pytest.skip("검증 원본 프로젝트가 DB 에 없음")
    shared = Path(resolve_stored_project_path(rows[0]["storage_path"]))
    with temp_project_storage(shared) as root:
        yield root


async def _stability(target: Path) -> dict:
    connection = await _connect()
    try:
        return await cp.copy_project(
            connection, SOURCE_ID, scope="stability", new_id="e" * 36, project_root=target
        )
    finally:
        connection.close()


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("*.json")
        if path.is_file()
    }


@pytest.fixture
def work():
    """pytest tmp_path 정리가 이 PC 임시 폴더 권한에 걸려 tempfile 로."""
    path = Path(tempfile.mkdtemp(prefix="aislo_532_"))
    yield path
    shutil.rmtree(path, ignore_errors=True)


def test_stability_copy(source_root, work):
    before = _hashes(source_root)
    manifest = asyncio.run(_stability(work / "e1"))
    root = src.source_dir(work / "e1")

    assert manifest["scope"] == "stability" and manifest["source_project_id"] == SOURCE_ID
    assert len(manifest["source_fingerprint"]) == 40
    assert manifest["total_bytes"] < 40 * 1024 * 1024
    banned = ("B03_FileInput", "B04_PreProcess/processed", "B07_", "B08_", "B09_", "initial_")
    assert not [name for name in manifest["files"] if name.startswith(banned)]
    assert src.terrain_grid_path(root).is_file()  # 지형 하나뿐
    assert src.corridor_path(root) is not None
    # 파일 안 옛 UUID = 없음(동결 db.json · manifest 의 원본 기록 빼고)
    for path in root.rglob("*.json"):
        if path.name not in (cp.STABILITY_DB_NAME, cp.MANIFEST_NAME):
            assert SOURCE_ID not in path.read_text(encoding="utf-8"), path

    # 복사본만으로 측점 · 설계선 = DB 최신 노선
    rows = asyncio.run(
        _query(
            "SELECT chainage_m, data FROM cross_sections WHERE route_id = %s ORDER BY chainage_m",
            (manifest["route_id"],),
        )
    )
    stations = src.stations(root)
    assert len(stations) == len(rows) > 0
    for station, row in zip(stations, rows, strict=True):
        assert station["chainage_m"] == pytest.approx(row["chainage_m"])
        cross = src.cross_section(root, station["chainage_m"])
        assert cross["design"]["design_line"] == json.loads(row["data"])["design"]["design_line"]
        assert len(cross["samples"]) > 10

    # 원본 그대로 · 원본을 고쳐도 복사본 불변 · 같은 원본 = 같은 지문
    assert _hashes(source_root) == before
    first = stations[0]["chainage_m"]
    kept = src.cross_section(root, first)
    cross_file = next((source_root / "B06_Section" / "cross_sections").glob("*.json"))
    data = json.loads(cross_file.read_text(encoding="utf-8"))
    for sample in data["samples"]:
        sample["z"] = -999.0
    cross_file.write_text(json.dumps(data), encoding="utf-8")  # 임시 사본만 고침
    assert src.cross_section(root, first) == kept
    again = asyncio.run(_stability(work / "e2"))
    assert again["source_fingerprint"] != manifest["source_fingerprint"]


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))],
    )


COUNT_TABLES = (
    "routes",
    "surface_models",
    "input_files",
    "project_workflow_stages",
    "project_members",
    "cross_sections",
    "longitudinal_sections",
)


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


def test_full_clone_cli(source_root, monkeypatch):
    monkeypatch.setattr(cp, "_copy_folder", _copytree)
    before = _hashes(source_root)
    new_id = asyncio.run(_cli().main([SOURCE_ID, "시험_528_4_복제"]))
    try:
        assert new_id and new_id != SOURCE_ID
        assert _counts(new_id) == _counts(SOURCE_ID)
        models = {
            r["id"]
            for r in asyncio.run(
                _query("SELECT id FROM surface_models WHERE project_id = %s", (new_id,))
            )
        }
        routes = asyncio.run(
            _query("SELECT surface_model_id FROM routes WHERE project_id = %s", (new_id,))
        )
        assert {r["surface_model_id"] for r in routes} <= models
        new_root = source_root.parent / new_id
        assert (new_root / cp.MANIFEST_NAME).is_file()
        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
        assert _hashes(source_root) == before
    finally:
        if new_id:
            asyncio.run(_query("DELETE FROM projects WHERE id = %s", (new_id,)))