141 lines
6.0 KiB
Python
141 lines
6.0 KiB
Python
"""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
|