diff --git a/ddpui/api/transform_api.py b/ddpui/api/transform_api.py index 5c80b1d86..04aad3a65 100644 --- a/ddpui/api/transform_api.py +++ b/ddpui/api/transform_api.py @@ -1,4 +1,5 @@ import uuid +import math import shutil from pathlib import Path from datetime import datetime, timedelta @@ -27,6 +28,7 @@ ModelSrcInputsForMultiInputOp, validate_operation_config_v2, TerminateChainAndCreateModelPayload, + UpdateCanvasLayoutPayload, ) from ddpui.core.orgdbt_manager import DbtProjectManager from ddpui.utils.taskprogress import TaskProgress @@ -55,6 +57,24 @@ load_dotenv() logger = CustomLogger("ddpui") +MAX_CANVAS_LAYOUT_NODES = 5000 +MAX_CANVAS_COORDINATE = 10_000_000 +CANVAS_LOCK_DURATION = timedelta(minutes=2) + + +def _canvas_lock_response(lock: CanvasLock) -> LockCanvasResponseSchema: + """Serialize the backend lock record for the canvas lock API.""" + return LockCanvasResponseSchema( + lock_token=lock.lock_token, + expires_at=lock.expires_at.isoformat(), + locked_by=lock.locked_by.user.email, + ) + + +def _locked_workspace(orgdbt: OrgDbt) -> OrgDbt: + """Serialize lock lifecycle changes on the stable workspace row.""" + return OrgDbt.objects.select_for_update().get(pk=orgdbt.pk) + @transform_router.post("/dbt_project/") @has_permission(["can_create_dbt_workspace"]) @@ -255,7 +275,7 @@ def get_warehouse_datatypes(request): @transform_router.post("/dbt_project/canvas/lock/", response=LockCanvasResponseSchema) @has_permission(["can_edit_dbt_model"]) def lock_canvas(request): - """Lock canvas for editing""" + """Acquire or idempotently refresh the current user's canvas lock.""" orguser: OrgUser = request.orguser org = orguser.org @@ -264,40 +284,36 @@ def lock_canvas(request): if orgdbt is None: raise HttpError(404, "dbt workspace not setup") - # Check if already locked - try: - lock: CanvasLock = orgdbt.canvas_lock - if not lock.is_expired(): - if lock.locked_by == orguser: - # Refresh lock with 2-minute duration - lock.expires_at = timezone.now() + timedelta(minutes=2) - lock.save() - return LockCanvasResponseSchema( - lock_token=lock.lock_token, - expires_at=lock.expires_at.isoformat(), - locked_by=lock.locked_by.user.email, - ) - else: + # Locking the OrgDbt row closes the race where two requests both observe + # that the OneToOne CanvasLock row does not exist and then try to create it. + with transaction.atomic(): + orgdbt = _locked_workspace(orgdbt) + lock = ( + CanvasLock.objects.select_for_update() + .select_related("locked_by__user") + .filter(dbt=orgdbt) + .first() + ) + + if lock and not lock.is_expired(): + if lock.locked_by_id != orguser.id: raise HttpError(423, f"Canvas is already locked by {lock.locked_by.user.email}") - else: - # Delete expired lock + + lock.expires_at = timezone.now() + CANVAS_LOCK_DURATION + lock.save(update_fields=["expires_at", "updated_at"]) + return _canvas_lock_response(lock) + + if lock: lock.delete() - except CanvasLock.DoesNotExist: - pass - - # Create new lock with 2-minute duration - lock = CanvasLock.objects.create( - dbt=orgdbt, - locked_by=orguser, - lock_token=str(uuid.uuid4()), - expires_at=timezone.now() + timedelta(minutes=2), - ) - return LockCanvasResponseSchema( - lock_token=lock.lock_token, - expires_at=lock.expires_at.isoformat(), - locked_by=lock.locked_by.user.email, - ) + lock = CanvasLock.objects.create( + dbt=orgdbt, + locked_by=orguser, + lock_token=str(uuid.uuid4()), + expires_at=timezone.now() + CANVAS_LOCK_DURATION, + ) + lock = CanvasLock.objects.select_related("locked_by__user").get(pk=lock.pk) + return _canvas_lock_response(lock) @transform_router.put("/dbt_project/canvas/lock/refresh/") @@ -312,26 +328,27 @@ def refresh_canvas_lock(request): if orgdbt is None: raise HttpError(404, "dbt workspace not setup") - try: - lock: CanvasLock = orgdbt.canvas_lock + with transaction.atomic(): + orgdbt = _locked_workspace(orgdbt) + lock = ( + CanvasLock.objects.select_for_update() + .select_related("locked_by__user") + .filter(dbt=orgdbt) + .first() + ) + if lock is None: + raise HttpError(404, "No active lock found") if lock.is_expired(): raise HttpError(410, "Lock has expired") - if lock.locked_by != orguser: + if lock.locked_by_id != orguser.id: raise HttpError(403, "You can only refresh your own locks") - # Refresh lock with 2-minute duration - lock.expires_at = timezone.now() + timedelta(minutes=2) - lock.save() + lock.expires_at = timezone.now() + CANVAS_LOCK_DURATION + lock.save(update_fields=["expires_at", "updated_at"]) - logger.info(f"Refreshed lock for canvas") + logger.info("Refreshed lock for canvas") - return LockCanvasResponseSchema( - lock_token=lock.lock_token, - expires_at=lock.expires_at.isoformat(), - locked_by=lock.locked_by.user.email, - ) - except CanvasLock.DoesNotExist: - raise HttpError(404, "No active lock found") + return _canvas_lock_response(lock) @transform_router.delete("/dbt_project/canvas/lock/") @@ -346,19 +363,20 @@ def unlock_canvas(request): if orgdbt is None: raise HttpError(404, "dbt workspace not setup") - try: - lock = orgdbt.canvas_lock - if lock.locked_by != orguser: + with transaction.atomic(): + orgdbt = _locked_workspace(orgdbt) + lock = CanvasLock.objects.select_for_update().filter(dbt=orgdbt).first() + if lock is None: + return {"success": True} + if lock.locked_by_id != orguser.id: raise HttpError(403, "You can only unlock your own locks") lock.delete() - except CanvasLock.DoesNotExist: - pass # Already unlocked return {"success": True} # Canvas Lock Helper Function -def validate_canvas_lock(orguser: OrgUser, orgdbt): +def validate_canvas_lock(orguser: OrgUser, orgdbt: OrgDbt): """ Validate that the canvas is properly locked by the requesting user. Similar to dashboard lock validation but for canvas operations. @@ -372,20 +390,25 @@ def validate_canvas_lock(orguser: OrgUser, orgdbt): HttpError: 423 if canvas is locked by another user HttpError: 410 if canvas lock has expired """ - # Check canvas lock status - try: - lock: CanvasLock = orgdbt.canvas_lock + with transaction.atomic(): + orgdbt = _locked_workspace(orgdbt) + lock = ( + CanvasLock.objects.select_for_update() + .select_related("locked_by__user") + .filter(dbt=orgdbt) + .first() + ) + if lock is None: + raise HttpError( + 423, "Canvas is not locked. Please acquire a lock before making changes." + ) if lock.is_expired(): - # Clean up expired lock - lock.delete() raise HttpError(410, "Canvas lock has expired. Please acquire a new lock.") - elif lock.locked_by != orguser: + if lock.locked_by_id != orguser.id: raise HttpError(423, f"Canvas is locked by {lock.locked_by.user.email}") - # Lock is valid and owned by the user - refresh expiry and proceed - lock.expires_at = timezone.now() + timedelta(minutes=2) - lock.save() - except CanvasLock.DoesNotExist: - raise HttpError(423, "Canvas is not locked. Please acquire a lock before making changes.") + + lock.expires_at = timezone.now() + CANVAS_LOCK_DURATION + lock.save(update_fields=["expires_at", "updated_at"]) # ============================================================================== @@ -458,7 +481,7 @@ def delete_orgdbtmodel(request, model_uuid, canvas_lock_id: str = None, cascade: validate_canvas_lock(orguser, orgdbt) - orgdbt_model = OrgDbtModel.objects.filter(uuid=model_uuid).first() + orgdbt_model = OrgDbtModel.objects.filter(uuid=model_uuid, orgdbt=orgdbt).first() if not orgdbt_model: raise HttpError(404, "model not found") @@ -569,6 +592,78 @@ def get_dbt_project_DAG_v2(request): raise HttpError(500, f"Failed to generate DAG: {str(e)}") +@transform_router.put("/v2/dbt_project/graph/layout/") +@has_permission(["can_edit_dbt_model"]) +def put_canvas_layout(request, payload: UpdateCanvasLayoutPayload): + """Persist an atomic batch of top-left React Flow node coordinates.""" + orguser: OrgUser = request.orguser + orgdbt = orguser.org.dbt + + if not orgdbt: + raise HttpError(404, "dbt workspace not setup") + + if not payload.nodes: + raise HttpError(422, "at least one canvas node position is required") + + if len(payload.nodes) > MAX_CANVAS_LAYOUT_NODES: + raise HttpError( + 422, + f"canvas layout update cannot exceed {MAX_CANVAS_LAYOUT_NODES} nodes", + ) + + node_uuids = [item.uuid for item in payload.nodes] + if len(node_uuids) != len(set(node_uuids)): + raise HttpError(422, "canvas layout update contains duplicate node UUIDs") + + for item in payload.nodes: + coordinates = (item.position.x, item.position.y) + if not all(math.isfinite(coordinate) for coordinate in coordinates): + raise HttpError(422, "canvas coordinates must be finite numbers") + if any(abs(coordinate) > MAX_CANVAS_COORDINATE for coordinate in coordinates): + raise HttpError( + 422, + f"canvas coordinates must be within +/-{MAX_CANVAS_COORDINATE}", + ) + + with transaction.atomic(): + validate_canvas_lock(orguser, orgdbt) + canvas_nodes = list( + CanvasNode.objects.select_for_update().filter( + orgdbt=orgdbt, + uuid__in=node_uuids, + ) + ) + node_by_uuid = {node.uuid: node for node in canvas_nodes} + + # Scope the lookup to this workspace and return one generic error so a + # caller cannot use this endpoint to discover another org's node UUIDs. + if len(node_by_uuid) != len(node_uuids): + raise HttpError(422, "one or more canvas nodes were not found") + + now = timezone.now() + for item in payload.nodes: + canvas_node = node_by_uuid[item.uuid] + canvas_node.position_x = item.position.x + canvas_node.position_y = item.position.y + canvas_node.updated_at = now + + CanvasNode.objects.bulk_update( + canvas_nodes, + ["position_x", "position_y", "updated_at"], + ) + + return { + "updated": len(payload.nodes), + "nodes": [ + { + "uuid": str(item.uuid), + "position": {"x": item.position.x, "y": item.position.y}, + } + for item in payload.nodes + ], + } + + # V2 CRUD operations for CanvasNode @transform_router.post("/v2/dbt_project/models/{dbtmodel_uuid}/nodes/") @has_permission(["can_create_dbt_model"]) @@ -589,9 +684,9 @@ def post_create_src_model_node(request, dbtmodel_uuid: str): if not orgdbt: raise HttpError(404, "dbt workspace not setup") - try: - # TODO: apply canvas locking logic + validate_canvas_lock(orguser, orgdbt) + try: org_dbt_model = OrgDbtModel.objects.filter(uuid=dbtmodel_uuid, orgdbt=orgdbt).first() if not org_dbt_model: raise HttpError(404, "model not found") @@ -681,9 +776,9 @@ def post_add_operation_node(request, payload: CreateOperationNodePayload): if not orgdbt: raise HttpError(404, "dbt workspace not setup") - logger.info(f"creating operation: {payload.op_type}") + validate_canvas_lock(orguser, orgdbt) - # TODO: apply canvas locking logic + logger.info(f"creating operation: {payload.op_type}") try: main_input_node: CanvasNode = CanvasNode.objects.select_related("dbtmodel").get( @@ -822,9 +917,9 @@ def put_operation_node(request, node_uuid: str, payload: EditOperationNodePayloa if not orgdbt: raise HttpError(404, "dbt workspace not setup") - logger.info(f"updating operation: {payload.op_type}") + validate_canvas_lock(orguser, orgdbt) - # TODO: apply canvas locking logic + logger.info(f"updating operation: {payload.op_type}") # Fetch the operation node first, outside the main try block try: @@ -958,7 +1053,7 @@ def post_terminate_operation_node( if not orgdbt: raise HttpError(404, "dbt workspace not setup") - # TODO: apply canvas locking logic + validate_canvas_lock(orguser, orgdbt) try: terminal_node = CanvasNode.objects.get( @@ -1066,6 +1161,8 @@ def delete_canvas_node(request, node_uuid: str): if not orgdbt: raise HttpError(404, "dbt workspace not setup") + validate_canvas_lock(orguser, orgdbt) + try: canvas_node = CanvasNode.objects.get(uuid=node_uuid, orgdbt=orgdbt) dbtmodel = canvas_node.dbtmodel @@ -1171,6 +1268,8 @@ def sync_remote_dbtproject_to_canvas(request): "message": "dbt workspace is not of GIT type, skipping sync to canvas", } + validate_canvas_lock(orguser, orgdbt) + # Get warehouse try: warehouse_obj = OrgWarehouse.objects.get(org=org) diff --git a/ddpui/core/dbtautomation_service.py b/ddpui/core/dbtautomation_service.py index f0e2d1bbf..7f2e3a040 100644 --- a/ddpui/core/dbtautomation_service.py +++ b/ddpui/core/dbtautomation_service.py @@ -935,6 +935,11 @@ def convert_canvas_node_to_frontend_format( "name": canvas_node.name, "operation_config": canvas_node.operation_config, "output_columns": canvas_node.output_cols, + "position": ( + {"x": canvas_node.position_x, "y": canvas_node.position_y} + if canvas_node.position_x is not None and canvas_node.position_y is not None + else None + ), "dbtmodel": ( { "schema": canvas_node.dbtmodel.schema, diff --git a/ddpui/core/trial/dbt_clone.py b/ddpui/core/trial/dbt_clone.py index 3022a06b5..eea5e5120 100644 --- a/ddpui/core/trial/dbt_clone.py +++ b/ddpui/core/trial/dbt_clone.py @@ -57,6 +57,8 @@ def _copy_canvas(template_dbt: OrgDbt, trial_dbt: OrgDbt, model_map: dict) -> No operation_config=node.operation_config, output_cols=node.output_cols, dbtmodel=new_dbtmodel, + position_x=node.position_x, + position_y=node.position_y, ) node_map[node.id] = new_node diff --git a/ddpui/migrations/0177_canvasnode_position_x_canvasnode_position_y.py b/ddpui/migrations/0177_canvasnode_position_x_canvasnode_position_y.py new file mode 100644 index 000000000..6072e2360 --- /dev/null +++ b/ddpui/migrations/0177_canvasnode_position_x_canvasnode_position_y.py @@ -0,0 +1,24 @@ +# Generated by Django 4.2 on 2026-08-19 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + """Store persistent top-left coordinates for UI4T canvas nodes.""" + + dependencies = [ + ("ddpui", "0176_trialsignup"), + ] + + operations = [ + migrations.AddField( + model_name="canvasnode", + name="position_x", + field=models.FloatField(blank=True, null=True), + ), + migrations.AddField( + model_name="canvasnode", + name="position_y", + field=models.FloatField(blank=True, null=True), + ), + ] diff --git a/ddpui/models/canvas_models.py b/ddpui/models/canvas_models.py index 2b03951d5..02c15ff16 100644 --- a/ddpui/models/canvas_models.py +++ b/ddpui/models/canvas_models.py @@ -48,6 +48,11 @@ class CanvasNode(models.Model): OrgDbtModel, on_delete=models.SET_NULL, null=True ) # For MODEL/SOURCE nodes + # Canonical React Flow position. Coordinates represent the top-left corner + # of the node in flow space and are intentionally independent of pan/zoom. + position_x = models.FloatField(null=True, blank=True) + position_y = models.FloatField(null=True, blank=True) + # Timestamps created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) diff --git a/ddpui/schemas/dbt_workflow_schema.py b/ddpui/schemas/dbt_workflow_schema.py index 740ee7d31..bfafa60fb 100644 --- a/ddpui/schemas/dbt_workflow_schema.py +++ b/ddpui/schemas/dbt_workflow_schema.py @@ -1,5 +1,6 @@ from ninja import Field, Schema from typing import Union, Any, Literal, Optional +from uuid import UUID from pydantic import ConfigDict from ddpui.models.dbt_workflow import OrgDbtModel @@ -71,6 +72,26 @@ class LockCanvasResponseSchema(Schema): locked_by: str +class CanvasPositionSchema(Schema): + """Top-left node coordinate in React Flow space.""" + + x: float + y: float + + +class CanvasNodePositionUpdateSchema(Schema): + """Position update for one persisted canvas node.""" + + uuid: UUID + position: CanvasPositionSchema + + +class UpdateCanvasLayoutPayload(Schema): + """Atomic batch of canvas node position updates.""" + + nodes: list[CanvasNodePositionUpdateSchema] + + # ============================================================================== # UI4T V2 API - New unified architecture using CanvasNode and CanvasEdge # ============================================================================== diff --git a/ddpui/tests/api_tests/test_canvas_layout_api.py b/ddpui/tests/api_tests/test_canvas_layout_api.py new file mode 100644 index 000000000..9cc9e4bfa --- /dev/null +++ b/ddpui/tests/api_tests/test_canvas_layout_api.py @@ -0,0 +1,227 @@ +import uuid +from datetime import timedelta + +import pytest +from django.utils import timezone +from ninja.errors import HttpError + +from ddpui.api.transform_api import put_canvas_layout +from ddpui.core.dbtautomation_service import convert_canvas_node_to_frontend_format +from ddpui.models.canvas_models import CanvasNode, CanvasNodeType +from ddpui.models.canvaslock import CanvasLock +from ddpui.models.org import OrgDbt, TransformType +from ddpui.schemas.dbt_workflow_schema import UpdateCanvasLayoutPayload + +# authuser and org_without_workspace register dependencies of the shared fixtures. +from ddpui.tests.api_tests.test_user_org_api import ( # skipcq: PY-W2000 + authuser, + mock_request, + nonadminorguser, + org_without_workspace, + orguser, + seed_db, +) + + +pytestmark = pytest.mark.django_db + + +def _setup_workspace(orguser): + orgdbt = OrgDbt.objects.create( + gitrepo_url=None, + project_dir="test_project_dir", + dbt_venv="test_venv", + target_type="postgres", + default_schema="default_schema", + transform_type=TransformType.GIT, + ) + orguser.org.dbt = orgdbt + orguser.org.save(update_fields=["dbt"]) + return orgdbt + + +def _lock_workspace(orgdbt, orguser): + return CanvasLock.objects.create( + dbt=orgdbt, + locked_by=orguser, + lock_token=str(uuid.uuid4()), + expires_at=timezone.now() + timedelta(minutes=2), + ) + + +def _payload(*updates): + return UpdateCanvasLayoutPayload( + nodes=[ + { + "uuid": node_uuid, + "position": {"x": x, "y": y}, + } + for node_uuid, x, y in updates + ] + ) + + +def test_canvas_node_serializer_returns_nullable_position(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + ) + + assert convert_canvas_node_to_frontend_format(node)["position"] is None + + node.position_x = -125.5 + node.position_y = 240.25 + node.save(update_fields=["position_x", "position_y"]) + + assert convert_canvas_node_to_frontend_format(node)["position"] == { + "x": -125.5, + "y": 240.25, + } + + +def test_put_canvas_layout_updates_batch_atomically(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + _lock_workspace(orgdbt, orguser) + first = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="first", + ) + second = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.OPERATION, + name="second", + ) + + response = put_canvas_layout( + mock_request(orguser), + _payload((first.uuid, -10.5, 20.25), (second.uuid, 300.0, -400.0)), + ) + + first.refresh_from_db() + second.refresh_from_db() + assert response["updated"] == 2 + assert (first.position_x, first.position_y) == (-10.5, 20.25) + assert (second.position_x, second.position_y) == (300.0, -400.0) + + +def test_put_canvas_layout_rejects_unknown_node_without_partial_update(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + _lock_workspace(orgdbt, orguser) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + position_x=1.0, + position_y=2.0, + ) + + with pytest.raises(HttpError) as excinfo: + put_canvas_layout( + mock_request(orguser), + _payload((node.uuid, 50.0, 60.0), (uuid.uuid4(), 70.0, 80.0)), + ) + + assert excinfo.value.status_code == 422 + node.refresh_from_db() + assert (node.position_x, node.position_y) == (1.0, 2.0) + + +def test_put_canvas_layout_requires_owned_lock(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + ) + + with pytest.raises(HttpError) as excinfo: + put_canvas_layout(mock_request(orguser), _payload((node.uuid, 10.0, 20.0))) + + assert excinfo.value.status_code == 423 + node.refresh_from_db() + assert node.position_x is None + assert node.position_y is None + + +def test_put_canvas_layout_rejects_another_users_lock(seed_db, orguser, nonadminorguser): + orgdbt = _setup_workspace(orguser) + _lock_workspace(orgdbt, nonadminorguser) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + ) + + with pytest.raises(HttpError) as excinfo: + put_canvas_layout(mock_request(orguser), _payload((node.uuid, 10.0, 20.0))) + + assert excinfo.value.status_code == 423 + node.refresh_from_db() + assert node.position_x is None + assert node.position_y is None + + +def test_put_canvas_layout_rejects_expired_lock(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + lock = _lock_workspace(orgdbt, orguser) + lock.expires_at = timezone.now() - timedelta(seconds=1) + lock.save(update_fields=["expires_at"]) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + ) + + with pytest.raises(HttpError) as excinfo: + put_canvas_layout(mock_request(orguser), _payload((node.uuid, 10.0, 20.0))) + + assert excinfo.value.status_code == 410 + node.refresh_from_db() + assert node.position_x is None + assert node.position_y is None + + +def test_put_canvas_layout_rejects_duplicate_uuids(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + _lock_workspace(orgdbt, orguser) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + ) + duplicate_updates = ((node.uuid, 1.0, 2.0), (node.uuid, 3.0, 4.0)) + + with pytest.raises(HttpError) as excinfo: + put_canvas_layout(mock_request(orguser), _payload(*duplicate_updates)) + + assert excinfo.value.status_code == 422 + assert "duplicate" in str(excinfo.value) + + +@pytest.mark.parametrize( + "x,y", + [ + (float("nan"), 0.0), + (0.0, float("inf")), + (10_000_001.0, 0.0), + ], +) +def test_put_canvas_layout_rejects_invalid_coordinates(seed_db, orguser, x, y): + orgdbt = _setup_workspace(orguser) + _lock_workspace(orgdbt, orguser) + node = CanvasNode.objects.create( + orgdbt=orgdbt, + node_type=CanvasNodeType.SOURCE, + name="source", + ) + + with pytest.raises(HttpError) as excinfo: + put_canvas_layout(mock_request(orguser), _payload((node.uuid, x, y))) + + assert excinfo.value.status_code == 422 + node.refresh_from_db() + assert node.position_x is None + assert node.position_y is None diff --git a/ddpui/tests/api_tests/test_canvas_locking_api.py b/ddpui/tests/api_tests/test_canvas_locking_api.py new file mode 100644 index 000000000..822909b68 --- /dev/null +++ b/ddpui/tests/api_tests/test_canvas_locking_api.py @@ -0,0 +1,175 @@ +import uuid +from concurrent.futures import ThreadPoolExecutor +from datetime import timedelta +from threading import Barrier + +import pytest +from django.db import close_old_connections +from django.utils import timezone +from ninja.errors import HttpError + +from ddpui.api.transform_api import ( + delete_canvas_node, + delete_orgdbtmodel, + lock_canvas, + post_add_operation_node, + post_create_src_model_node, + post_terminate_operation_node, + put_operation_node, + refresh_canvas_lock, + sync_remote_dbtproject_to_canvas, +) +from ddpui.models.canvaslock import CanvasLock +from ddpui.models.org import OrgDbt, OrgWarehouse, TransformType +from ddpui.models.org_user import OrgUser +from ddpui.schemas.dbt_workflow_schema import ( + CreateOperationNodePayload, + EditOperationNodePayload, + TerminateChainAndCreateModelPayload, +) + +# authuser and org_without_workspace register dependencies of the shared fixtures. +from ddpui.tests.api_tests.test_user_org_api import ( # skipcq: PY-W2000 + authuser, + mock_request, + nonadminorguser, + org_without_workspace, + orguser, + seed_db, +) + + +pytestmark = pytest.mark.django_db + + +def _setup_workspace(orguser): + orgdbt = OrgDbt.objects.create( + gitrepo_url=None, + project_dir="test_project_dir", + dbt_venv="test_venv", + target_type="postgres", + default_schema="default_schema", + transform_type=TransformType.GIT, + ) + orguser.org.dbt = orgdbt + orguser.org.save(update_fields=["dbt"]) + OrgWarehouse.objects.create( + org=orguser.org, + wtype="postgres", + airbyte_destination_id="test_destination_id", + ) + return orgdbt + + +def _create_lock(orgdbt, locked_by, *, expired=False): + expiry_delta = timedelta(minutes=-1 if expired else 2) + return CanvasLock.objects.create( + dbt=orgdbt, + locked_by=locked_by, + lock_token=str(uuid.uuid4()), + expires_at=timezone.now() + expiry_delta, + ) + + +def test_acquire_replaces_an_expired_lock(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + expired_lock = _create_lock(orgdbt, orguser, expired=True) + + response = lock_canvas(mock_request(orguser)) + + assert response.lock_token != expired_lock.lock_token + assert CanvasLock.objects.filter(dbt=orgdbt, lock_token=response.lock_token).count() == 1 + + +def test_refresh_reports_expired_lock(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + _create_lock(orgdbt, orguser, expired=True) + + with pytest.raises(HttpError) as excinfo: + refresh_canvas_lock(mock_request(orguser)) + + assert excinfo.value.status_code == 410 + + +def test_refresh_reports_missing_lock(seed_db, orguser): + _setup_workspace(orguser) + + with pytest.raises(HttpError) as excinfo: + refresh_canvas_lock(mock_request(orguser)) + + assert excinfo.value.status_code == 404 + + +def test_refresh_rejects_another_users_lock(seed_db, orguser, nonadminorguser): + orgdbt = _setup_workspace(orguser) + _create_lock(orgdbt, nonadminorguser) + + with pytest.raises(HttpError) as excinfo: + refresh_canvas_lock(mock_request(orguser)) + + assert excinfo.value.status_code == 403 + + +def _unlocked_mutations(request): + create_payload = CreateOperationNodePayload( + config={}, + input_node_uuid=str(uuid.uuid4()), + op_type="aggregate", + source_columns=["id"], + ) + edit_payload = EditOperationNodePayload( + config={}, + op_type="aggregate", + source_columns=["id"], + ) + terminate_payload = TerminateChainAndCreateModelPayload( + name="new_model", + display_name="New model", + dest_schema="analytics", + ) + node_uuid = str(uuid.uuid4()) + return [ + ("add source/model", lambda: post_create_src_model_node(request, node_uuid)), + ("add operation", lambda: post_add_operation_node(request, create_payload)), + ("edit operation", lambda: put_operation_node(request, node_uuid, edit_payload)), + ( + "terminate operation", + lambda: post_terminate_operation_node(request, node_uuid, terminate_payload), + ), + ("remove from canvas", lambda: delete_canvas_node(request, node_uuid)), + ("delete dbt model", lambda: delete_orgdbtmodel(request, node_uuid)), + ("sync remote graph", lambda: sync_remote_dbtproject_to_canvas(request)), + ] + + +def test_every_v2_canvas_mutation_requires_a_lock(seed_db, orguser): + _setup_workspace(orguser) + request = mock_request(orguser) + + for mutation_name, mutate in _unlocked_mutations(request): + with pytest.raises(HttpError) as excinfo: + mutate() + assert excinfo.value.status_code == 423, mutation_name + + +@pytest.mark.django_db(transaction=True) +def test_concurrent_same_user_acquisition_is_idempotent(seed_db, orguser): + orgdbt = _setup_workspace(orguser) + barrier = Barrier(2) + + def acquire(): + close_old_connections() + try: + fresh_orguser = OrgUser.objects.select_related("org__dbt", "user", "new_role").get( + pk=orguser.pk + ) + barrier.wait() + return lock_canvas(mock_request(fresh_orguser)).lock_token + finally: + close_old_connections() + + with ThreadPoolExecutor(max_workers=2) as executor: + tokens = list(executor.map(lambda _index: acquire(), range(2))) + + assert len(set(tokens)) == 1 + assert CanvasLock.objects.filter(dbt=orgdbt).count() == 1 diff --git a/ddpui/tests/api_tests/test_transform_api.py b/ddpui/tests/api_tests/test_transform_api.py index f7f894f1e..e934b2aeb 100644 --- a/ddpui/tests/api_tests/test_transform_api.py +++ b/ddpui/tests/api_tests/test_transform_api.py @@ -69,6 +69,16 @@ } +def lock_workspace_for_test(orgdbt: OrgDbt, orguser: OrgUser): + """Give endpoint-domain tests the lock required by the production contract.""" + return CanvasLock.objects.create( + dbt=orgdbt, + locked_by=orguser, + lock_token=str(uuid.uuid4()), + expires_at=timezone.now() + timedelta(minutes=2), + ) + + def create_canvas_graph(orgdbt): """Helper function to create a sample canvas graph with nodes and edges""" # Create canvas nodes @@ -1388,6 +1398,7 @@ def test_post_create_src_model_node_model_not_found(seed_db, orguser): ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) with pytest.raises(HttpError) as excinfo: post_create_src_model_node(request, str(uuid.uuid4())) @@ -1420,6 +1431,7 @@ def test_post_create_src_model_node_source_type_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create SOURCE type dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1496,6 +1508,7 @@ def test_post_create_src_model_node_model_type_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create MODEL type dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1558,6 +1571,7 @@ def test_post_create_src_model_node_existing_node_returns_existing( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1615,6 +1629,7 @@ def test_post_create_src_model_node_source_definition_error(mock_ensure_source, ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create SOURCE type dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1663,6 +1678,7 @@ def test_post_create_src_model_node_update_cols_error(mock_update_cols, seed_db, ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create MODEL type dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1715,6 +1731,7 @@ def test_post_create_src_model_node_source_no_existing_node_calls_all_functions( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create SOURCE type dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1773,6 +1790,7 @@ def test_post_create_src_model_node_model_no_existing_node_calls_required_functi ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create MODEL type dbt model org_dbt_model = OrgDbtModel.objects.create( @@ -1862,6 +1880,7 @@ def test_post_add_operation_node_input_node_not_found(seed_db, orguser): ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) payload = CreateOperationNodePayload( config={}, @@ -1901,6 +1920,7 @@ def test_post_add_operation_node_single_input_operation_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create a source node to use as input org_dbt_model = OrgDbtModel.objects.create( @@ -1991,6 +2011,7 @@ def test_post_add_operation_node_multi_input_operation_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create two source models for join org_dbt_model1 = OrgDbtModel.objects.create( @@ -2102,6 +2123,7 @@ def test_post_add_operation_node_invalid_operation_config(mock_validate_config, ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create input node input_node = CanvasNode.objects.create( @@ -2158,6 +2180,7 @@ def test_post_add_operation_node_get_output_cols_error( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create input node input_node = CanvasNode.objects.create( @@ -2215,6 +2238,7 @@ def test_post_add_operation_node_validate_inputs_error( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create input node input_node = CanvasNode.objects.create( @@ -2274,6 +2298,7 @@ def test_post_add_operation_node_verify_function_calls( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create input node input_node = CanvasNode.objects.create( @@ -2382,6 +2407,7 @@ def test_put_operation_node_operation_node_not_found(seed_db, orguser): ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) payload = EditOperationNodePayload(config={}, op_type="aggregate", source_columns=["id"]) @@ -2411,6 +2437,7 @@ def test_put_operation_node_non_operation_node(seed_db, orguser): ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create a source node (not operation node) source_node = CanvasNode.objects.create( @@ -2453,6 +2480,7 @@ def test_put_operation_node_single_input_operation_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create an existing operation node existing_operation = CanvasNode.objects.create( @@ -2531,6 +2559,7 @@ def test_put_operation_node_multi_input_operation_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create two source models for join org_dbt_model1 = OrgDbtModel.objects.create( @@ -2665,6 +2694,7 @@ def test_put_operation_node_invalid_operation_config(mock_validate_config, seed_ ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create existing operation node existing_operation = CanvasNode.objects.create( @@ -2720,6 +2750,7 @@ def test_put_operation_node_get_output_cols_error( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create existing operation node existing_operation = CanvasNode.objects.create( @@ -2778,6 +2809,7 @@ def test_put_operation_node_validate_inputs_error( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create existing operation node existing_operation = CanvasNode.objects.create( @@ -2838,6 +2870,7 @@ def test_put_operation_node_verify_function_calls( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create existing operation node existing_operation = CanvasNode.objects.create( @@ -2916,6 +2949,7 @@ def test_put_operation_node_edge_cleanup_for_multi_input_operation( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create models for the multi-input operation org_dbt_model1 = OrgDbtModel.objects.create( @@ -3064,6 +3098,7 @@ def test_post_terminate_operation_node_operation_node_not_found(seed_db, orguser ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) payload = TerminateChainAndCreateModelPayload( name="test_model", display_name="Test Model", dest_schema="public" @@ -3096,6 +3131,7 @@ def test_post_terminate_operation_node_non_operation_node(seed_db, orguser): ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create a source node instead of operation source_node = CanvasNode.objects.create( @@ -3145,6 +3181,7 @@ def test_post_terminate_operation_node_create_new_model_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create operation node operation_node = CanvasNode.objects.create( @@ -3224,6 +3261,7 @@ def test_post_terminate_operation_node_update_existing_model_success( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create existing dbt model existing_dbt_model = OrgDbtModel.objects.create( @@ -3315,6 +3353,7 @@ def test_post_terminate_operation_node_create_dbt_model_error( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create operation node operation_node = CanvasNode.objects.create( @@ -3373,6 +3412,7 @@ def test_post_terminate_operation_node_with_rel_dir_to_models( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create operation node operation_node = CanvasNode.objects.create( @@ -3441,6 +3481,7 @@ def test_post_terminate_operation_node_with_root_directory_edge_case( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create operation node operation_node = CanvasNode.objects.create( orgdbt=orgdbt, @@ -3509,6 +3550,7 @@ def test_post_terminate_operation_node_traverse_graph_error( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create operation node operation_node = CanvasNode.objects.create( @@ -3565,6 +3607,7 @@ def test_post_terminate_operation_node_verify_transaction_atomicity( ) orguser.org.dbt = orgdbt orguser.org.save() + lock_workspace_for_test(orgdbt, orguser) # Create operation node operation_node = CanvasNode.objects.create( diff --git a/ddpui/tests/core/trial/test_dbt_clone.py b/ddpui/tests/core/trial/test_dbt_clone.py index 9d9860a5d..54f5bf9ba 100644 --- a/ddpui/tests/core/trial/test_dbt_clone.py +++ b/ddpui/tests/core/trial/test_dbt_clone.py @@ -125,6 +125,8 @@ def test_copy_dbt_dag_copies_canvas(): name="stg_customers", output_cols=["id"], dbtmodel=src_model, + position_x=-120.5, + position_y=80.25, ) op_node = CanvasNode.objects.create( orgdbt=template_dbt, @@ -158,6 +160,8 @@ def test_copy_dbt_dag_copies_canvas(): assert new_op_node.dbtmodel_id is None assert new_op_node.operation_config == {"operation_type": "rename"} assert new_src_node.uuid != src_node.uuid + assert new_src_node.position_x == -120.5 + assert new_src_node.position_y == 80.25 trial_edges = list(CanvasEdge.objects.filter(from_node__orgdbt=trial_dbt).order_by("seq")) assert len(trial_edges) == 2 diff --git a/ddpui/tests/services/dbt_service/test_parse_dbt_manifest_to_canvas.py b/ddpui/tests/services/dbt_service/test_parse_dbt_manifest_to_canvas.py index 37562d728..7331b7b4e 100644 --- a/ddpui/tests/services/dbt_service/test_parse_dbt_manifest_to_canvas.py +++ b/ddpui/tests/services/dbt_service/test_parse_dbt_manifest_to_canvas.py @@ -230,6 +230,8 @@ def test_parse_dbt_manifest_to_canvas_update_existing(org_with_dbt_workspace: Or node_type=CanvasNodeType.SOURCE, output_cols=["old_col1", "old_col2"], dbtmodel=existing_orgdbt_model, + position_x=-75.5, + position_y=125.25, ) mock_warehouse = Mock() @@ -260,6 +262,7 @@ def test_parse_dbt_manifest_to_canvas_update_existing(org_with_dbt_workspace: Or assert updated_node.id == existing_node.id # Same node assert "id" in updated_node.output_cols # New columns added assert "name" in updated_node.output_cols + assert (updated_node.position_x, updated_node.position_y) == (-75.5, 125.25) @pytest.mark.django_db