ccd06c9ed4
* feat: flexible intake — deferred variables + prepared sessions
Remove blocking intake form modal. Variables are now filled inline during
flow execution or pre-filled via prepared sessions. Adds PATCH /sessions/{id}/variables
endpoint, POST /sessions/prepare for session pre-staging, inline variable prompts
in StepDetail, editable Session Variables panel, and "Prepared for You" dashboard section.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* fix: pass treeData directly to startSession to avoid stale state
setTree(treeData) hasn't committed when startSession runs immediately
after, so tree is still null and getStepsFromTree returns []. This
caused the step detail area to render empty on new session start.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* feat: wire PrepareSessionModal entry point in Flow Library
Add "Prepare session" button (clipboard icon) to grid, list, and table
views for procedural/maintenance flows. Clicking fetches tree intake
fields and account members, then opens PrepareSessionModal.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
721 lines
25 KiB
Python
721 lines
25 KiB
Python
"""Tests for procedural flows feature (tree_type='procedural')."""
|
|
import pytest
|
|
from app.core.tree_validation import (
|
|
validate_procedural_structure,
|
|
can_publish_tree,
|
|
)
|
|
from app.schemas.tree import IntakeFormField, TreeCreate
|
|
|
|
|
|
# --- Helper Data ---
|
|
|
|
def make_valid_procedural_steps():
|
|
"""Return a valid procedural step array."""
|
|
return {
|
|
"steps": [
|
|
{
|
|
"id": "step-1",
|
|
"type": "procedure_step",
|
|
"title": "Configure Static IP",
|
|
"description": "Set the IP to [VAR:ip_address]",
|
|
"content_type": "action",
|
|
},
|
|
{
|
|
"id": "step-2",
|
|
"type": "procedure_step",
|
|
"title": "Install AD DS Role",
|
|
"description": "Add the AD DS server role",
|
|
"content_type": "action",
|
|
},
|
|
{
|
|
"id": "step-end",
|
|
"type": "procedure_end",
|
|
"title": "Procedure Complete",
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
def make_valid_intake_form():
|
|
"""Return valid intake form field dicts."""
|
|
return [
|
|
{
|
|
"variable_name": "server_name",
|
|
"label": "Server Name",
|
|
"field_type": "text",
|
|
"required": True,
|
|
"display_order": 1,
|
|
},
|
|
{
|
|
"variable_name": "ip_address",
|
|
"label": "IP Address",
|
|
"field_type": "ip_address",
|
|
"required": True,
|
|
"display_order": 2,
|
|
},
|
|
{
|
|
"variable_name": "notes",
|
|
"label": "Additional Notes",
|
|
"field_type": "textarea",
|
|
"required": False,
|
|
"display_order": 3,
|
|
},
|
|
]
|
|
|
|
|
|
# --- Procedural Validation Unit Tests ---
|
|
|
|
class TestValidateProceduralStructure:
|
|
"""Unit tests for validate_procedural_structure()."""
|
|
|
|
def test_valid_procedural_tree(self):
|
|
is_valid, errors = validate_procedural_structure(make_valid_procedural_steps())
|
|
assert is_valid
|
|
assert errors == []
|
|
|
|
def test_empty_structure(self):
|
|
is_valid, errors = validate_procedural_structure({})
|
|
assert not is_valid
|
|
assert any("empty" in e["message"].lower() for e in errors)
|
|
|
|
def test_none_structure(self):
|
|
is_valid, errors = validate_procedural_structure(None)
|
|
assert not is_valid
|
|
|
|
def test_missing_steps_array(self):
|
|
is_valid, errors = validate_procedural_structure({"name": "test"})
|
|
assert not is_valid
|
|
assert any("steps" in e["message"] for e in errors)
|
|
|
|
def test_empty_steps_array(self):
|
|
is_valid, errors = validate_procedural_structure({"steps": []})
|
|
assert not is_valid
|
|
|
|
def test_step_missing_id(self):
|
|
structure = {
|
|
"steps": [
|
|
{"type": "procedure_step", "title": "Step 1"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("id" in e["field"] for e in errors)
|
|
|
|
def test_step_missing_type(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "title": "Step 1"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("type" in e["field"] for e in errors)
|
|
|
|
def test_step_missing_title(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("title" in e["field"] for e in errors)
|
|
|
|
def test_invalid_step_type(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "decision", "title": "Bad Type"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("Invalid step type" in e["message"] for e in errors)
|
|
|
|
def test_no_end_step(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1"},
|
|
{"id": "s2", "type": "procedure_step", "title": "Step 2"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("procedure_end" in e["message"] for e in errors)
|
|
|
|
def test_multiple_end_steps(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1"},
|
|
{"id": "end1", "type": "procedure_end", "title": "End 1"},
|
|
{"id": "end2", "type": "procedure_end", "title": "End 2"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
|
|
def test_end_step_not_last(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "end", "type": "procedure_end", "title": "End"},
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("last step" in e["message"] for e in errors)
|
|
|
|
def test_duplicate_step_ids(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1"},
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 2"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("Duplicate" in e["message"] for e in errors)
|
|
|
|
def test_invalid_content_type(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1", "content_type": "bad_type"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert not is_valid
|
|
assert any("content_type" in e["field"] for e in errors)
|
|
|
|
def test_valid_content_types(self):
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1", "content_type": "action"},
|
|
{"id": "s2", "type": "procedure_step", "title": "Step 2", "content_type": "informational"},
|
|
{"id": "s3", "type": "procedure_step", "title": "Step 3", "content_type": "verification"},
|
|
{"id": "s4", "type": "procedure_step", "title": "Step 4", "content_type": "warning"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert is_valid
|
|
|
|
def test_single_step_with_end(self):
|
|
"""Minimal valid procedural tree: one step + end."""
|
|
structure = {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Only Step"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
}
|
|
is_valid, errors = validate_procedural_structure(structure)
|
|
assert is_valid
|
|
|
|
|
|
# --- can_publish_tree dispatch ---
|
|
|
|
class TestCanPublishTreeDispatch:
|
|
"""Test can_publish_tree dispatches correctly by tree_type."""
|
|
|
|
def test_troubleshooting_uses_tree_validation(self):
|
|
"""Default tree_type uses troubleshooting validation."""
|
|
structure = {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"question": "Test?",
|
|
"children": [
|
|
{"id": "y", "type": "solution", "title": "Yes"},
|
|
{"id": "n", "type": "solution", "title": "No"},
|
|
]
|
|
}
|
|
can, errors = can_publish_tree(structure, "My Tree", tree_type="troubleshooting")
|
|
assert can
|
|
|
|
def test_procedural_uses_procedural_validation(self):
|
|
can, errors = can_publish_tree(
|
|
make_valid_procedural_steps(),
|
|
"DC Build Procedure",
|
|
tree_type="procedural",
|
|
)
|
|
assert can
|
|
|
|
def test_procedural_rejects_troubleshooting_structure(self):
|
|
"""A troubleshooting structure should fail procedural validation."""
|
|
ts_structure = {
|
|
"id": "root",
|
|
"type": "decision",
|
|
"question": "Test?",
|
|
}
|
|
can, errors = can_publish_tree(ts_structure, "My Tree", tree_type="procedural")
|
|
assert not can
|
|
|
|
def test_procedural_validates_intake_form(self):
|
|
can, errors = can_publish_tree(
|
|
make_valid_procedural_steps(),
|
|
"DC Build",
|
|
tree_type="procedural",
|
|
intake_form=make_valid_intake_form(),
|
|
)
|
|
assert can
|
|
|
|
def test_procedural_rejects_duplicate_variable_names(self):
|
|
intake = [
|
|
{"variable_name": "name", "label": "Name", "field_type": "text", "required": True, "display_order": 1},
|
|
{"variable_name": "name", "label": "Name 2", "field_type": "text", "required": False, "display_order": 2},
|
|
]
|
|
can, errors = can_publish_tree(
|
|
make_valid_procedural_steps(),
|
|
"DC Build",
|
|
tree_type="procedural",
|
|
intake_form=intake,
|
|
)
|
|
assert not can
|
|
assert any("Duplicate" in e["message"] for e in errors)
|
|
|
|
def test_procedural_rejects_select_without_options(self):
|
|
intake = [
|
|
{"variable_name": "role", "label": "Server Role", "field_type": "select", "required": True, "display_order": 1},
|
|
]
|
|
can, errors = can_publish_tree(
|
|
make_valid_procedural_steps(),
|
|
"DC Build",
|
|
tree_type="procedural",
|
|
intake_form=intake,
|
|
)
|
|
assert not can
|
|
assert any("option" in e["message"].lower() for e in errors)
|
|
|
|
def test_empty_name_blocks_publish(self):
|
|
can, errors = can_publish_tree(
|
|
make_valid_procedural_steps(),
|
|
"",
|
|
tree_type="procedural",
|
|
)
|
|
assert not can
|
|
assert any("name" in e["field"] for e in errors)
|
|
|
|
|
|
# --- IntakeFormField Pydantic Schema Tests ---
|
|
|
|
class TestIntakeFormFieldSchema:
|
|
"""Test IntakeFormField Pydantic validation."""
|
|
|
|
def test_valid_text_field(self):
|
|
field = IntakeFormField(
|
|
variable_name="server_name",
|
|
label="Server Name",
|
|
field_type="text",
|
|
required=True,
|
|
display_order=1,
|
|
)
|
|
assert field.variable_name == "server_name"
|
|
|
|
def test_invalid_variable_name_uppercase(self):
|
|
with pytest.raises(Exception):
|
|
IntakeFormField(
|
|
variable_name="ServerName",
|
|
label="Server Name",
|
|
field_type="text",
|
|
required=True,
|
|
display_order=1,
|
|
)
|
|
|
|
def test_invalid_variable_name_starts_with_number(self):
|
|
with pytest.raises(Exception):
|
|
IntakeFormField(
|
|
variable_name="1server",
|
|
label="Server Name",
|
|
field_type="text",
|
|
required=True,
|
|
display_order=1,
|
|
)
|
|
|
|
def test_valid_variable_name_with_underscores(self):
|
|
field = IntakeFormField(
|
|
variable_name="ip_address_v4",
|
|
label="IPv4 Address",
|
|
field_type="ip_address",
|
|
required=True,
|
|
display_order=1,
|
|
)
|
|
assert field.variable_name == "ip_address_v4"
|
|
|
|
def test_select_requires_options(self):
|
|
with pytest.raises(Exception):
|
|
IntakeFormField(
|
|
variable_name="role",
|
|
label="Role",
|
|
field_type="select",
|
|
required=True,
|
|
display_order=1,
|
|
)
|
|
|
|
def test_select_with_options_valid(self):
|
|
field = IntakeFormField(
|
|
variable_name="role",
|
|
label="Role",
|
|
field_type="select",
|
|
required=True,
|
|
options=["AD DS", "DNS", "DHCP"],
|
|
display_order=1,
|
|
)
|
|
assert field.options == ["AD DS", "DNS", "DHCP"]
|
|
|
|
def test_multi_select_requires_options(self):
|
|
with pytest.raises(Exception):
|
|
IntakeFormField(
|
|
variable_name="roles",
|
|
label="Roles",
|
|
field_type="multi_select",
|
|
required=True,
|
|
display_order=1,
|
|
)
|
|
|
|
def test_checkbox_field(self):
|
|
field = IntakeFormField(
|
|
variable_name="confirm_backup",
|
|
label="Backup confirmed?",
|
|
field_type="checkbox",
|
|
required=False,
|
|
display_order=1,
|
|
)
|
|
assert field.field_type == "checkbox"
|
|
|
|
|
|
# --- TreeCreate Schema with Procedural Fields ---
|
|
|
|
class TestTreeCreateProceduralSchema:
|
|
"""Test TreeCreate schema with tree_type and intake_form."""
|
|
|
|
def test_defaults_to_troubleshooting(self):
|
|
tree = TreeCreate(
|
|
name="Test",
|
|
tree_structure={"id": "root", "type": "decision", "question": "Test?"},
|
|
)
|
|
assert tree.tree_type == "troubleshooting"
|
|
assert tree.intake_form is None
|
|
|
|
def test_procedural_with_intake_form(self):
|
|
tree = TreeCreate(
|
|
name="DC Build",
|
|
tree_type="procedural",
|
|
tree_structure=make_valid_procedural_steps(),
|
|
intake_form=[
|
|
IntakeFormField(
|
|
variable_name="server_name",
|
|
label="Server Name",
|
|
field_type="text",
|
|
required=True,
|
|
display_order=1,
|
|
),
|
|
],
|
|
)
|
|
assert tree.tree_type == "procedural"
|
|
assert len(tree.intake_form) == 1
|
|
|
|
def test_duplicate_variable_names_rejected(self):
|
|
with pytest.raises(Exception):
|
|
TreeCreate(
|
|
name="DC Build",
|
|
tree_type="procedural",
|
|
tree_structure=make_valid_procedural_steps(),
|
|
intake_form=[
|
|
IntakeFormField(
|
|
variable_name="name",
|
|
label="Name",
|
|
field_type="text",
|
|
required=True,
|
|
display_order=1,
|
|
),
|
|
IntakeFormField(
|
|
variable_name="name",
|
|
label="Name Again",
|
|
field_type="text",
|
|
required=False,
|
|
display_order=2,
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
# --- API Integration Tests ---
|
|
|
|
@pytest.mark.asyncio
|
|
class TestProceduralFlowsAPI:
|
|
"""Integration tests for procedural flow CRUD via API."""
|
|
|
|
async def test_create_procedural_draft(self, client, auth_headers):
|
|
"""Create a procedural flow as draft."""
|
|
tree_data = {
|
|
"name": "DC Build Procedure",
|
|
"description": "Domain Controller setup procedure",
|
|
"tree_type": "procedural",
|
|
"status": "draft",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
"intake_form": make_valid_intake_form(),
|
|
}
|
|
response = await client.post(
|
|
"/api/v1/trees",
|
|
json=tree_data,
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200 or response.status_code == 201
|
|
data = response.json()
|
|
assert data["tree_type"] == "procedural"
|
|
assert data["intake_form"] is not None
|
|
assert len(data["intake_form"]) == 3
|
|
|
|
async def test_create_procedural_published(self, client, auth_headers):
|
|
"""Create a procedural flow and publish it (validation runs)."""
|
|
tree_data = {
|
|
"name": "M365 User Onboarding",
|
|
"tree_type": "procedural",
|
|
"status": "published",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
}
|
|
response = await client.post(
|
|
"/api/v1/trees",
|
|
json=tree_data,
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200 or response.status_code == 201
|
|
|
|
async def test_create_procedural_published_invalid_fails(self, client, auth_headers):
|
|
"""Publish should fail if procedural structure is invalid."""
|
|
tree_data = {
|
|
"name": "Bad Procedure",
|
|
"tree_type": "procedural",
|
|
"status": "published",
|
|
"tree_structure": {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step 1"},
|
|
# No end step
|
|
]
|
|
},
|
|
}
|
|
response = await client.post(
|
|
"/api/v1/trees",
|
|
json=tree_data,
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
async def test_list_trees_filter_by_type(self, client, auth_headers):
|
|
"""Create both types and filter by tree_type."""
|
|
# Create troubleshooting
|
|
await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "Troubleshooting Tree",
|
|
"status": "draft",
|
|
"tree_structure": {"id": "root", "type": "decision", "question": "Test?"},
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
# Create procedural
|
|
await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "Procedure Flow",
|
|
"tree_type": "procedural",
|
|
"status": "draft",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Filter by procedural
|
|
response = await client.get(
|
|
"/api/v1/trees?tree_type=procedural",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(t["tree_type"] == "procedural" for t in data)
|
|
|
|
# Filter by troubleshooting
|
|
response = await client.get(
|
|
"/api/v1/trees?tree_type=troubleshooting",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert all(t["tree_type"] == "troubleshooting" for t in data)
|
|
|
|
async def test_update_procedural_tree(self, client, auth_headers):
|
|
"""Update a procedural tree's intake form."""
|
|
# Create
|
|
create_resp = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "Procedure",
|
|
"tree_type": "procedural",
|
|
"status": "draft",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Update with intake form
|
|
update_resp = await client.put(
|
|
f"/api/v1/trees/{tree_id}",
|
|
json={
|
|
"intake_form": make_valid_intake_form(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert update_resp.status_code == 200
|
|
data = update_resp.json()
|
|
assert data["intake_form"] is not None
|
|
|
|
async def test_start_session_procedural_with_variables(self, client, auth_headers):
|
|
"""Start a procedural session with intake form variables."""
|
|
# Create published procedural tree with intake form
|
|
create_resp = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "DC Build",
|
|
"tree_type": "procedural",
|
|
"status": "published",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
"intake_form": make_valid_intake_form(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert create_resp.status_code in (200, 201)
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Start session with variables
|
|
session_resp = await client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"tree_id": tree_id,
|
|
"session_variables": {
|
|
"server_name": "DC01",
|
|
"ip_address": "192.168.1.10",
|
|
},
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert session_resp.status_code in (200, 201)
|
|
session_data = session_resp.json()
|
|
assert session_data["session_variables"]["server_name"] == "DC01"
|
|
assert session_data["tree_snapshot"]["tree_type"] == "procedural"
|
|
|
|
async def test_start_session_procedural_deferred_variables(self, client, auth_headers):
|
|
"""Starting a procedural session without required intake fields should succeed (deferred variables)."""
|
|
# Create published procedural tree with required intake form
|
|
create_resp = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "DC Build",
|
|
"tree_type": "procedural",
|
|
"status": "published",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
"intake_form": make_valid_intake_form(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Start without required fields — should succeed (deferred variables)
|
|
session_resp = await client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"tree_id": tree_id,
|
|
# Missing server_name and ip_address — will be filled inline later
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert session_resp.status_code == 201
|
|
session_id = session_resp.json()["id"]
|
|
|
|
# Fill variables via PATCH endpoint
|
|
patch_resp = await client.patch(
|
|
f"/api/v1/sessions/{session_id}/variables",
|
|
json={"variables": {"server_name": "DC-01", "ip_address": "10.0.0.1"}},
|
|
headers=auth_headers,
|
|
)
|
|
assert patch_resp.status_code == 200
|
|
assert patch_resp.json()["session_variables"]["server_name"] == "DC-01"
|
|
assert patch_resp.json()["session_variables"]["ip_address"] == "10.0.0.1"
|
|
|
|
async def test_start_session_procedural_optional_fields_ok(self, client, auth_headers):
|
|
"""Starting a session with only required fields (optional missing) should work."""
|
|
create_resp = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "DC Build",
|
|
"tree_type": "procedural",
|
|
"status": "published",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
"intake_form": make_valid_intake_form(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Start with only required fields (notes is optional)
|
|
session_resp = await client.post(
|
|
"/api/v1/sessions",
|
|
json={
|
|
"tree_id": tree_id,
|
|
"session_variables": {
|
|
"server_name": "DC01",
|
|
"ip_address": "192.168.1.10",
|
|
# notes is optional, not provided
|
|
},
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert session_resp.status_code in (200, 201)
|
|
|
|
async def test_fork_preserves_tree_type_and_intake_form(self, client, auth_headers):
|
|
"""Forking a procedural tree should preserve tree_type and intake_form."""
|
|
# Create procedural tree
|
|
create_resp = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "DC Build Original",
|
|
"tree_type": "procedural",
|
|
"status": "published",
|
|
"tree_structure": make_valid_procedural_steps(),
|
|
"intake_form": make_valid_intake_form(),
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
tree_id = create_resp.json()["id"]
|
|
|
|
# Fork it
|
|
fork_resp = await client.post(
|
|
f"/api/v1/trees/{tree_id}/fork",
|
|
json={"fork_reason": "Customized for Client X"},
|
|
headers=auth_headers,
|
|
)
|
|
assert fork_resp.status_code in (200, 201)
|
|
fork_data = fork_resp.json()
|
|
assert fork_data["tree_type"] == "procedural"
|
|
assert fork_data["intake_form"] is not None
|
|
assert len(fork_data["intake_form"]) == 3
|
|
|
|
async def test_existing_trees_default_troubleshooting(self, client, auth_headers):
|
|
"""Trees created without tree_type should default to troubleshooting."""
|
|
response = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "Legacy Tree",
|
|
"status": "draft",
|
|
"tree_structure": {"id": "root", "type": "decision", "question": "Test?"},
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code in (200, 201)
|
|
data = response.json()
|
|
assert data["tree_type"] == "troubleshooting"
|
|
assert data.get("intake_form") is None
|