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