Files
resolutionflow/backend/tests/test_draft_trees.py
Michael Chihlas c7b2c59ef6 feat: implement tree sharing, draft trees, and session-to-tree conversion (Issues #16, #25, #17)
Backend features:
- Tree sharing via secure tokens with expiration (Issue #16)
- Draft tree status with conditional validation (Issue #25)
- Save session as custom tree with fork tracking (Issue #17)
- Tree validation system for publish requirements
- Session-to-tree conversion preserving custom steps

Database migrations:
- 024: Tree sharing (tree_shares table, visibility field)
- 025: Tree status field (draft/published)
- 25b: Merge migration for indexes

New endpoints:
- POST /api/v1/trees/{id}/share - Generate share token
- GET /api/v1/shared/{token} - Public tree access
- POST /api/v1/trees/{id}/can-publish - Validate tree
- POST /api/v1/sessions/{id}/save-as-tree - Convert session

Test coverage:
- 20 tests for draft trees functionality
- 14 tests for session-to-tree conversion
- 15 tests for tree sharing

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-07 23:06:13 -05:00

350 lines
13 KiB
Python

"""Tests for draft trees feature (Issue #25)."""
import pytest
from httpx import AsyncClient
from uuid import UUID
from app.models.tree import Tree
from app.core.tree_validation import validate_tree_structure, can_publish_tree
class TestTreeValidation:
"""Test suite for tree validation helper functions."""
def test_valid_tree_structure(self):
"""Test validation of a valid tree structure."""
tree_structure = {
"id": "root",
"type": "decision",
"question": "Is the server responding?",
"children": [
{
"id": "yes",
"type": "solution",
"solution": "Server is healthy",
"children": []
},
{
"id": "no",
"type": "action",
"action": "Restart the server",
"children": []
}
]
}
is_valid, errors = validate_tree_structure(tree_structure)
assert is_valid
assert len(errors) == 0
def test_empty_tree_structure(self):
"""Test validation of empty tree structure."""
is_valid, errors = validate_tree_structure({})
assert not is_valid
assert len(errors) > 0
assert any("empty" in error["message"].lower() for error in errors)
def test_missing_root_type(self):
"""Test validation when root node has no type."""
tree_structure = {
"id": "root",
"question": "Test?"
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("type" in error["field"] for error in errors)
def test_decision_node_missing_question(self):
"""Test validation when decision node has no question."""
tree_structure = {
"id": "root",
"type": "decision",
"children": []
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("question" in error["field"] for error in errors)
def test_decision_node_one_child(self):
"""Test validation when decision node has only one child."""
tree_structure = {
"id": "root",
"type": "decision",
"question": "Test?",
"children": [
{"id": "child1", "type": "solution", "solution": "Fix"}
]
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("at least 2" in error["message"] for error in errors)
def test_action_node_missing_action(self):
"""Test validation when action node has no action."""
tree_structure = {
"id": "root",
"type": "action",
"children": []
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("action" in error["field"] for error in errors)
def test_solution_node_missing_solution(self):
"""Test validation when solution node has no solution."""
tree_structure = {
"id": "root",
"type": "solution",
"children": []
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("solution" in error["field"] for error in errors)
def test_unknown_node_type(self):
"""Test validation with unknown node type."""
tree_structure = {
"id": "root",
"type": "unknown_type",
"children": []
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("unknown" in error["message"].lower() for error in errors)
def test_can_publish_with_empty_name(self):
"""Test can_publish with empty name."""
tree_structure = {"id": "root", "type": "solution", "solution": "Fix"}
can_publish, errors = can_publish_tree(tree_structure, "", None)
assert not can_publish
assert any("name" in error["field"] for error in errors)
def test_can_publish_valid_tree(self):
"""Test can_publish with valid tree and name."""
tree_structure = {"id": "root", "type": "solution", "solution": "Fix"}
can_publish, errors = can_publish_tree(tree_structure, "Valid Tree", "Description")
assert can_publish
assert len(errors) == 0
class TestDraftTreesAPI:
"""Test suite for draft trees API endpoints."""
async def test_create_draft_tree(self, client: AsyncClient, auth_headers):
"""Test creating a draft tree with incomplete structure."""
response = await client.post(
"/api/v1/trees",
json={
"name": "Draft Tree",
"description": "Work in progress",
"tree_structure": {"id": "root", "type": "decision"}, # Incomplete
"status": "draft"
},
headers=auth_headers
)
assert response.status_code == 201
data = response.json()
assert data["status"] == "draft"
assert data["name"] == "Draft Tree"
async def test_create_published_tree_with_validation(self, client: AsyncClient, auth_headers):
"""Test creating a published tree requires validation."""
response = await client.post(
"/api/v1/trees",
json={
"name": "Published Tree",
"description": "Complete tree",
"tree_structure": {
"id": "root",
"type": "decision",
"question": "Is it working?",
"children": [
{"id": "yes", "type": "solution", "solution": "Great!"},
{"id": "no", "type": "action", "action": "Fix it"}
]
},
"status": "published"
},
headers=auth_headers
)
assert response.status_code == 201
data = response.json()
assert data["status"] == "published"
async def test_create_published_tree_invalid_fails(self, client: AsyncClient, auth_headers):
"""Test creating published tree with invalid structure fails."""
response = await client.post(
"/api/v1/trees",
json={
"name": "Invalid Published Tree",
"tree_structure": {"id": "root", "type": "decision"}, # Missing question
"status": "published"
},
headers=auth_headers
)
assert response.status_code == 422
data = response.json()
assert "validation errors" in data["detail"]["message"].lower()
assert len(data["detail"]["errors"]) > 0
async def test_update_draft_to_published(self, client: AsyncClient, auth_headers, test_db, test_user):
"""Test updating a draft tree to published status."""
from uuid import UUID
# Create a draft tree
tree = Tree(
name="Draft to Published",
description="Test tree",
tree_structure={"id": "root", "type": "decision", "question": "Test?", "children": [
{"id": "yes", "type": "solution", "solution": "Yes"},
{"id": "no", "type": "solution", "solution": "No"}
]},
author_id=UUID(test_user["user_data"]["id"]),
account_id=UUID(test_user["user_data"]["account_id"]),
status='draft'
)
test_db.add(tree)
await test_db.commit()
await test_db.refresh(tree)
# Update to published
response = await client.put(
f"/api/v1/trees/{tree.id}",
json={"status": "published"},
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "published"
async def test_update_to_published_with_invalid_structure_fails(self, client: AsyncClient, auth_headers, test_db, test_user):
"""Test updating to published with invalid structure fails."""
from uuid import UUID
# Create a draft tree with invalid structure
tree = Tree(
name="Invalid Draft",
description="Test tree",
tree_structure={"id": "root", "type": "decision"}, # Missing question
author_id=UUID(test_user["user_data"]["id"]),
account_id=UUID(test_user["user_data"]["account_id"]),
status='draft'
)
test_db.add(tree)
await test_db.commit()
await test_db.refresh(tree)
# Try to update to published
response = await client.put(
f"/api/v1/trees/{tree.id}",
json={"status": "published"},
headers=auth_headers
)
assert response.status_code == 422
data = response.json()
assert "validation errors" in data["detail"]["message"].lower()
async def test_can_publish_endpoint(self, client: AsyncClient, auth_headers, test_db, test_user):
"""Test the can-publish validation endpoint."""
from uuid import UUID
# Create a valid draft tree
tree = Tree(
name="Valid Draft",
description="Test tree",
tree_structure={
"id": "root",
"type": "decision",
"question": "Is it working?",
"children": [
{"id": "yes", "type": "solution", "solution": "Great!"},
{"id": "no", "type": "action", "action": "Fix it"}
]
},
author_id=UUID(test_user["user_data"]["id"]),
account_id=UUID(test_user["user_data"]["account_id"]),
status='draft'
)
test_db.add(tree)
await test_db.commit()
await test_db.refresh(tree)
# Check if can publish
response = await client.post(
f"/api/v1/trees/{tree.id}/can-publish",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["can_publish"] is True
assert len(data["errors"]) == 0
async def test_can_publish_endpoint_invalid_tree(self, client: AsyncClient, auth_headers, test_db, test_user):
"""Test can-publish endpoint with invalid tree."""
from uuid import UUID
# Create an invalid draft tree
tree = Tree(
name="Invalid Draft",
description="Test tree",
tree_structure={"id": "root", "type": "decision"}, # Missing question
author_id=UUID(test_user["user_data"]["id"]),
account_id=UUID(test_user["user_data"]["account_id"]),
status='draft'
)
test_db.add(tree)
await test_db.commit()
await test_db.refresh(tree)
# Check if can publish
response = await client.post(
f"/api/v1/trees/{tree.id}/can-publish",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["can_publish"] is False
assert len(data["errors"]) > 0
assert any("question" in error["field"] for error in data["errors"])
async def test_list_trees_includes_status(self, client: AsyncClient, auth_headers):
"""Test that tree list includes status field."""
response = await client.get("/api/v1/trees", headers=auth_headers)
assert response.status_code == 200
trees = response.json()
if len(trees) > 0:
assert "status" in trees[0]
async def test_get_tree_includes_status(self, client: AsyncClient, auth_headers, test_db, test_user):
"""Test that get tree endpoint includes status field."""
from uuid import UUID
tree = Tree(
name="Test Tree",
description="Test",
tree_structure={"id": "root", "type": "solution", "solution": "Fix"},
author_id=UUID(test_user["user_data"]["id"]),
account_id=UUID(test_user["user_data"]["account_id"]),
status='published'
)
test_db.add(tree)
await test_db.commit()
await test_db.refresh(tree)
response = await client.get(f"/api/v1/trees/{tree.id}", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert "status" in data
assert data["status"] == "published"
async def test_migration_defaults_to_published(self, test_db):
"""Test that migration defaults existing trees to published status."""
# Create a tree without specifying status (relies on DB default)
from uuid import UUID, uuid4
tree = Tree(
name="Legacy Tree",
description="Created before status field",
tree_structure={"id": "root", "type": "solution", "solution": "Fix"},
author_id=None,
account_id=None
)
test_db.add(tree)
await test_db.commit()
await test_db.refresh(tree)
# Should default to 'published'
assert tree.status == 'published'