758cd61621
Service layer (production code): - branch_manager: set account_id on SessionBranch (root + fork) and ForkPoint from session.account_id; load session in create_fork for this purpose - handoff_manager: set account_id on SessionHandoff from session.account_id - ai_suggestions endpoint: set account_id on AISuggestion from current_user - steps endpoint (/feedback): set account_id on StepRating from current_user - ratings endpoint: set account_id on StepRating from current_user Test infrastructure: - conftest.py: seed PLATFORM_ACCOUNT_ID (00000000-...-0001) account after Base.metadata.create_all so global categories and gallery items have a valid FK - test_rls_isolation: add _ensure_rls_schema fixture that runs 'alembic upgrade head' before module tests — previous function-scoped test_db fixtures drop the schema, leaving the RLS tests with no tables - test_branding: create Account before User in helper functions - test_admin_gallery: set account_id=PLATFORM_ACCOUNT_ID on Tree/ScriptTemplate - test_public_templates: set account_id=PLATFORM_ACCOUNT_ID on Tree, ScriptTemplate, TreeCategory - test_resolution_outputs: set account_id=session.account_id on SessionResolutionOutput - test_analytics_phase5: set account_id on PsaPostLog - test_draft_trees: replace account_id=None with PLATFORM_ACCOUNT_ID in migration default test (NOT NULL now enforced) - test_maintenance_schedules: set account_id on other_tree - test_save_session_as_tree: set account_id on all 5 Session() constructors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
351 lines
13 KiB
Python
351 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
|
|
_platform_id = UUID("00000000-0000-0000-0000-000000000001")
|
|
tree = Tree(
|
|
name="Legacy Tree",
|
|
description="Created before status field",
|
|
tree_structure={"id": "root", "type": "solution", "title": "Fix"},
|
|
author_id=None,
|
|
account_id=_platform_id,
|
|
)
|
|
test_db.add(tree)
|
|
await test_db.commit()
|
|
await test_db.refresh(tree)
|
|
|
|
# Should default to 'published'
|
|
assert tree.status == 'published'
|