502 lines
18 KiB
Python
502 lines
18 KiB
Python
"""Integration tests for tree endpoints."""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.models.folder import user_folder_trees
|
|
from app.models.tag import tree_tag_assignments
|
|
|
|
|
|
class TestTrees:
|
|
"""Test suite for decision tree endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_tree(self, client: AsyncClient, auth_headers: dict):
|
|
"""Test creating a new decision tree."""
|
|
tree_data = {
|
|
"name": "Network Troubleshooting",
|
|
"description": "Troubleshoot network connectivity issues",
|
|
"category": "Networking",
|
|
"tree_structure": {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"question": "Can you ping the gateway?",
|
|
"options": [
|
|
{"id": "yes", "label": "Yes", "next_node_id": "check_dns"},
|
|
{"id": "no", "label": "No", "next_node_id": "check_cable"}
|
|
],
|
|
"children": [
|
|
{
|
|
"id": "check_dns",
|
|
"type": "decision",
|
|
"question": "Can you resolve DNS?",
|
|
"options": [],
|
|
"children": []
|
|
},
|
|
{
|
|
"id": "check_cable",
|
|
"type": "solution",
|
|
"title": "Check Network Cable",
|
|
"description": "Verify the network cable is connected",
|
|
"solution": "Check and reseat the network cable"
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
response = await client.post(
|
|
"/api/v1/trees",
|
|
json=tree_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["name"] == tree_data["name"]
|
|
assert data["category"] == tree_data["category"]
|
|
assert data["is_active"] is True
|
|
assert data["version"] == 1
|
|
assert "id" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_trees(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test listing decision trees."""
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 1
|
|
# Check that our test tree is in the list
|
|
tree_ids = [tree["id"] for tree in data]
|
|
assert test_tree["id"] in tree_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tree_by_id(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test getting a specific tree by ID."""
|
|
response = await client.get(
|
|
f"/api/v1/trees/{test_tree['id']}",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == test_tree["id"]
|
|
assert data["name"] == test_tree["name"]
|
|
assert "tree_structure" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_nonexistent_tree(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""Test getting a tree that doesn't exist."""
|
|
fake_id = "00000000-0000-0000-0000-000000000000"
|
|
response = await client.get(
|
|
f"/api/v1/trees/{fake_id}",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_trees(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test full-text search for trees."""
|
|
response = await client.get(
|
|
"/api/v1/trees/search?q=test",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
# Should find our test tree
|
|
if len(data) > 0:
|
|
assert any(tree["id"] == test_tree["id"] for tree in data)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_categories(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test getting unique tree categories."""
|
|
response = await client.get(
|
|
"/api/v1/trees/categories",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
# Should include our test tree's category
|
|
assert test_tree["category"] in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_tree(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test updating a tree."""
|
|
update_data = {
|
|
"name": "Updated Tree Name",
|
|
"description": "Updated description"
|
|
}
|
|
|
|
response = await client.put(
|
|
f"/api/v1/trees/{test_tree['id']}",
|
|
json=update_data,
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["name"] == update_data["name"]
|
|
assert data["description"] == update_data["description"]
|
|
# Version only increments when tree_structure is updated
|
|
assert data["version"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_tree(
|
|
self, client: AsyncClient, admin_auth_headers: dict, test_tree: dict
|
|
):
|
|
"""Test soft-deleting a tree (admin only)."""
|
|
response = await client.delete(
|
|
f"/api/v1/trees/{test_tree['id']}",
|
|
headers=admin_auth_headers
|
|
)
|
|
|
|
assert response.status_code == 204
|
|
|
|
# Verify tree is no longer in active list
|
|
list_response = await client.get("/api/v1/trees", headers=admin_auth_headers)
|
|
active_trees = list_response.json()
|
|
active_ids = [tree["id"] for tree in active_trees]
|
|
assert test_tree["id"] not in active_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_super_admin_sees_all_trees(
|
|
self, client: AsyncClient, auth_headers: dict, admin_auth_headers: dict
|
|
):
|
|
"""Test that super admin can see all trees including private ones from other users."""
|
|
# Create a private (non-public, non-default) tree as a regular user
|
|
tree_data = {
|
|
"name": "Private User Tree",
|
|
"description": "Only visible to author and super admin",
|
|
"tree_structure": {
|
|
"id": "root",
|
|
"type": "solution",
|
|
"title": "Private",
|
|
"description": "Private tree",
|
|
"solution": "Private solution"
|
|
},
|
|
"is_public": False,
|
|
"is_default": False
|
|
}
|
|
|
|
create_response = await client.post(
|
|
"/api/v1/trees", json=tree_data, headers=auth_headers
|
|
)
|
|
assert create_response.status_code == 201
|
|
private_tree_id = create_response.json()["id"]
|
|
|
|
# Super admin should see it in list
|
|
list_response = await client.get("/api/v1/trees", headers=admin_auth_headers)
|
|
assert list_response.status_code == 200
|
|
tree_ids = [t["id"] for t in list_response.json()]
|
|
assert private_tree_id in tree_ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_tree_cleans_up_folder_and_tag_assignments(
|
|
self, client: AsyncClient, auth_headers: dict, admin_auth_headers: dict, test_db: AsyncSession
|
|
):
|
|
"""Test that soft-deleting a tree removes folder and tag junction entries."""
|
|
# Create a tree
|
|
tree_data = {
|
|
"name": "Cascade Test Tree",
|
|
"description": "Will be deleted",
|
|
"tree_structure": {
|
|
"id": "root",
|
|
"type": "solution",
|
|
"title": "Test",
|
|
"description": "Test tree",
|
|
"solution": "Test solution"
|
|
}
|
|
}
|
|
create_resp = await client.post("/api/v1/trees", json=tree_data, headers=auth_headers)
|
|
assert create_resp.status_code == 201
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Create a folder and add the tree to it
|
|
folder_resp = await client.post(
|
|
"/api/v1/folders", json={"name": "Test Folder"}, headers=auth_headers
|
|
)
|
|
assert folder_resp.status_code == 201
|
|
folder_id = folder_resp.json()["id"]
|
|
|
|
add_resp = await client.post(
|
|
f"/api/v1/folders/{folder_id}/trees",
|
|
json={"tree_id": tree_id},
|
|
headers=auth_headers
|
|
)
|
|
assert add_resp.status_code == 201
|
|
|
|
# Add tags to the tree
|
|
tag_resp = await client.post(
|
|
f"/api/v1/tags/trees/{tree_id}",
|
|
json={"tags": ["cascade-test-tag"]},
|
|
headers=auth_headers
|
|
)
|
|
assert tag_resp.status_code == 200
|
|
|
|
# Verify junction rows exist
|
|
folder_rows = await test_db.execute(
|
|
select(user_folder_trees).where(user_folder_trees.c.tree_id == tree_id)
|
|
)
|
|
assert len(folder_rows.fetchall()) > 0
|
|
|
|
tag_rows = await test_db.execute(
|
|
select(tree_tag_assignments).where(tree_tag_assignments.c.tree_id == tree_id)
|
|
)
|
|
assert len(tag_rows.fetchall()) > 0
|
|
|
|
# Delete the tree (admin only)
|
|
del_resp = await client.delete(f"/api/v1/trees/{tree_id}", headers=admin_auth_headers)
|
|
assert del_resp.status_code == 204
|
|
|
|
# Verify junction rows are gone
|
|
folder_rows_after = await test_db.execute(
|
|
select(user_folder_trees).where(user_folder_trees.c.tree_id == tree_id)
|
|
)
|
|
assert len(folder_rows_after.fetchall()) == 0
|
|
|
|
tag_rows_after = await test_db.execute(
|
|
select(tree_tag_assignments).where(tree_tag_assignments.c.tree_id == tree_id)
|
|
)
|
|
assert len(tag_rows_after.fetchall()) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tag_search_escapes_wildcards(
|
|
self, client: AsyncClient, admin_auth_headers: dict
|
|
):
|
|
"""Test that SQL wildcards in tag search are escaped, not interpreted."""
|
|
# Create tags as admin (can create global tags)
|
|
resp1 = await client.post(
|
|
"/api/v1/tags",
|
|
json={"name": "test_underscore"},
|
|
headers=admin_auth_headers
|
|
)
|
|
assert resp1.status_code == 201
|
|
|
|
resp2 = await client.post(
|
|
"/api/v1/tags",
|
|
json={"name": "testXunderscore"},
|
|
headers=admin_auth_headers
|
|
)
|
|
assert resp2.status_code == 201
|
|
|
|
# Search for literal underscore — should only match the first tag
|
|
response = await client.get(
|
|
"/api/v1/tags/search",
|
|
params={"q": "test_under"},
|
|
headers=admin_auth_headers
|
|
)
|
|
assert response.status_code == 200
|
|
names = [t["name"] for t in response.json()]
|
|
assert "test_underscore" in names
|
|
assert "testXunderscore" not in names
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_tree_unauthorized(self, client: AsyncClient):
|
|
"""Test that creating a tree without auth fails."""
|
|
tree_data = {
|
|
"name": "Unauthorized Tree",
|
|
"tree_structure": {"id": "root", "type": "decision"}
|
|
}
|
|
|
|
response = await client.post("/api/v1/trees", json=tree_data)
|
|
|
|
assert response.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_trees_sorting(self, client: AsyncClient, auth_headers: dict):
|
|
"""Test sorting trees by different criteria."""
|
|
# Create multiple trees with different attributes
|
|
import asyncio
|
|
|
|
# Create trees with different names and versions
|
|
trees_data = [
|
|
{"name": "Alpha Tree", "description": "First alphabetically"},
|
|
{"name": "Zulu Tree", "description": "Last alphabetically"},
|
|
{"name": "Beta Tree", "description": "Second alphabetically"},
|
|
]
|
|
|
|
created_trees = []
|
|
for tree_data in trees_data:
|
|
tree_data["tree_structure"] = {
|
|
"id": "root",
|
|
"type": "solution",
|
|
"title": "Test",
|
|
"description": "Test tree",
|
|
"solution": "Test solution"
|
|
}
|
|
response = await client.post("/api/v1/trees", json=tree_data, headers=auth_headers)
|
|
assert response.status_code == 201
|
|
created_trees.append(response.json())
|
|
# Small delay to ensure different timestamps
|
|
await asyncio.sleep(0.1)
|
|
|
|
# Test sorting by name (A-Z)
|
|
response = await client.get("/api/v1/trees?sort_by=name", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
trees = response.json()
|
|
names = [t["name"] for t in trees if t["id"] in [c["id"] for c in created_trees]]
|
|
assert names == sorted(names)
|
|
|
|
# Test sorting by name descending (Z-A)
|
|
response = await client.get("/api/v1/trees?sort_by=name_desc", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
trees = response.json()
|
|
names = [t["name"] for t in trees if t["id"] in [c["id"] for c in created_trees]]
|
|
assert names == sorted(names, reverse=True)
|
|
|
|
# Test sorting by created_at (most recent first)
|
|
response = await client.get("/api/v1/trees?sort_by=created_at", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
trees = response.json()
|
|
# Most recently created should be first
|
|
filtered_trees = [t for t in trees if t["id"] in [c["id"] for c in created_trees]]
|
|
if len(filtered_trees) >= 2:
|
|
# Verify descending order
|
|
for i in range(len(filtered_trees) - 1):
|
|
assert filtered_trees[i]["created_at"] >= filtered_trees[i+1]["created_at"]
|
|
|
|
# Test sorting by updated_at
|
|
response = await client.get("/api/v1/trees?sort_by=updated_at", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
trees = response.json()
|
|
assert isinstance(trees, list)
|
|
|
|
# Test sorting by version
|
|
response = await client.get("/api/v1/trees?sort_by=version", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
trees = response.json()
|
|
assert isinstance(trees, list)
|
|
|
|
# Test default sorting (usage_count)
|
|
response = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
trees = response.json()
|
|
assert isinstance(trees, list)
|
|
|
|
|
|
class TestVisibilityFilter:
|
|
"""Test that visibility filtering and author_name work correctly."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_private_tree_only_visible_to_author(
|
|
self, client: AsyncClient, auth_headers: dict, test_user: dict
|
|
):
|
|
"""A private tree created by test_user should appear in their own list."""
|
|
tree_data = {
|
|
"name": "Private Flow Test",
|
|
"tree_structure": {"id": "root", "type": "decision", "question": "Q?", "options": [], "children": []},
|
|
}
|
|
create_resp = await client.post("/api/v1/trees", json=tree_data, headers=auth_headers)
|
|
assert create_resp.status_code == 201
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Set visibility to private
|
|
vis_resp = await client.patch(
|
|
f"/api/v1/trees/{tree_id}/visibility",
|
|
json={"visibility": "private"},
|
|
headers=auth_headers
|
|
)
|
|
assert vis_resp.status_code == 200
|
|
|
|
# Verify it still appears for the author
|
|
list_resp = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert list_resp.status_code == 200
|
|
ids = [t["id"] for t in list_resp.json()]
|
|
assert tree_id in ids
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_visibility_query_param_filters_correctly(
|
|
self, client: AsyncClient, auth_headers: dict
|
|
):
|
|
"""?visibility=public should only return trees with visibility='public'."""
|
|
resp = await client.get("/api/v1/trees?visibility=public", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
trees = resp.json()
|
|
for tree in trees:
|
|
assert tree["visibility"] == "public", f"Tree {tree['id']} has visibility={tree['visibility']}"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_author_name_present_in_list_response(
|
|
self, client: AsyncClient, auth_headers: dict, test_tree: dict
|
|
):
|
|
"""TreeListResponse should include author_name field."""
|
|
resp = await client.get("/api/v1/trees", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
trees = resp.json()
|
|
assert len(trees) >= 1
|
|
# author_name key should be present (value may be None for system/default trees)
|
|
assert "author_name" in trees[0]
|
|
# visibility key should be present
|
|
assert "visibility" in trees[0]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_tree_returns_404_not_403_for_other_account_tree(
|
|
self, client: AsyncClient, auth_headers: dict, test_db: AsyncSession
|
|
):
|
|
"""Account A must not learn that Account B's private tree exists."""
|
|
from app.models.tree import Tree
|
|
from app.models.account import Account
|
|
from app.models.user import User
|
|
from app.core.security import get_password_hash
|
|
import uuid
|
|
|
|
# Create a second account and user
|
|
account_b = Account(name="Other Corp", display_code="OTH00001")
|
|
test_db.add(account_b)
|
|
await test_db.flush()
|
|
|
|
user_b = User(
|
|
email=f"user-b-{uuid.uuid4().hex[:6]}@example.com",
|
|
name="User B",
|
|
password_hash=get_password_hash("TestPass123!"),
|
|
is_active=True,
|
|
account_id=account_b.id,
|
|
account_role="engineer",
|
|
)
|
|
test_db.add(user_b)
|
|
await test_db.flush()
|
|
|
|
# Create a private tree belonging to account_b
|
|
private_tree = Tree(
|
|
name="Secret Tree",
|
|
account_id=account_b.id,
|
|
author_id=user_b.id,
|
|
visibility="private",
|
|
tree_type="troubleshooting",
|
|
tree_structure={"id": "root", "type": "start", "children": []},
|
|
is_active=True,
|
|
is_default=False,
|
|
is_public=False,
|
|
status="published",
|
|
)
|
|
test_db.add(private_tree)
|
|
await test_db.commit()
|
|
|
|
response = await client.get(
|
|
f"/api/v1/trees/{private_tree.id}",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 404, (
|
|
f"Expected 404 but got {response.status_code} — "
|
|
"leaking tree existence to wrong tenant"
|
|
)
|