224 lines
7.9 KiB
Python
224 lines
7.9 KiB
Python
"""Tests for task module"""
|
|
import pytest
|
|
from luxx.services.task import (
|
|
Task,
|
|
Step,
|
|
TaskGraph,
|
|
TaskService,
|
|
TaskStatus,
|
|
StepStatus,
|
|
task_service
|
|
)
|
|
|
|
|
|
class TestStep:
|
|
"""Tests for Step dataclass"""
|
|
|
|
def test_step_creation(self):
|
|
"""Should create step with required fields"""
|
|
step = Step(id="step_1", name="Test Step")
|
|
assert step.id == "step_1"
|
|
assert step.name == "Test Step"
|
|
assert step.status == StepStatus.PENDING
|
|
assert step.depends_on == []
|
|
|
|
def test_step_with_dependencies(self):
|
|
"""Should create step with dependencies"""
|
|
step = Step(
|
|
id="step_2",
|
|
name="Dependent Step",
|
|
depends_on=["step_1"]
|
|
)
|
|
assert "step_1" in step.depends_on
|
|
|
|
def test_step_to_dict(self):
|
|
"""Should convert step to dict"""
|
|
step = Step(id="step_1", name="Test")
|
|
result = step.to_dict()
|
|
assert isinstance(result, dict)
|
|
assert result["id"] == "step_1"
|
|
assert result["name"] == "Test"
|
|
assert result["status"] == "pending"
|
|
|
|
|
|
class TestTask:
|
|
"""Tests for Task dataclass"""
|
|
|
|
def test_task_creation(self):
|
|
"""Should create task with required fields"""
|
|
task = Task(id="task_1", name="Test Task")
|
|
assert task.id == "task_1"
|
|
assert task.name == "Test Task"
|
|
assert task.status == TaskStatus.PENDING
|
|
assert task.steps == []
|
|
|
|
def test_task_with_steps(self):
|
|
"""Should create task with steps"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
step2 = Step(id="step_2", name="Step 2")
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
assert len(task.steps) == 2
|
|
|
|
def test_task_to_dict(self):
|
|
"""Should convert task to dict"""
|
|
task = Task(id="task_1", name="Test", goal="Complete task")
|
|
result = task.to_dict()
|
|
assert result["id"] == "task_1"
|
|
assert result["goal"] == "Complete task"
|
|
assert result["status"] == "pending"
|
|
|
|
|
|
class TestTaskGraph:
|
|
"""Tests for TaskGraph class"""
|
|
|
|
def test_graph_creation(self):
|
|
"""Should create graph from task"""
|
|
task = Task(id="task_1", name="Test")
|
|
graph = TaskGraph(task)
|
|
assert graph.task is task
|
|
|
|
def test_topological_sort_no_dependencies(self):
|
|
"""Should sort steps without dependencies"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
step2 = Step(id="step_2", name="Step 2")
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
graph = TaskGraph(task)
|
|
sorted_steps = graph.topological_sort()
|
|
assert len(sorted_steps) == 2
|
|
|
|
def test_topological_sort_with_dependencies(self):
|
|
"""Should respect dependencies in sort"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
graph = TaskGraph(task)
|
|
sorted_steps = graph.topological_sort()
|
|
ids = [s.id for s in sorted_steps]
|
|
assert ids.index("step_1") < ids.index("step_2")
|
|
|
|
def test_get_ready_steps(self):
|
|
"""Should return steps ready to execute"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
graph = TaskGraph(task)
|
|
ready = graph.get_ready_steps([])
|
|
assert len(ready) == 1
|
|
assert ready[0].id == "step_1"
|
|
|
|
def test_get_ready_steps_after_completion(self):
|
|
"""Should return dependent steps after completion"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
graph = TaskGraph(task)
|
|
ready = graph.get_ready_steps(["step_1"])
|
|
assert len(ready) == 1
|
|
assert ready[0].id == "step_2"
|
|
|
|
def test_detect_cycles_no_cycle(self):
|
|
"""Should return empty for no cycles"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
step2 = Step(id="step_2", name="Step 2")
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
graph = TaskGraph(task)
|
|
cycles = graph.detect_cycles()
|
|
assert cycles == []
|
|
|
|
def test_detect_cycles_with_cycle(self):
|
|
"""Should detect circular dependency"""
|
|
step1 = Step(id="step_1", name="Step 1", depends_on=["step_2"])
|
|
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
graph = TaskGraph(task)
|
|
cycles = graph.detect_cycles()
|
|
assert len(cycles) > 0
|
|
|
|
def test_validate_valid_graph(self):
|
|
"""Should validate valid graph"""
|
|
step1 = Step(id="step_1", name="Step 1")
|
|
task = Task(id="task_1", name="Test", steps=[step1])
|
|
graph = TaskGraph(task)
|
|
is_valid, error = graph.validate()
|
|
assert is_valid is True
|
|
assert error is None
|
|
|
|
def test_validate_invalid_dependency(self):
|
|
"""Should fail validation for invalid dependency"""
|
|
step1 = Step(id="step_1", name="Step 1", depends_on=["nonexistent"])
|
|
task = Task(id="task_1", name="Test", steps=[step1])
|
|
graph = TaskGraph(task)
|
|
is_valid, error = graph.validate()
|
|
assert is_valid is False
|
|
assert error is not None
|
|
|
|
|
|
class TestTaskService:
|
|
"""Tests for TaskService class"""
|
|
|
|
def test_create_task(self):
|
|
"""Should create a new task"""
|
|
service = TaskService()
|
|
task = service.create_task(name="Test Task", goal="Complete test")
|
|
assert task is not None
|
|
assert task.name == "Test Task"
|
|
assert task.goal == "Complete test"
|
|
|
|
def test_create_task_with_steps(self):
|
|
"""Should create task with steps"""
|
|
service = TaskService()
|
|
steps = [
|
|
{"name": "Step 1", "description": "First step"},
|
|
{"name": "Step 2", "description": "Second step"}
|
|
]
|
|
task = service.create_task(name="Test", goal="Goal", steps=steps)
|
|
assert len(task.steps) == 2
|
|
|
|
def test_get_task(self):
|
|
"""Should retrieve task by id"""
|
|
service = TaskService()
|
|
created = service.create_task(name="Test", goal="Goal")
|
|
retrieved = service.get_task(created.id)
|
|
assert retrieved is not None
|
|
assert retrieved.id == created.id
|
|
|
|
def test_get_nonexistent_task(self):
|
|
"""Should return None for nonexistent task"""
|
|
service = TaskService()
|
|
result = service.get_task("nonexistent_id")
|
|
assert result is None
|
|
|
|
def test_update_task_status(self):
|
|
"""Should update task status"""
|
|
service = TaskService()
|
|
task = service.create_task(name="Test", goal="Goal")
|
|
updated = service.update_task_status(task.id, TaskStatus.RUNNING)
|
|
assert updated is not None
|
|
assert updated.status == TaskStatus.RUNNING
|
|
|
|
def test_add_steps(self):
|
|
"""Should add steps to existing task"""
|
|
service = TaskService()
|
|
task = service.create_task(name="Test", goal="Goal")
|
|
steps = [{"name": "New Step"}]
|
|
added = service.add_steps(task.id, steps)
|
|
assert added is not None
|
|
assert len(added) == 1
|
|
assert len(task.steps) == 1
|
|
|
|
def test_delete_task(self):
|
|
"""Should delete task"""
|
|
service = TaskService()
|
|
task = service.create_task(name="Test", goal="Goal")
|
|
result = service.delete_task(task.id)
|
|
assert result is True
|
|
assert service.get_task(task.id) is None
|
|
|
|
def test_build_graph(self):
|
|
"""Should build graph for task"""
|
|
service = TaskService()
|
|
task = service.create_task(name="Test", goal="Goal")
|
|
graph = service.build_graph(task.id)
|
|
assert graph is not None
|
|
assert isinstance(graph, TaskGraph)
|