"""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