"""Simple tests for src/commands/flow.py.
Focuses on basic functionality that can be easily tested to achieve coverage.
"""
import pytest
from src.commands.flow import (
BreakCommand,
ComparisonOperator,
ConditionalCommand,
ConditionType,
LoopCommand,
LoopType,
)
from src.core.types import CommandId, CommandParameters, ExecutionContext
class TestConditionType:
"""Test ConditionType enum values."""
def test_condition_type_enum_values(self):
"""Test ConditionType enum has expected values."""
assert ConditionType.EQUALS.value == "equals"
assert ConditionType.NOT_EQUALS.value == "not_equals"
assert ConditionType.CONTAINS.value == "contains"
assert ConditionType.NOT_CONTAINS.value == "not_contains"
assert ConditionType.GREATER_THAN.value == "greater_than"
assert ConditionType.LESS_THAN.value == "less_than"
assert ConditionType.GREATER_EQUAL.value == "greater_equal"
assert ConditionType.LESS_EQUAL.value == "less_equal"
assert ConditionType.REGEX_MATCH.value == "regex_match"
assert ConditionType.IS_EMPTY.value == "is_empty"
assert ConditionType.IS_NOT_EMPTY.value == "is_not_empty"
class TestLoopType:
"""Test LoopType enum values."""
def test_loop_type_enum_values(self):
"""Test LoopType enum has expected values."""
assert LoopType.FOR_COUNT.value == "for_count"
assert LoopType.WHILE_CONDITION.value == "while_condition"
assert LoopType.FOR_EACH.value == "for_each"
class TestComparisonOperator:
"""Test ComparisonOperator enum values."""
def test_comparison_operator_enum_values(self):
"""Test ComparisonOperator enum has expected values."""
assert ComparisonOperator.EQ.value == "=="
assert ComparisonOperator.NE.value == "!="
assert ComparisonOperator.GT.value == ">"
assert ComparisonOperator.LT.value == "<"
assert ComparisonOperator.GE.value == ">="
assert ComparisonOperator.LE.value == "<="
class TestConditionalCommand:
"""Test ConditionalCommand basic functionality."""
@pytest.fixture
def sample_context(self):
"""Create sample ExecutionContext for testing."""
return ExecutionContext.create_test_context()
def test_conditional_command_creation(self):
"""Test ConditionalCommand creation with valid data."""
command = ConditionalCommand(
command_id=CommandId("cond-001"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.EQUALS,
"left_value": "test",
"right_value": "test",
"then_commands": [],
"else_commands": [],
}
),
)
assert command.command_id == CommandId("cond-001")
assert command.parameters.get("condition_type") == ConditionType.EQUALS
assert command.parameters.get("left_value") == "test"
assert command.parameters.get("right_value") == "test"
assert command.parameters.get("then_commands") == []
assert command.parameters.get("else_commands") == []
def test_conditional_command_basic_execution(self, sample_context):
"""Test ConditionalCommand basic execution."""
command = ConditionalCommand(
command_id=CommandId("cond-basic"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.EQUALS,
"left_value": "hello",
"right_value": "hello",
"then_commands": [],
"else_commands": [],
}
),
)
result = command.execute(sample_context)
assert result is not None
assert hasattr(result, "success")
def test_conditional_command_validation(self):
"""Test ConditionalCommand validation."""
command = ConditionalCommand(
command_id=CommandId("cond-valid"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.EQUALS,
"left_value": "test",
"right_value": "test",
"then_commands": [],
"else_commands": [],
}
),
)
result = command.validate()
assert isinstance(result, bool)
def test_conditional_command_security_risk_level(self):
"""Test ConditionalCommand security risk level method."""
command = ConditionalCommand(
command_id=CommandId("cond-security"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.EQUALS,
"left_value": "test",
"right_value": "test",
"then_commands": [],
"else_commands": [],
}
),
)
risk_level = command.get_security_risk_level()
assert isinstance(risk_level, str)
def test_conditional_command_get_required_permissions(self):
"""Test ConditionalCommand get_required_permissions method."""
command = ConditionalCommand(
command_id=CommandId("cond-perms"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.EQUALS,
"left_value": "test",
"right_value": "test",
"then_commands": [],
"else_commands": [],
}
),
)
permissions = command.get_required_permissions()
assert hasattr(permissions, "__iter__") # Should be iterable (set or list)
def test_conditional_command_different_condition_types(self, sample_context):
"""Test ConditionalCommand with different condition types."""
# Test NOT_EQUALS
command_ne = ConditionalCommand(
command_id=CommandId("cond-ne"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.NOT_EQUALS,
"left_value": "hello",
"right_value": "world",
"then_commands": [],
"else_commands": [],
}
),
)
result_ne = command_ne.execute(sample_context)
assert result_ne is not None
# Test CONTAINS
command_contains = ConditionalCommand(
command_id=CommandId("cond-contains"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.CONTAINS,
"left_value": "hello world",
"right_value": "world",
"then_commands": [],
"else_commands": [],
}
),
)
result_contains = command_contains.execute(sample_context)
assert result_contains is not None
def test_conditional_command_numeric_comparisons(self, sample_context):
"""Test ConditionalCommand with numeric comparison types."""
# Test GREATER_THAN
command_gt = ConditionalCommand(
command_id=CommandId("cond-gt"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.GREATER_THAN,
"left_value": "10",
"right_value": "5",
"then_commands": [],
"else_commands": [],
}
),
)
result_gt = command_gt.execute(sample_context)
assert result_gt is not None
# Test LESS_THAN
command_lt = ConditionalCommand(
command_id=CommandId("cond-lt"),
parameters=CommandParameters(
data={
"condition_type": ConditionType.LESS_THAN,
"left_value": "3",
"right_value": "7",
"then_commands": [],
"else_commands": [],
}
),
)
result_lt = command_lt.execute(sample_context)
assert result_lt is not None
class TestLoopCommand:
"""Test LoopCommand basic functionality."""
@pytest.fixture
def sample_context(self):
"""Create sample ExecutionContext for testing."""
return ExecutionContext.create_test_context()
def test_loop_command_creation(self):
"""Test LoopCommand creation with valid data."""
command = LoopCommand(
command_id=CommandId("loop-001"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 3,
"commands": [],
}
),
)
assert command.command_id == CommandId("loop-001")
assert command.parameters.get("loop_type") == LoopType.FOR_COUNT
assert command.parameters.get("count") == 3
assert command.parameters.get("commands") == []
def test_loop_command_basic_execution(self, sample_context):
"""Test LoopCommand basic execution."""
command = LoopCommand(
command_id=CommandId("loop-basic"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 3,
"commands": [],
}
),
)
result = command.execute(sample_context)
assert result is not None
assert hasattr(result, "success")
def test_loop_command_validation(self):
"""Test LoopCommand validation."""
command = LoopCommand(
command_id=CommandId("loop-valid"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 5,
"commands": [],
}
),
)
result = command.validate()
assert isinstance(result, bool)
def test_loop_command_security_risk_level(self):
"""Test LoopCommand security risk level method."""
command = LoopCommand(
command_id=CommandId("loop-security"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 2,
"commands": [],
}
),
)
risk_level = command.get_security_risk_level()
assert isinstance(risk_level, str)
def test_loop_command_get_required_permissions(self):
"""Test LoopCommand get_required_permissions method."""
command = LoopCommand(
command_id=CommandId("loop-perms"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 2,
"commands": [],
}
),
)
permissions = command.get_required_permissions()
assert hasattr(permissions, "__iter__") # Should be iterable (set or list)
def test_loop_command_different_types(self, sample_context):
"""Test LoopCommand with different loop types."""
# Test WHILE_CONDITION
command_while = LoopCommand(
command_id=CommandId("loop-while"),
parameters=CommandParameters(
data={
"loop_type": LoopType.WHILE_CONDITION,
"condition": "false",
"commands": [],
}
),
)
result_while = command_while.execute(sample_context)
assert result_while is not None
# Test FOR_EACH
command_each = LoopCommand(
command_id=CommandId("loop-each"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_EACH,
"items": ["a", "b", "c"],
"commands": [],
}
),
)
result_each = command_each.execute(sample_context)
assert result_each is not None
def test_loop_command_small_count(self, sample_context):
"""Test LoopCommand with small count."""
command = LoopCommand(
command_id=CommandId("loop-small"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 1,
"commands": [],
}
),
)
result = command.execute(sample_context)
assert result is not None
def test_loop_command_zero_count(self, sample_context):
"""Test LoopCommand with zero count."""
command = LoopCommand(
command_id=CommandId("loop-zero"),
parameters=CommandParameters(
data={
"loop_type": LoopType.FOR_COUNT,
"count": 0,
"commands": [],
}
),
)
result = command.execute(sample_context)
assert result is not None
class TestBreakCommand:
"""Test BreakCommand basic functionality."""
@pytest.fixture
def sample_context(self):
"""Create sample ExecutionContext for testing."""
return ExecutionContext.create_test_context()
def test_break_command_creation(self):
"""Test BreakCommand creation with valid data."""
command = BreakCommand(
command_id=CommandId("break-001"),
parameters=CommandParameters(data={}),
)
assert command.command_id == CommandId("break-001")
def test_break_command_basic_execution(self, sample_context):
"""Test BreakCommand basic execution."""
command = BreakCommand(
command_id=CommandId("break-basic"),
parameters=CommandParameters(data={}),
)
result = command.execute(sample_context)
assert result is not None
assert hasattr(result, "success")
def test_break_command_validation(self):
"""Test BreakCommand validation."""
command = BreakCommand(
command_id=CommandId("break-valid"),
parameters=CommandParameters(data={}),
)
result = command.validate()
assert isinstance(result, bool)
def test_break_command_security_risk_level(self):
"""Test BreakCommand security risk level method."""
command = BreakCommand(
command_id=CommandId("break-security"),
parameters=CommandParameters(data={}),
)
risk_level = command.get_security_risk_level()
assert isinstance(risk_level, str)
def test_break_command_get_required_permissions(self):
"""Test BreakCommand get_required_permissions method."""
command = BreakCommand(
command_id=CommandId("break-perms"),
parameters=CommandParameters(data={}),
)
permissions = command.get_required_permissions()
assert hasattr(permissions, "__iter__") # Should be iterable (set or list)
def test_break_command_minimal_parameters(self, sample_context):
"""Test BreakCommand with minimal parameters."""
command = BreakCommand(
command_id=CommandId("break-minimal"),
parameters=CommandParameters(data={}),
)
result = command.execute(sample_context)
assert result is not None
def test_break_command_with_additional_parameters(self, sample_context):
"""Test BreakCommand with additional parameters."""
command = BreakCommand(
command_id=CommandId("break-extra"),
parameters=CommandParameters(
data={
"message": "Custom break message",
"level": "inner",
}
),
)
result = command.execute(sample_context)
assert result is not None
