Add tests

Author: shangyian

datajunction-server/tests/api/namespaces_test.py

@@ -918,3 +918,80 @@ def get_deny_all_service():
 
     assert response.status_code in (200, 201)
     assert response.json() == []
+
+
+@pytest.mark.asyncio
+async def test_create_namespace_auto_creates_owner_role(
+    client_with_service_setup: AsyncClient,
+) -> None:
+    """
+    Test that creating a namespace automatically creates an owner role.
+
+    When a namespace is created, the system should:
+    1. Create a role named "{namespace}-owner"
+    2. Add a MANAGE scope on the namespace
+    3. Assign the role to the creating user
+    """
+    # Create a new namespace
+    response = await client_with_service_setup.post(
+        "/namespaces/testrbac.autorole",
+    )
+    assert response.status_code == 201
+
+    # Verify the owner role was created
+    response = await client_with_service_setup.get("/roles/testrbac-owner")
+    assert response.status_code == 200
+    role_data = response.json()
+    assert role_data["name"] == "testrbac-owner"
+    assert role_data["description"] == "Owner role for namespace testrbac"
+
+    # Check scopes - should have MANAGE on the namespace
+    assert len(role_data["scopes"]) == 1
+    scope = role_data["scopes"][0]
+    assert scope["action"] == "manage"
+    assert scope["scope_type"] == "namespace"
+    assert scope["scope_value"] == "testrbac"
+
+    # Verify the child namespace also got an owner role
+    response = await client_with_service_setup.get("/roles/testrbac.autorole-owner")
+    assert response.status_code == 200
+    child_role_data = response.json()
+    assert child_role_data["name"] == "testrbac.autorole-owner"
+
+    # Check the role is assigned to the creator
+    response = await client_with_service_setup.get("/roles/testrbac-owner/assignments")
+    assert response.status_code == 200
+    assignments = response.json()
+    assert len(assignments) >= 1
+    # The creator should be assigned this role
+    assert any(a["principal"]["username"] == "dj" for a in assignments)
+
+
+@pytest.mark.asyncio
+async def test_create_namespace_skips_existing_role(
+    client_with_service_setup: AsyncClient,
+) -> None:
+    """
+    Test that creating a namespace skips role creation if role already exists.
+    """
+    # First, manually create a role with the expected name
+    response = await client_with_service_setup.post(
+        "/roles/",
+        json={
+            "name": "preexisting-owner",
+            "description": "Manually created role",
+        },
+    )
+    assert response.status_code == 201
+
+    # Now create a namespace with the same name
+    response = await client_with_service_setup.post(
+        "/namespaces/preexisting",
+    )
+    assert response.status_code == 201
+
+    # The role should still be the original one (not overwritten)
+    response = await client_with_service_setup.get("/roles/preexisting-owner")
+    assert response.status_code == 200
+    role_data = response.json()
+    assert role_data["description"] == "Manually created role"  # Original description

datajunction-server/tests/api/nodes_test.py

@@ -6298,3 +6298,114 @@ async def test_copy_nodes(
             # ]
             # for copied in copied_dimensions:
             #     assert copied in original_dimensions
+
+
+@pytest.mark.asyncio
+async def test_create_node_auto_creates_owner_role(
+    client_with_service_setup: AsyncClient,
+) -> None:
+    """
+    Test that creating a node automatically creates an owner role.
+
+    When a node is created, the system should:
+    1. Create a role named "{node_name}-owner"
+    2. Add a MANAGE scope on the node
+    3. Assign the role to the creating user
+    """
+    # First create a namespace for our test node
+    response = await client_with_service_setup.post("/namespaces/nodeautotest")
+    assert response.status_code == 201
+
+    # Create a source node
+    response = await client_with_service_setup.post(
+        "/nodes/source/",
+        json={
+            "name": "nodeautotest.test_source",
+            "description": "Test source for auto-role creation",
+            "catalog": "default",
+            "schema_": "test_schema",
+            "table": "test_table",
+            "columns": [
+                {"name": "id", "type": "int"},
+                {"name": "name", "type": "string"},
+            ],
+            "mode": "published",
+        },
+    )
+    assert response.status_code == 201
+
+    # Verify the owner role was created for the node
+    response = await client_with_service_setup.get(
+        "/roles/nodeautotest.test_source-owner",
+    )
+    assert response.status_code == 200
+    role_data = response.json()
+    assert role_data["name"] == "nodeautotest.test_source-owner"
+    assert role_data["description"] == "Owner role for node nodeautotest.test_source"
+
+    # Check scopes - should have MANAGE on the node
+    assert len(role_data["scopes"]) == 1
+    scope = role_data["scopes"][0]
+    assert scope["action"] == "manage"
+    assert scope["scope_type"] == "node"
+    assert scope["scope_value"] == "nodeautotest.test_source"
+
+    # Check the role is assigned to the creator
+    response = await client_with_service_setup.get(
+        "/roles/nodeautotest.test_source-owner/assignments",
+    )
+    assert response.status_code == 200
+    assignments = response.json()
+    assert len(assignments) >= 1
+    # The creator should be assigned this role
+    assert any(a["principal"]["username"] == "dj" for a in assignments)
+
+
+@pytest.mark.asyncio
+async def test_create_transform_node_auto_creates_owner_role(
+    client_with_service_setup: AsyncClient,
+) -> None:
+    """
+    Test that creating a transform node also auto-creates owner role.
+    """
+    # First create namespace and source node
+    await client_with_service_setup.post("/namespaces/transformtest")
+
+    await client_with_service_setup.post(
+        "/nodes/source/",
+        json={
+            "name": "transformtest.source",
+            "description": "Source for transform test",
+            "catalog": "default",
+            "schema_": "test_schema",
+            "table": "test_table",
+            "columns": [
+                {"name": "id", "type": "int"},
+                {"name": "value", "type": "float"},
+            ],
+            "mode": "published",
+        },
+    )
+
+    # Create a transform node
+    response = await client_with_service_setup.post(
+        "/nodes/transform/",
+        json={
+            "name": "transformtest.my_transform",
+            "description": "Test transform",
+            "query": "SELECT id, value FROM transformtest.source",
+            "mode": "published",
+        },
+    )
+    assert response.status_code == 201
+
+    # Verify owner role was created
+    response = await client_with_service_setup.get(
+        "/roles/transformtest.my_transform-owner",
+    )
+    assert response.status_code == 200
+    role_data = response.json()
+    assert role_data["name"] == "transformtest.my_transform-owner"
+    assert len(role_data["scopes"]) == 1
+    assert role_data["scopes"][0]["action"] == "manage"
+    assert role_data["scopes"][0]["scope_type"] == "node"