Merge remote-tracking branches 'origin/main_laptop_1' and 'origin/sub_laptop_1' into sub_laptop_2
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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)}
|
||||
@@ -25,6 +25,10 @@
|
||||
[--owner <사용자 id>] [--members <id,id>] (없으면 원본 주인 · 참여자 그대로)
|
||||
셈만(쓰지 않음 — 종류 · 표마다 행 수 · 목록 밖 표 · 폴더 크기)
|
||||
./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
|
||||
|
||||
@@ -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"):
|
||||
|
||||
@@ -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}
|
||||
Reference in New Issue
Block a user