From 4db52b55049fe3dc2bff4862c347b0ab060ae392 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 3 Nov 2026 12:57:40 +0900 Subject: [PATCH] =?UTF-8?q?feat(761-2):=20=EB=B3=B5=EC=A0=9C=20=EB=8F=84?= =?UTF-8?q?=EA=B5=AC=20=EC=9E=90=EB=A6=AC=20=EB=B0=94=EA=BF=94=20=EB=84=A3?= =?UTF-8?q?=EA=B8=B0=20--replace=20=E2=80=94=20=EB=8C=80=EC=83=81=20id=20?= =?UTF-8?q?=C2=B7=20=EC=9D=B4=EB=A6=84=20=C2=B7=20=EC=A3=BC=EC=9D=B8=20?= =?UTF-8?q?=C2=B7=20=EC=B0=B8=EC=97=AC=EC=9E=90=20=C2=B7=20=EC=A2=85?= =?UTF-8?q?=EB=A5=98=20=EC=9C=A0=EC=A7=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - common_util_project_copy_replace: 대상 옛 행(복제 표 목록) 지우고 원본 행으로 채움 · projects 는 유지 칸 밖만 원본 값 · storage 를 원본과 똑같이(robocopy /MIR) · 넣기 전 대상 행 덤프(--backup-dir) - common_util_project_copy: 행 넣기를 _insert_rows 로 떼어 full 복제 · 바꿔 넣기가 함께 씀 · robocopy 출력 글자 깨짐 무시 - 지킴 시험 test_761_2_clone_replace(유지 칸 · 지우는 표 = 복제 표 · id 다시 잇기) - test_532_1 낡은 단정 고침: 지표면 지워진 옛 노선의 NULL(ON DELETE SET NULL)을 새 id 밖 값으로 잘못 셈 - 실제 한 번: 가이드 → 임시 복제 → 그 자리에 --replace → 행 수 · 파일 같음 · 추천 4 꺼내짐 → 앱 영구 삭제 길로 지움 Co-Authored-By: Claude Opus 5.5 --- common_util/common_util_project_copy.py | 182 +++++++++--------- .../common_util_project_copy_replace.py | 163 ++++++++++++++++ db_management/tools_clone_project.py | 16 +- resources/tester/test_532_1_project_copy.py | 3 +- resources/tester/test_761_2_clone_replace.py | 118 ++++++++++++ 5 files changed, 392 insertions(+), 90 deletions(-) create mode 100644 common_util/common_util_project_copy_replace.py create mode 100644 resources/tester/test_761_2_clone_replace.py diff --git a/common_util/common_util_project_copy.py b/common_util/common_util_project_copy.py index 24e51e462..19c96812f 100644 --- a/common_util/common_util_project_copy.py +++ b/common_util/common_util_project_copy.py @@ -76,7 +76,8 @@ async def _insert(cursor, table, row: dict) -> int: return int(cursor.lastrowid) -def _copy_folder(source_root: Path, new_root: Path) -> None: +def _copy_folder(source_root: Path, new_root: Path, *, mirror: bool = False) -> None: + """mirror = 대상을 원본과 똑같이(/MIR · 원본에 없는 파일 지움 · 잠긴 파일 재시도 짧게).""" print(f"파일 복사 시작: {source_root.name} -> {new_root.name}") started = time.perf_counter() result = subprocess.run( @@ -84,7 +85,7 @@ def _copy_folder(source_root: Path, new_root: Path) -> None: "robocopy", str(source_root), str(new_root), - "/E", + *(["/MIR", "/R:2", "/W:2"] if mirror else ["/E"]), "/NFL", "/NDL", "/NJH", @@ -93,6 +94,7 @@ def _copy_folder(source_root: Path, new_root: Path) -> None: ], capture_output=True, text=True, + errors="replace", # robocopy 는 콘솔 코드 페이지로 씀 ) if result.returncode >= 8: raise RuntimeError(f"robocopy 실패 (코드 {result.returncode})") @@ -262,95 +264,12 @@ async def _insert_full( ) for user in member_ids: await _insert(cursor, "project_members", {"project_id": new_id, "user_id": user}) - - file_map: dict[int, int] = {} - for row in rows["input_files"]: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - file_map[old] = await _insert(cursor, "input_files", row) - - cloud_map: dict[int, int] = {} - for row in rows["processed_point_cloud"]: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - row["input_file_id"] = file_map.get(row["input_file_id"], row["input_file_id"]) - cloud_map[old] = await _insert(cursor, "processed_point_cloud", row) - - model_map: dict[int, int] = {} - for row in rows["surface_models"]: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - row["source_file_id"] = file_map.get(row["source_file_id"], row["source_file_id"]) - row["processed_cloud_id"] = cloud_map.get( - row["processed_cloud_id"], row["processed_cloud_id"] - ) - model_map[old] = await _insert(cursor, "surface_models", row) - - for row in rows["terrain_layers"]: - row = dict(row) - row.pop("id") - row["surface_model_id"] = model_map[row["surface_model_id"]] - await _insert(cursor, "terrain_layers", row) - - route_map: dict[int, int] = {} - for row in rows["routes"]: - row = dict(row) - old = row.pop("id") - row["project_id"] = new_id - row["surface_model_id"] = model_map.get( - row["surface_model_id"], row["surface_model_id"] - ) - route_map[old] = await _insert(cursor, "routes", row) - - for table in ( - "route_points", - "route_statistics", - "longitudinal_sections", - "cross_sections", - ): - for row in rows[table]: - row = dict(row) - row.pop("id") - row["route_id"] = route_map.get(row["route_id"], row["route_id"]) - if "project_id" in row: - row["project_id"] = new_id - await _insert(cursor, table, row) - - for row in rows["project_workflow_stages"]: - row = dict(row) - row.pop("id") - row["project_id"] = new_id - params = row.get("params") - if params: - # 단계 설정에 박힌 옛 id 를 새 id 로 바꾼다(입력 파일·지표면 모델). - data = json.loads(params) - if data.get("input_file_id") is not None: - old_file = int(data["input_file_id"]) - data["input_file_id"] = str(file_map.get(old_file, old_file)) - if data.get("surface_model_id") is not None: - old_model = int(data["surface_model_id"]) - data["surface_model_id"] = model_map.get(old_model, old_model) - row["params"] = json.dumps(data, ensure_ascii=False) - await _insert(cursor, "project_workflow_stages", row) - - for table in _PLAN_TABLES: - for row in rows[table]: - await _insert(cursor, table, {**row, "project_id": new_id}) - for row in rows["plan_route_assignees"]: - row = dict(row) - row.pop("id") - row["project_id"] = new_id - # 업체가 이은 임도는 원본 D 몫 — 복제본은 잇지 않은 채로 시작 - row["linked_project_id"] = None - await _insert(cursor, "plan_route_assignees", row) + model_map = await _insert_rows(cursor, rows, new_id) await connection.commit() print( - f"DB 복제 완료: 참여자 {len(member_ids)} · 입력파일 {len(file_map)}" + f"DB 복제 완료: 참여자 {len(member_ids)} · 입력파일 {len(rows['input_files'])}" f" · 지표면 {len(model_map)} · 지형층 {len(rows['terrain_layers'])}" - f" · 노선 {len(route_map)}" + f" · 노선 {len(rows['routes'])}" f" · 정점 {len(rows['route_points'])} · 횡단 {len(rows['cross_sections'])}" f" · 단계 {len(rows['project_workflow_stages'])}" f" · 예정노선 {' · '.join(f'{t} {len(rows[t])}' for t in _PLAN_ALL)}" @@ -358,6 +277,93 @@ async def _insert_full( return model_map +async def _insert_rows(cursor, rows: dict[str, list], new_id: str) -> dict[int, int]: + """projects · project_members 밖 복제 표 행을 new_id 로 넣고 자동증가 id 를 다시 이음 → + 지표면 id 짝(옛 → 새) · 트랜잭션은 부르는 쪽 몫(full 복제 · 자리 바꿔 넣기).""" + file_map: dict[int, int] = {} + for row in rows["input_files"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + file_map[old] = await _insert(cursor, "input_files", row) + + cloud_map: dict[int, int] = {} + for row in rows["processed_point_cloud"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + row["input_file_id"] = file_map.get(row["input_file_id"], row["input_file_id"]) + cloud_map[old] = await _insert(cursor, "processed_point_cloud", row) + + model_map: dict[int, int] = {} + for row in rows["surface_models"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + row["source_file_id"] = file_map.get(row["source_file_id"], row["source_file_id"]) + row["processed_cloud_id"] = cloud_map.get( + row["processed_cloud_id"], row["processed_cloud_id"] + ) + model_map[old] = await _insert(cursor, "surface_models", row) + + for row in rows["terrain_layers"]: + row = dict(row) + row.pop("id") + row["surface_model_id"] = model_map[row["surface_model_id"]] + await _insert(cursor, "terrain_layers", row) + + route_map: dict[int, int] = {} + for row in rows["routes"]: + row = dict(row) + old = row.pop("id") + row["project_id"] = new_id + row["surface_model_id"] = model_map.get(row["surface_model_id"], row["surface_model_id"]) + route_map[old] = await _insert(cursor, "routes", row) + + for table in ( + "route_points", + "route_statistics", + "longitudinal_sections", + "cross_sections", + ): + for row in rows[table]: + row = dict(row) + row.pop("id") + row["route_id"] = route_map.get(row["route_id"], row["route_id"]) + if "project_id" in row: + row["project_id"] = new_id + await _insert(cursor, table, row) + + for row in rows["project_workflow_stages"]: + row = dict(row) + row.pop("id") + row["project_id"] = new_id + params = row.get("params") + if params: + # 단계 설정에 박힌 옛 id 를 새 id 로 바꾼다(입력 파일·지표면 모델). + data = json.loads(params) + if data.get("input_file_id") is not None: + old_file = int(data["input_file_id"]) + data["input_file_id"] = str(file_map.get(old_file, old_file)) + if data.get("surface_model_id") is not None: + old_model = int(data["surface_model_id"]) + data["surface_model_id"] = model_map.get(old_model, old_model) + row["params"] = json.dumps(data, ensure_ascii=False) + await _insert(cursor, "project_workflow_stages", row) + + for table in _PLAN_TABLES: + for row in rows[table]: + await _insert(cursor, table, {**row, "project_id": new_id}) + for row in rows["plan_route_assignees"]: + row = dict(row) + row.pop("id") + row["project_id"] = new_id + # 업체가 이은 임도는 원본 D 몫 — 복제본은 잇지 않은 채로 시작 + row["linked_project_id"] = None + await _insert(cursor, "plan_route_assignees", row) + return model_map + + async def clone_full( connection, source_id: str, diff --git a/common_util/common_util_project_copy_replace.py b/common_util/common_util_project_copy_replace.py new file mode 100644 index 000000000..fe3b810a1 --- /dev/null +++ b/common_util/common_util_project_copy_replace.py @@ -0,0 +1,163 @@ +"""프로젝트 자리 바꿔 넣기 — 원본 내용을 대상 프로젝트 id 자리에 넣음 (요구 603 · PLAN 761-2). + +`common_util_project_copy` 의 full 복제와 같은 표 · 같은 id 다시 잇기 · 같은 파일 치환을 쓰되 +새 프로젝트를 만들지 않고 **대상 행을 그대로 둠**(id · 이름 · 주인 · 참여자 · 종류 · 회사 · +담당자 · 저장 자리 · 만든 때). 대상의 옛 행(복제 표 목록)은 지우고 원본 행으로 채움 · +storage 폴더는 원본과 똑같이 맞춤(robocopy /MIR — 원본에 없는 파일 지움). + +넣기 전 대상 행을 `backup_dir/<대상 id>.json` 에 덤프(되돌림 재료) · storage 는 백업 안 함. +원본은 읽기만. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import aiomysql + +from common_util.common_util_project_copy import ( + KIND_LABELS, + _copy_folder, + _fix_files, + _insert_rows, + _source_rows, + _tree_files, + _uncovered, + _write_manifest, +) +from common_util.common_util_storage import resolve_stored_project_path +from common_util.common_util_time import utc_now_text + +#: 대상 projects 행에서 그대로 두는 칸 — 나머지 칸은 원본 값 +KEEP_PROJECT_COLUMNS = ( + "id", + "name", + "user_id", + "company_id", + "project_kind", + "storage_path", + "created_at", + "deleted_at", + "pm_user_id", + "field_lead_user_id", + "designer_user_id", + "logo_asset_id", + "signature_asset_id", +) +#: 대상에서 지우는 표(자식 → 부모) — projects · project_members 는 둠 +ROUTE_CHILD_TABLES = ("cross_sections", "longitudinal_sections", "route_statistics", "route_points") +PROJECT_TABLES = ( + "routes", + "terrain_layers", + "surface_models", + "processed_point_cloud", + "input_files", + "project_workflow_stages", + "plan_route_years", + "plan_route_assignees", +) + + +def merged_project_row(source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]: + """대상 projects 행에 쓸 값 — 유지 칸은 대상 · 나머지는 원본(id 뺀 SET 몫).""" + return { + key: (target[key] if key in KEEP_PROJECT_COLUMNS else value) + for key, value in source.items() + if key != "id" and key != "updated_at" + } + + +async def _delete_target_rows(cursor, target_id: str, rows: dict[str, list]) -> dict[str, int]: + """대상 옛 행 지움(자식 → 부모) → 표 → 지운 행 수.""" + removed: dict[str, int] = {} + route_ids = [route["id"] for route in rows["routes"]] + for table in ROUTE_CHILD_TABLES: + removed[table] = 0 + if route_ids: + holders = ", ".join(["%s"] * len(route_ids)) + removed[table] = await cursor.execute( + f"DELETE FROM `{table}` WHERE route_id IN ({holders})", route_ids + ) + for table in PROJECT_TABLES: + if table == "terrain_layers": + model_ids = [model["id"] for model in rows["surface_models"]] + removed[table] = 0 + if model_ids: + holders = ", ".join(["%s"] * len(model_ids)) + removed[table] = await cursor.execute( + f"DELETE FROM terrain_layers WHERE surface_model_id IN ({holders})", model_ids + ) + continue + removed[table] = await cursor.execute( + f"DELETE FROM `{table}` WHERE project_id = %s", (target_id,) + ) + return removed + + +async def replace_project( + connection, source_id: str, target_id: str, *, backup_dir: Path +) -> dict[str, Any]: + """원본 내용을 대상 id 자리에 넣음 → 보고(표 행 수 · 지운 행 수 · 백업 자리).""" + if source_id == target_id: + raise ValueError("원본과 대상이 같음") + async with connection.cursor(aiomysql.DictCursor) as cursor: + rows = await _source_rows(cursor, source_id) + skipped = await _uncovered(cursor, source_id, rows) + old = await _source_rows(cursor, target_id) + left = await _uncovered(cursor, target_id, old) + source, target = rows["projects"][0], old["projects"][0] + if source.get("project_kind") != target.get("project_kind"): + raise ValueError( + f"종류가 다름: 원본 {source.get('project_kind')} · 대상 {target.get('project_kind')}" + ) + source_root = Path(resolve_stored_project_path(source["storage_path"])) + target_root = Path(resolve_stored_project_path(target["storage_path"])) + # /MIR 은 대상 폴더를 원본과 똑같이 지움 — 자리가 대상 id 폴더일 때만 + if target_root.name != target_id or target_root.resolve() == source_root.resolve(): + raise ValueError(f"대상 저장 자리가 이상함: {target_root}") + + if skipped or left: + print("⚠ 복제 목록 밖 표(안 뜸 · 대상 행 그대로):", {"원본": skipped, "대상": left}) + backup_dir.mkdir(parents=True, exist_ok=True) + backup = backup_dir / f"{target_id}.json" + backup.write_text(json.dumps(old, ensure_ascii=False, default=str), encoding="utf-8") + print(f"대상 행 백업: {backup}") + + # 1) 파일 먼저 — 실패하면 DB 를 건드리지 않은 채로 끝남 + _copy_folder(source_root, target_root, mirror=True) + # 2) DB — 지우기 · projects 칸 · 다시 넣기 한 트랜잭션 + await connection.begin() + try: + async with connection.cursor(aiomysql.DictCursor) as cursor: + removed = await _delete_target_rows(cursor, target_id, old) + values = merged_project_row(source, target) + sets = ", ".join(f"`{key}` = %s" for key in values) + await cursor.execute( + f"UPDATE projects SET {sets} WHERE id = %s", (*values.values(), target_id) + ) + model_map = await _insert_rows(cursor, rows, target_id) + await connection.commit() + except Exception: + await connection.rollback() + raise + # 3) 파일 안 옛 프로젝트 id · 초기값 스냅샷 지표면 id + _fix_files(target_root, source_id, target_id, model_map) + tables = {name: len(items) for name, items in rows.items()} + _write_manifest( + target_root, + { + "scope": "replace", + "kind": KIND_LABELS.get(source.get("project_kind"), source.get("project_kind")), + "source_project_id": source_id, + "project_id": target_id, + "copied_at": utc_now_text(), + "tables": tables, + "removed": removed, + "skipped_tables": skipped, + "files": _tree_files(target_root), + }, + ) + print(f"자리 바꿔 넣기 완료: {target['name']} ({target_id}) ← {source['name']}") + return {"tables": tables, "removed": removed, "skipped": skipped, "backup": str(backup)} diff --git a/db_management/tools_clone_project.py b/db_management/tools_clone_project.py index 788c07445..176eb48f1 100644 --- a/db_management/tools_clone_project.py +++ b/db_management/tools_clone_project.py @@ -25,6 +25,10 @@ [--owner <사용자 id>] [--members ] (없으면 원본 주인 · 참여자 그대로) 셈만(쓰지 않음 — 종류 · 표마다 행 수 · 목록 밖 표 · 폴더 크기) ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> - --dry-run + 자리 바꿔 넣기(대상 id · 이름 · 주인 · 참여자 · 종류 유지 · 대상 옛 행 지우고 원본으로 채움 · + storage 를 원본과 똑같이(/MIR) · 넣기 전 대상 행을 --backup-dir/<대상 id>.json 에 덤프) + ./venv/Scripts/python.exe db_management/tools_clone_project.py <원본 UUID> - \ + --replace <대상 UUID> [--backup-dir tmp/clone_replace_bk] storage 만(DB 는 이미 복제됨 — 다른 PC 의 storage 를 맞출 때) ./venv/Scripts/python.exe db_management/tools_clone_project.py \\ <원본 UUID> <새 UUID> --storage-only @@ -40,6 +44,7 @@ sys.path.append(str(Path(__file__).resolve().parent.parent)) import aiomysql from common_util.common_util_project_copy import clone_full, clone_storage_only, dry_run +from common_util.common_util_project_copy_replace import replace_project from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER @@ -59,10 +64,14 @@ def _args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument( "--dry-run", action="store_true", help="쓰지 않고 표마다 행 수 · 폴더 크기만" ) + parser.add_argument("--replace", metavar="TARGET_ID", help="원본 내용을 넣을 대상 UUID") + parser.add_argument( + "--backup-dir", default="tmp/clone_replace_bk", help="--replace 전 대상 행 덤프 폴더" + ) parser.add_argument("--owner", type=int, help="새 주인 사용자 id(없으면 원본 주인)") parser.add_argument("--members", help="참여자 사용자 id 쉼표 목록(없으면 원본 참여자)") args = parser.parse_args(argv) - if not args.dry_run and not args.new_name: + if not args.dry_run and not args.replace and not args.new_name: parser.error("new_name 필요") return args @@ -81,6 +90,11 @@ async def main(argv: list[str] | None = None) -> str | None: if args.dry_run: await dry_run(connection, args.source_id) return None + if args.replace: + await replace_project( + connection, args.source_id, args.replace, backup_dir=Path(args.backup_dir) + ) + return args.replace if args.storage_only: await clone_storage_only(connection, args.source_id, args.new_name) return None diff --git a/resources/tester/test_532_1_project_copy.py b/resources/tester/test_532_1_project_copy.py index 0d7ac1031..605a6fe65 100644 --- a/resources/tester/test_532_1_project_copy.py +++ b/resources/tester/test_532_1_project_copy.py @@ -191,7 +191,8 @@ def test_full_clone_cli(source_root, monkeypatch): 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 + # 지표면 지워진 옛 노선은 NULL(ON DELETE SET NULL) — 새 id 아닌 값만 막음 + assert {r["surface_model_id"] for r in routes} - {None} <= models new_root = source_root.parent / new_id assert (new_root / cp.MANIFEST_NAME).is_file() for path in new_root.rglob("*.json"): diff --git a/resources/tester/test_761_2_clone_replace.py b/resources/tester/test_761_2_clone_replace.py new file mode 100644 index 000000000..e9e9669d3 --- /dev/null +++ b/resources/tester/test_761_2_clone_replace.py @@ -0,0 +1,118 @@ +"""761-2 복제 도구 자리 바꿔 넣기(--replace) — 순수 부분(DB · 파일 안 씀). + +- 유지 칸: 대상 id · 이름 · 주인 · 회사 · 종류 · 저장 자리 · 담당자는 대상 값 · 나머지는 원본 +- 지우는 표 목록 = 복제 표 목록(projects · project_members 뺌)과 같음 — 표가 늘면 여기서 빨강 +- id 다시 잇기: 새 행이 새 노선 · 새 지표면 · 새 입력 파일 id 를 가리킴 · 단계 설정 id 도 +""" + +import asyncio +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from common_util import common_util_project_copy as cp # noqa: E402 +from common_util import common_util_project_copy_replace as rp # noqa: E402 + + +class _Cursor: + """execute 를 적어 두고 · INSERT 마다 새 자동증가 id · SELECT 는 projects 만 한 줄.""" + + def __init__(self): + self.inserts: list[tuple[str, dict]] = [] + self.selects: list[str] = [] + self.lastrowid = 1000 + self._last = "" + + async def execute(self, sql, args=()): + self._last = sql + if sql.startswith("INSERT INTO"): + table = sql.split("`")[1] + names = [part.strip("` ") for part in sql.split("(")[1].split(")")[0].split(",")] + self.inserts.append((table, dict(zip(names, args, strict=True)))) + self.lastrowid += 1 + else: + self.selects.append(sql) + return 0 + + async def fetchall(self): + return [{"id": "p", "name": "x"}] if "FROM projects" in self._last else [] + + +def test_merged_project_row_keeps_target_identity(): + source = { + "id": "src", + "name": "가이드", + "user_id": 3, + "company_id": 1, + "project_kind": "ROAD", + "storage_path": "storage/1/3/src", + "region": "용화", + "status": "WF3_ANALYZING", + "crs_epsg": 5178, + "pm_user_id": 3, + "updated_at": "t", + } + target = { + **{key: f"T-{key}" for key in source}, + "id": "dst", + "storage_path": "storage/1/9/dst", + } + merged = rp.merged_project_row(source, target) + assert "id" not in merged and "updated_at" not in merged + for key in ("name", "user_id", "company_id", "project_kind", "storage_path", "pm_user_id"): + assert merged[key] == target[key] + for key in ("region", "status", "crs_epsg"): + assert merged[key] == source[key] + + +def test_delete_tables_match_clone_tables(): + cursor = _Cursor() + rows = asyncio.run(cp._source_rows(cursor, "p")) + cloned = set(rows) - {"projects", "project_members"} + assert set(rp.ROUTE_CHILD_TABLES) | set(rp.PROJECT_TABLES) == cloned + + +def test_insert_rows_relinks_ids(): + rows = { + "input_files": [{"id": 1, "project_id": "src", "name": "a.las"}], + "processed_point_cloud": [{"id": 2, "project_id": "src", "input_file_id": 1}], + "surface_models": [ + {"id": 3, "project_id": "src", "source_file_id": 1, "processed_cloud_id": 2} + ], + "terrain_layers": [{"id": 4, "surface_model_id": 3}], + "routes": [{"id": 5, "project_id": "src", "surface_model_id": 3}], + "route_points": [{"id": 6, "route_id": 5}], + "route_statistics": [], + "longitudinal_sections": [{"id": 7, "route_id": 5, "project_id": "src"}], + "cross_sections": [], + "project_workflow_stages": [ + { + "id": 8, + "project_id": "src", + "stage_no": 2, + "params": json.dumps({"input_file_id": "1", "surface_model_id": 3}), + } + ], + **{table: [] for table in cp._PLAN_ALL}, + } + cursor = _Cursor() + model_map = asyncio.run(cp._insert_rows(cursor, rows, "dst")) + new = {table: row for table, row in cursor.inserts} + assert [table for table, _ in cursor.inserts][0] == "input_files" + assert "projects" not in new and "project_members" not in new + assert all(row.get("project_id", "dst") == "dst" for _, row in cursor.inserts) + assert all("id" not in row for _, row in cursor.inserts) + file_id, cloud_id, model_id = 1001, 1002, 1003 + assert new["processed_point_cloud"]["input_file_id"] == file_id + assert new["surface_models"]["source_file_id"] == file_id + assert new["surface_models"]["processed_cloud_id"] == cloud_id + assert model_map == {3: model_id} + assert new["terrain_layers"]["surface_model_id"] == model_id + assert new["routes"]["surface_model_id"] == model_id + route_id = 1005 + assert new["route_points"]["route_id"] == route_id + assert new["longitudinal_sections"]["route_id"] == route_id + params = json.loads(new["project_workflow_stages"]["params"]) + assert params == {"input_file_id": str(file_id), "surface_model_id": model_id}