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>
386 lines
14 KiB
Python
386 lines
14 KiB
Python
"""Tests for save session as tree feature (Issue #17)."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from uuid import UUID
|
|
from datetime import datetime, timezone
|
|
|
|
from app.models.tree import Tree
|
|
from app.models.session import Session
|
|
from app.core.session_to_tree import (
|
|
convert_session_to_tree,
|
|
generate_tree_name_from_session,
|
|
_find_node_in_tree
|
|
)
|
|
|
|
|
|
class TestSessionToTreeConversion:
|
|
"""Test suite for session to tree conversion logic."""
|
|
|
|
def test_convert_empty_session(self):
|
|
"""Test converting a session with no path."""
|
|
tree_structure = convert_session_to_tree([], {}, [], [])
|
|
assert tree_structure["type"] == "solution"
|
|
assert "no recorded path" in tree_structure["title"].lower()
|
|
|
|
def test_convert_simple_linear_path(self):
|
|
"""Test converting a simple linear path."""
|
|
tree_snapshot = {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"question": "Is it working?",
|
|
"children": [
|
|
{"id": "yes", "type": "solution", "title": "Great!"},
|
|
{"id": "no", "type": "action", "title": "Fix it"}
|
|
]
|
|
}
|
|
path_taken = ["root", "no"]
|
|
decisions = [
|
|
{"node_id": "root", "answer": "No", "timestamp": datetime.now(timezone.utc).isoformat()},
|
|
{"node_id": "no", "action_performed": "Restarted service", "timestamp": datetime.now(timezone.utc).isoformat()}
|
|
]
|
|
|
|
result = convert_session_to_tree(path_taken, tree_snapshot, [], decisions)
|
|
|
|
assert result["type"] == "decision"
|
|
assert "Is it working?" in result["question"]
|
|
assert len(result["children"]) == 1
|
|
assert result["children"][0]["type"] == "action"
|
|
|
|
def test_convert_with_custom_steps(self):
|
|
"""Test converting a session with custom steps."""
|
|
tree_snapshot = {
|
|
"id": "root",
|
|
"type": "solution",
|
|
"title": "Done"
|
|
}
|
|
custom_step_id = "custom-123"
|
|
path_taken = ["root", custom_step_id]
|
|
custom_steps = [
|
|
{
|
|
"id": custom_step_id,
|
|
"type": "action",
|
|
"content": "Custom troubleshooting step",
|
|
"notes": "This worked!"
|
|
}
|
|
]
|
|
|
|
result = convert_session_to_tree(path_taken, tree_snapshot, custom_steps, [])
|
|
|
|
assert result["type"] == "solution"
|
|
assert len(result["children"]) == 1
|
|
custom_node = result["children"][0]
|
|
assert custom_node["type"] == "action"
|
|
assert "Custom troubleshooting step" in custom_node["title"]
|
|
|
|
def test_find_node_in_tree(self):
|
|
"""Test finding a node in nested tree structure."""
|
|
tree = {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"children": [
|
|
{
|
|
"id": "child1",
|
|
"type": "action",
|
|
"children": [
|
|
{"id": "grandchild", "type": "solution"}
|
|
]
|
|
},
|
|
{"id": "child2", "type": "solution"}
|
|
]
|
|
}
|
|
|
|
# Find root
|
|
assert _find_node_in_tree(tree, "root")["id"] == "root"
|
|
|
|
# Find child
|
|
assert _find_node_in_tree(tree, "child2")["type"] == "solution"
|
|
|
|
# Find grandchild
|
|
assert _find_node_in_tree(tree, "grandchild")["type"] == "solution"
|
|
|
|
# Not found
|
|
assert _find_node_in_tree(tree, "nonexistent") is None
|
|
|
|
def test_generate_tree_name_basic(self):
|
|
"""Test generating tree name without ticket or client."""
|
|
name = generate_tree_name_from_session("Network Troubleshooting")
|
|
assert "Network Troubleshooting" in name
|
|
assert "Session" in name
|
|
|
|
def test_generate_tree_name_with_ticket(self):
|
|
"""Test generating tree name with ticket number."""
|
|
name = generate_tree_name_from_session("VPN Issues", ticket_number="T-12345")
|
|
assert "VPN Issues" in name
|
|
assert "T-12345" in name
|
|
|
|
def test_generate_tree_name_with_client(self):
|
|
"""Test generating tree name with client name."""
|
|
name = generate_tree_name_from_session("Email Problems", client_name="Acme Corp")
|
|
assert "Email Problems" in name
|
|
assert "Acme Corp" in name
|
|
|
|
def test_generate_tree_name_full(self):
|
|
"""Test generating tree name with all parameters."""
|
|
name = generate_tree_name_from_session(
|
|
"Server Down",
|
|
ticket_number="INC-999",
|
|
client_name="Tech Startup"
|
|
)
|
|
assert "Server Down" in name
|
|
assert "INC-999" in name
|
|
assert "Tech Startup" in name
|
|
|
|
|
|
class TestSaveSessionAsTreeAPI:
|
|
"""Test suite for save session as tree API endpoint."""
|
|
|
|
async def test_save_session_as_tree_basic(self, client: AsyncClient, auth_headers, test_db, test_user):
|
|
"""Test basic save session as tree."""
|
|
from uuid import UUID
|
|
|
|
# Create a tree
|
|
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)
|
|
|
|
# Create a session
|
|
session = Session(
|
|
tree_id=tree.id,
|
|
user_id=UUID(test_user["user_data"]["id"]),
|
|
account_id=UUID(test_user["user_data"]["account_id"]),
|
|
tree_snapshot=tree.tree_structure,
|
|
path_taken=["root"],
|
|
decisions=[{"node_id": "root", "timestamp": datetime.now(timezone.utc).isoformat()}],
|
|
custom_steps=[]
|
|
)
|
|
test_db.add(session)
|
|
await test_db.commit()
|
|
await test_db.refresh(session)
|
|
|
|
# Save as tree
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session.id}/save-as-tree",
|
|
json={
|
|
"tree_name": "Saved Session Tree",
|
|
"description": "Saved from session",
|
|
"status": "draft"
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert "tree_id" in data
|
|
assert data["tree_name"] == "Saved Session Tree"
|
|
assert "draft" in data["message"]
|
|
|
|
async def test_save_session_auto_generated_name(self, client: AsyncClient, auth_headers, test_db, test_user):
|
|
"""Test save session with auto-generated tree name."""
|
|
from uuid import UUID
|
|
|
|
tree = Tree(
|
|
name="Original Tree",
|
|
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)
|
|
|
|
session = Session(
|
|
tree_id=tree.id,
|
|
user_id=UUID(test_user["user_data"]["id"]),
|
|
account_id=UUID(test_user["user_data"]["account_id"]),
|
|
tree_snapshot=tree.tree_structure,
|
|
path_taken=["root"],
|
|
decisions=[],
|
|
custom_steps=[],
|
|
ticket_number="T-123"
|
|
)
|
|
test_db.add(session)
|
|
await test_db.commit()
|
|
await test_db.refresh(session)
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session.id}/save-as-tree",
|
|
json={"status": "draft"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert "Original Tree" in data["tree_name"]
|
|
assert "T-123" in data["tree_name"]
|
|
|
|
async def test_save_session_as_published_requires_validation(self, client: AsyncClient, auth_headers, test_db, test_user):
|
|
"""Test saving session as published tree validates structure."""
|
|
from uuid import UUID
|
|
|
|
# Create a simple tree with just a solution (will convert to valid linear tree)
|
|
tree = Tree(
|
|
name="Test Tree",
|
|
tree_structure={"id": "root", "type": "solution", "title": "Fixed"},
|
|
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)
|
|
|
|
session = Session(
|
|
tree_id=tree.id,
|
|
user_id=UUID(test_user["user_data"]["id"]),
|
|
account_id=UUID(test_user["user_data"]["account_id"]),
|
|
tree_snapshot=tree.tree_structure,
|
|
path_taken=["root"],
|
|
decisions=[],
|
|
custom_steps=[]
|
|
)
|
|
test_db.add(session)
|
|
await test_db.commit()
|
|
await test_db.refresh(session)
|
|
|
|
# Try to save as published - should succeed with valid structure
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session.id}/save-as-tree",
|
|
json={
|
|
"tree_name": "Published Tree",
|
|
"status": "published"
|
|
},
|
|
headers=auth_headers
|
|
)
|
|
|
|
# Should succeed since the converted tree structure is valid (solution node)
|
|
assert response.status_code == 201
|
|
|
|
async def test_save_session_links_to_original_tree(self, client: AsyncClient, auth_headers, test_db, test_user):
|
|
"""Test that saved tree is linked to original via fork relationship."""
|
|
from uuid import UUID
|
|
|
|
tree = Tree(
|
|
name="Original Tree",
|
|
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)
|
|
|
|
session = Session(
|
|
tree_id=tree.id,
|
|
user_id=UUID(test_user["user_data"]["id"]),
|
|
account_id=UUID(test_user["user_data"]["account_id"]),
|
|
tree_snapshot=tree.tree_structure,
|
|
path_taken=["root"],
|
|
decisions=[],
|
|
custom_steps=[]
|
|
)
|
|
test_db.add(session)
|
|
await test_db.commit()
|
|
await test_db.refresh(session)
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{session.id}/save-as-tree",
|
|
json={"tree_name": "Forked Tree", "status": "draft"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
|
|
# Verify the fork relationship by fetching the tree
|
|
from sqlalchemy import select
|
|
result = await test_db.execute(
|
|
select(Tree).where(Tree.id == UUID(data["tree_id"]))
|
|
)
|
|
saved_tree = result.scalar_one()
|
|
assert saved_tree.parent_tree_id == tree.id
|
|
assert saved_tree.fork_depth == 1
|
|
|
|
async def test_save_session_not_found(self, client: AsyncClient, auth_headers):
|
|
"""Test saving non-existent session returns 404."""
|
|
from uuid import uuid4
|
|
|
|
response = await client.post(
|
|
f"/api/v1/sessions/{uuid4()}/save-as-tree",
|
|
json={"status": "draft"},
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_save_other_user_session_forbidden(self, client: AsyncClient, test_db, test_user):
|
|
"""Test cannot save another user's session."""
|
|
from uuid import UUID
|
|
from app.models.user import User
|
|
|
|
# Create a tree
|
|
tree = Tree(
|
|
name="Test Tree",
|
|
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)
|
|
|
|
# Create another user in same account
|
|
other_user = User(
|
|
email="other@example.com",
|
|
password_hash="hashed",
|
|
name="Other User",
|
|
is_active=True,
|
|
account_id=UUID(test_user["user_data"]["account_id"]),
|
|
account_role="engineer"
|
|
)
|
|
test_db.add(other_user)
|
|
await test_db.commit()
|
|
await test_db.refresh(other_user)
|
|
|
|
# Create session for the other user
|
|
session = Session(
|
|
tree_id=tree.id,
|
|
user_id=other_user.id,
|
|
account_id=UUID(test_user["user_data"]["account_id"]),
|
|
tree_snapshot=tree.tree_structure,
|
|
path_taken=["root"],
|
|
decisions=[],
|
|
custom_steps=[]
|
|
)
|
|
test_db.add(session)
|
|
await test_db.commit()
|
|
await test_db.refresh(session)
|
|
|
|
# Try to save the session as test_user (should fail - filtered by user_id)
|
|
from httpx import AsyncClient, ASGITransport
|
|
async with AsyncClient(transport=ASGITransport(app=client._transport.app), base_url="http://test") as test_client: # type: ignore
|
|
# Login as test_user
|
|
login_response = await test_client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": test_user["email"], "password": test_user["password"]}
|
|
)
|
|
token = login_response.json()["access_token"]
|
|
|
|
response = await test_client.post(
|
|
f"/api/v1/sessions/{session.id}/save-as-tree",
|
|
json={"status": "draft"},
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 404 # Session not found (filtered by user_id)
|