"""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"]