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