"""Tests for flow-to-library step sync.""" import pytest from uuid import uuid4 from app.core.step_sync import extract_steps_for_sync, resolve_step_visibility class TestResolveStepVisibility: """Test visibility resolution logic.""" def test_public_flow_gives_public_steps(self): result = resolve_step_visibility(is_public=True, account_id=None, node_override=None) assert result == 'public' def test_team_flow_gives_team_steps(self): result = resolve_step_visibility(is_public=False, account_id=uuid4(), node_override=None) assert result == 'team' def test_private_flow_gives_team_steps(self): result = resolve_step_visibility(is_public=False, account_id=None, node_override=None) assert result == 'team' def test_node_override_takes_precedence(self): result = resolve_step_visibility(is_public=True, account_id=None, node_override='team') assert result == 'team' def test_public_override_on_team_flow(self): result = resolve_step_visibility(is_public=False, account_id=uuid4(), node_override='public') assert result == 'public' class TestExtractStepsForSync: """Test step extraction from tree structures.""" def test_extracts_procedure_steps_from_procedural_flow(self): tree_structure = { "steps": [ {"id": "step_1", "type": "procedure_step", "title": "Verify prerequisites", "description": "Check all prereqs", "content_type": "action"}, {"id": "end_1", "type": "procedure_end", "title": "Done"}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) assert len(results) == 1 assert results[0]['source_node_id'] == 'step_1' assert results[0]['title'] == 'Verify prerequisites' assert results[0]['step_type'] == 'action' assert results[0]['content']['instructions'] == 'Check all prereqs' def test_skips_section_header_nodes(self): tree_structure = { "steps": [ {"id": "sec_1", "type": "section_header", "title": "Phase 1"}, {"id": "step_1", "type": "procedure_step", "title": "First step", "description": "Do this"}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) assert len(results) == 1 assert results[0]['source_node_id'] == 'step_1' def test_captures_section_header_as_group_label(self): tree_structure = { "steps": [ {"id": "sec_1", "type": "section_header", "title": "Cable Checks"}, {"id": "step_1", "type": "procedure_step", "title": "Check cable", "description": "Verify cable is seated"}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) assert results[0]['content']['group_label'] == 'Cable Checks' def test_normalizes_string_commands(self): tree_structure = { "steps": [ {"id": "step_1", "type": "procedure_step", "title": "Run command", "description": "Execute this", "commands": "ping 8.8.8.8"}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) assert results[0]['content']['commands'] == [{"label": "", "command": "ping 8.8.8.8", "command_type": None}] def test_normalizes_commandblock_commands(self): tree_structure = { "steps": [ {"id": "step_1", "type": "procedure_step", "title": "Run PS", "description": "Run powershell", "commands": [{"code": "Get-Service", "language": "powershell", "label": "Check services"}]}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) cmds = results[0]['content']['commands'] assert len(cmds) == 1 assert cmds[0]['command'] == 'Get-Service' assert cmds[0]['command_type'] == 'powershell' assert cmds[0]['label'] == 'Check services' def test_extracts_action_and_solution_from_troubleshooting(self): tree_structure = { "id": "root", "type": "decision", "question": "What is wrong?", "options": [{"id": "o1", "label": "Thing A", "next_node_id": "act_1"}], "children": [ {"id": "act_1", "type": "action", "title": "Fix thing A", "description": "Do the fix", "next_node_id": "sol_1", "children": [{"id": "sol_1", "type": "solution", "title": "All fixed", "description": "Problem resolved"}]}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='troubleshooting')) node_ids = {r['source_node_id'] for r in results} assert 'act_1' in node_ids assert 'sol_1' in node_ids types = {r['source_node_id']: r['step_type'] for r in results} assert types['act_1'] == 'action' assert types['sol_1'] == 'solution' def test_uses_title_as_instructions_fallback(self): tree_structure = { "steps": [ {"id": "step_1", "type": "procedure_step", "title": "Do the thing"}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) assert results[0]['content']['instructions'] == 'Do the thing' def test_empty_steps_list(self): tree_structure = {"steps": []} results = list(extract_steps_for_sync(tree_structure, tree_type='procedural')) assert results == [] def test_maintenance_treated_same_as_procedural(self): tree_structure = { "steps": [ {"id": "step_1", "type": "procedure_step", "title": "Maintenance step", "description": "Do maintenance"}, ] } results = list(extract_steps_for_sync(tree_structure, tree_type='maintenance')) assert len(results) == 1 class TestSyncOnPublish: """Integration tests — sync triggered by publishing a tree.""" @pytest.mark.asyncio async def test_publishing_procedural_tree_creates_library_steps( self, client, auth_headers ): # Create a procedural tree with two steps tree_resp = await client.post("/api/v1/trees", json={ "name": "Test Procedure", "tree_type": "procedural", "status": "draft", "tree_structure": { "steps": [ {"id": "step_1", "type": "procedure_step", "title": "First step", "description": "Do this first"}, {"id": "step_2", "type": "procedure_step", "title": "Second step", "description": "Do this second"}, {"id": "end_1", "type": "procedure_end", "title": "Done"}, ] } }, headers=auth_headers) assert tree_resp.status_code == 201 tree_id = tree_resp.json()["id"] # Publish the tree pub_resp = await client.put(f"/api/v1/trees/{tree_id}", json={"status": "published"}, headers=auth_headers) assert pub_resp.status_code == 200 # Check library has synced entries lib_resp = await client.get("/api/v1/steps", headers=auth_headers) assert lib_resp.status_code == 200 steps = lib_resp.json() synced = [s for s in steps if s.get("is_flow_synced")] assert len(synced) == 2 titles = {s["title"] for s in synced} assert "First step" in titles assert "Second step" in titles @pytest.mark.asyncio async def test_republishing_updates_existing_library_steps( self, client, auth_headers ): # Create a draft tree first, then publish tree_resp = await client.post("/api/v1/trees", json={ "name": "Update Test", "tree_type": "procedural", "status": "draft", "tree_structure": {"steps": [ {"id": "step_1", "type": "procedure_step", "title": "Original title", "description": "Original desc"}, {"id": "end_1", "type": "procedure_end", "title": "Done"}, ]} }, headers=auth_headers) tree_id = tree_resp.json()["id"] first_pub = await client.put(f"/api/v1/trees/{tree_id}", json={"status": "published"}, headers=auth_headers) assert first_pub.status_code == 200 # Republish with updated step title second_pub = await client.put(f"/api/v1/trees/{tree_id}", json={ "tree_structure": {"steps": [ {"id": "step_1", "type": "procedure_step", "title": "Updated title", "description": "Updated desc"}, {"id": "end_1", "type": "procedure_end", "title": "Done"}, ]}, "status": "published" }, headers=auth_headers) assert second_pub.status_code == 200 # Check library shows updated title (not a duplicate) lib_resp = await client.get("/api/v1/steps", headers=auth_headers) synced = [s for s in lib_resp.json() if s.get("is_flow_synced")] assert len(synced) == 1 assert synced[0]["title"] == "Updated title"