Files
resolutionflow/backend/tests/test_draft_trees.py
chihlasm 5acf94b6c2 fix: update tests to match action node schema (next_node_id, not children)
- Update _make_valid_tree() in test_ai_tree_validator to use next_node_id
  on action nodes (solution is a sibling, not a child)
- Fix test_dead_end_action_node → test_dead_end_decision_node (action nodes
  don't have child-based dead ends; dead ends are decision nodes with no children)
- Add test_action_missing_next_node_id for the new validation rule
- Update BRANCH_DETAIL_JSON in test_ai_endpoints to use next_node_id pattern
- Update test_draft_trees.py to use "title" field for action/solution nodes
  (tree_validation.py was updated this branch to require "title" not "action"/"solution")

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-02-22 23:18:10 -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",
"title": "Server is healthy",
"children": []
},
{
"id": "no",
"type": "action",
"title": "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", "title": "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_title(self):
"""Test validation when action node has no title."""
tree_structure = {
"id": "root",
"type": "action",
"children": []
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("title" in error["field"] for error in errors)
def test_solution_node_missing_title(self):
"""Test validation when solution node has no title."""
tree_structure = {
"id": "root",
"type": "solution",
"children": []
}
is_valid, errors = validate_tree_structure(tree_structure)
assert not is_valid
assert any("title" 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", "title": "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", "title": "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", "title": "Great!"},
{"id": "no", "type": "action", "title": "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", "title": "Yes"},
{"id": "no", "type": "solution", "title": "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", "title": "Great!"},
{"id": "no", "type": "action", "title": "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", "title": "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", "title": "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'