4b5a9c1d0a
Backend: - LOW-002: Add Query validation with max page size limits (100) - LOW-003: Replace magic strings with TaskStatus.is_done flag - LOW-004: Add 'creation' trigger type validation - Add action_executor.py with UpdateFieldAction and AutoAssignAction Frontend: - LOW-005: Replace TypeScript 'any' with 'unknown' + type guards - LOW-006: Add ConfirmModal component with A11Y support - LOW-007: Add ToastContext for user feedback notifications - LOW-009: Add Skeleton components (17 loading states replaced) - LOW-010: Setup Vitest with 21 tests for ConfirmModal and Skeleton Components updated: - App.tsx, ProtectedRoute.tsx, Spaces.tsx, Projects.tsx, Tasks.tsx - ProjectSettings.tsx, AuditPage.tsx, WorkloadPage.tsx, ProjectHealthPage.tsx - Comments.tsx, AttachmentList.tsx, TriggerList.tsx, TaskDetailModal.tsx - NotificationBell.tsx, BlockerDialog.tsx, CalendarView.tsx, WorkloadUserDetail.tsx 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
701 lines
24 KiB
Python
701 lines
24 KiB
Python
"""Tests for action_executor.py - FEAT-014 and FEAT-015.
|
|
|
|
This module tests the action executor framework and the two new action types:
|
|
- update_field: Update task field values
|
|
- auto_assign: Automatic task assignment with multiple strategies
|
|
"""
|
|
import pytest
|
|
import uuid
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
|
|
from app.models import User, Space, Project, Task, TaskStatus, Trigger, TriggerLog, Department
|
|
from app.services.action_executor import (
|
|
ActionExecutor,
|
|
UpdateFieldAction,
|
|
AutoAssignAction,
|
|
ActionExecutionError,
|
|
ActionValidationError,
|
|
)
|
|
from app.services.trigger_service import TriggerService
|
|
|
|
|
|
@pytest.fixture
|
|
def test_department(db):
|
|
"""Create a test department."""
|
|
dept = Department(
|
|
id=str(uuid.uuid4()),
|
|
name="Engineering",
|
|
)
|
|
db.add(dept)
|
|
db.commit()
|
|
return dept
|
|
|
|
|
|
@pytest.fixture
|
|
def test_user(db, test_department):
|
|
"""Create a test user."""
|
|
user = User(
|
|
id=str(uuid.uuid4()),
|
|
email="testuser@example.com",
|
|
name="Test User",
|
|
department_id=test_department.id,
|
|
role_id="00000000-0000-0000-0000-000000000003",
|
|
is_active=True,
|
|
is_system_admin=False,
|
|
capacity=Decimal("40.00"),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def additional_users(db, test_department):
|
|
"""Create additional test users for auto-assign tests."""
|
|
users = []
|
|
for i in range(3):
|
|
user = User(
|
|
id=str(uuid.uuid4()),
|
|
email=f"user{i}@example.com",
|
|
name=f"User {i}",
|
|
department_id=test_department.id,
|
|
role_id="00000000-0000-0000-0000-000000000003",
|
|
is_active=True,
|
|
is_system_admin=False,
|
|
capacity=Decimal("40.00"),
|
|
)
|
|
db.add(user)
|
|
users.append(user)
|
|
db.commit()
|
|
return users
|
|
|
|
|
|
@pytest.fixture
|
|
def test_space(db, test_user):
|
|
"""Create a test space."""
|
|
space = Space(
|
|
id=str(uuid.uuid4()),
|
|
name="Test Space",
|
|
description="Test space for action executor",
|
|
owner_id=test_user.id,
|
|
)
|
|
db.add(space)
|
|
db.commit()
|
|
return space
|
|
|
|
|
|
@pytest.fixture
|
|
def test_project(db, test_space, test_user, test_department):
|
|
"""Create a test project."""
|
|
project = Project(
|
|
id=str(uuid.uuid4()),
|
|
space_id=test_space.id,
|
|
title="Test Project",
|
|
description="Test project for action executor",
|
|
owner_id=test_user.id,
|
|
department_id=test_department.id,
|
|
)
|
|
db.add(project)
|
|
db.commit()
|
|
return project
|
|
|
|
|
|
@pytest.fixture
|
|
def test_statuses(db, test_project):
|
|
"""Create test task statuses."""
|
|
status_todo = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="To Do",
|
|
color="#808080",
|
|
position=0,
|
|
is_done=False,
|
|
)
|
|
status_in_progress = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="In Progress",
|
|
color="#0000FF",
|
|
position=1,
|
|
is_done=False,
|
|
)
|
|
status_done = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="Done",
|
|
color="#00FF00",
|
|
position=2,
|
|
is_done=True,
|
|
)
|
|
db.add_all([status_todo, status_in_progress, status_done])
|
|
db.commit()
|
|
return status_todo, status_in_progress, status_done
|
|
|
|
|
|
@pytest.fixture
|
|
def test_task(db, test_project, test_user, test_statuses):
|
|
"""Create a test task."""
|
|
task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title="Test Task",
|
|
description="Test task for action executor",
|
|
status_id=test_statuses[0].id,
|
|
created_by=test_user.id,
|
|
assignee_id=test_user.id,
|
|
priority="medium",
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
|
|
class TestUpdateFieldAction:
|
|
"""Tests for FEAT-014: Update Field Action."""
|
|
|
|
def test_validate_config_valid_priority(self, db):
|
|
"""Test validation of valid priority update config."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "priority", "value": "high"}
|
|
# Should not raise
|
|
action.validate_config(config, db)
|
|
|
|
def test_validate_config_valid_status_id(self, db):
|
|
"""Test validation of valid status_id update config."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "status_id", "value": "some-uuid"}
|
|
# Should not raise (status_id validation is deferred to execution)
|
|
action.validate_config(config, db)
|
|
|
|
def test_validate_config_valid_due_date(self, db):
|
|
"""Test validation of valid due_date update config."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "due_date", "value": "2025-12-31T23:59:59"}
|
|
# Should not raise
|
|
action.validate_config(config, db)
|
|
|
|
def test_validate_config_missing_field(self, db):
|
|
"""Test validation fails when field is missing."""
|
|
action = UpdateFieldAction()
|
|
config = {"value": "high"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Missing required 'field'" in str(exc.value)
|
|
|
|
def test_validate_config_missing_value(self, db):
|
|
"""Test validation fails when value is missing."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "priority"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Missing required 'value'" in str(exc.value)
|
|
|
|
def test_validate_config_invalid_field(self, db):
|
|
"""Test validation fails for unsupported field."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "invalid_field", "value": "test"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Invalid field" in str(exc.value)
|
|
|
|
def test_validate_config_invalid_priority(self, db):
|
|
"""Test validation fails for invalid priority value."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "priority", "value": "critical"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Invalid priority value" in str(exc.value)
|
|
|
|
def test_validate_config_invalid_due_date_format(self, db):
|
|
"""Test validation fails for invalid due_date format."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "due_date", "value": "not-a-date"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Invalid due_date format" in str(exc.value)
|
|
|
|
def test_execute_update_priority(self, db, test_task, test_user):
|
|
"""Test executing priority update."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "priority", "value": "urgent"}
|
|
context = {"current_user": test_user}
|
|
|
|
assert test_task.priority == "medium"
|
|
|
|
result = action.execute(db, test_task, config, context)
|
|
db.commit()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["field"] == "priority"
|
|
assert result["old_value"] == "medium"
|
|
assert result["new_value"] == "urgent"
|
|
assert test_task.priority == "urgent"
|
|
|
|
def test_execute_update_status_id(self, db, test_task, test_statuses, test_user):
|
|
"""Test executing status_id update."""
|
|
action = UpdateFieldAction()
|
|
new_status = test_statuses[1] # In Progress
|
|
config = {"field": "status_id", "value": new_status.id}
|
|
context = {"current_user": test_user}
|
|
|
|
result = action.execute(db, test_task, config, context)
|
|
db.commit()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["field"] == "status_id"
|
|
assert test_task.status_id == new_status.id
|
|
|
|
def test_execute_update_status_id_invalid(self, db, test_task, test_user):
|
|
"""Test executing status_id update with invalid status."""
|
|
action = UpdateFieldAction()
|
|
config = {"field": "status_id", "value": "non-existent-status"}
|
|
context = {"current_user": test_user}
|
|
|
|
with pytest.raises(ActionExecutionError) as exc:
|
|
action.execute(db, test_task, config, context)
|
|
assert "not found in project" in str(exc.value)
|
|
|
|
def test_execute_update_due_date(self, db, test_task, test_user):
|
|
"""Test executing due_date update."""
|
|
action = UpdateFieldAction()
|
|
due_date_str = "2025-12-31T23:59:59"
|
|
config = {"field": "due_date", "value": due_date_str}
|
|
context = {"current_user": test_user}
|
|
|
|
result = action.execute(db, test_task, config, context)
|
|
db.commit()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["field"] == "due_date"
|
|
assert test_task.due_date is not None
|
|
|
|
|
|
class TestAutoAssignAction:
|
|
"""Tests for FEAT-015: Auto Assign Action."""
|
|
|
|
def test_validate_config_valid_round_robin(self, db):
|
|
"""Test validation of valid round_robin config."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "round_robin"}
|
|
# Should not raise
|
|
action.validate_config(config, db)
|
|
|
|
def test_validate_config_valid_least_loaded(self, db):
|
|
"""Test validation of valid least_loaded config."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "least_loaded"}
|
|
# Should not raise
|
|
action.validate_config(config, db)
|
|
|
|
def test_validate_config_valid_specific_user(self, db, test_user):
|
|
"""Test validation of valid specific_user config."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "specific_user", "user_id": test_user.id}
|
|
# Should not raise
|
|
action.validate_config(config, db)
|
|
|
|
def test_validate_config_missing_strategy(self, db):
|
|
"""Test validation fails when strategy is missing."""
|
|
action = AutoAssignAction()
|
|
config = {}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Missing required 'strategy'" in str(exc.value)
|
|
|
|
def test_validate_config_invalid_strategy(self, db):
|
|
"""Test validation fails for invalid strategy."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "invalid_strategy"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Invalid strategy" in str(exc.value)
|
|
|
|
def test_validate_config_specific_user_missing_user_id(self, db):
|
|
"""Test validation fails when user_id is missing for specific_user."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "specific_user"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "Missing required 'user_id'" in str(exc.value)
|
|
|
|
def test_validate_config_specific_user_invalid_user_id(self, db):
|
|
"""Test validation fails for non-existent user."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "specific_user", "user_id": "non-existent-user"}
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
action.validate_config(config, db)
|
|
assert "not found or inactive" in str(exc.value)
|
|
|
|
def test_execute_specific_user(self, db, test_task, test_user, additional_users):
|
|
"""Test executing specific_user assignment."""
|
|
action = AutoAssignAction()
|
|
target_user = additional_users[0]
|
|
config = {"strategy": "specific_user", "user_id": target_user.id}
|
|
context = {"current_user": test_user}
|
|
|
|
result = action.execute(db, test_task, config, context)
|
|
db.commit()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["strategy"] == "specific_user"
|
|
assert result["new_assignee_id"] == target_user.id
|
|
assert test_task.assignee_id == target_user.id
|
|
|
|
def test_execute_round_robin(self, db, test_task, test_user, additional_users, test_project):
|
|
"""Test executing round_robin assignment."""
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "round_robin"}
|
|
context = {"current_user": test_user}
|
|
|
|
# Clear round robin state for this project
|
|
AutoAssignAction._round_robin_index.pop(test_project.id, None)
|
|
|
|
# First assignment
|
|
result1 = action.execute(db, test_task, config, context)
|
|
db.commit()
|
|
first_assignee = result1["new_assignee_id"]
|
|
|
|
assert result1["status"] == "success"
|
|
assert result1["strategy"] == "round_robin"
|
|
assert first_assignee is not None
|
|
|
|
# Create a new task for second assignment
|
|
task2 = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title="Test Task 2",
|
|
status_id=test_task.status_id,
|
|
created_by=test_user.id,
|
|
priority="medium",
|
|
)
|
|
db.add(task2)
|
|
db.commit()
|
|
db.refresh(task2)
|
|
|
|
# Second assignment should get next member in rotation
|
|
result2 = action.execute(db, task2, config, context)
|
|
db.commit()
|
|
second_assignee = result2["new_assignee_id"]
|
|
|
|
assert result2["status"] == "success"
|
|
# Round robin should cycle through members
|
|
assert second_assignee is not None
|
|
|
|
def test_execute_least_loaded(self, db, test_task, test_user, additional_users, test_project, test_statuses):
|
|
"""Test executing least_loaded assignment."""
|
|
# Create tasks assigned to different users to create varied workloads
|
|
for i, user in enumerate(additional_users):
|
|
for j in range(i + 1): # User 0 gets 1 task, User 1 gets 2, etc.
|
|
task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title=f"Task for user {i} - {j}",
|
|
status_id=test_statuses[0].id, # To Do (not done)
|
|
created_by=test_user.id,
|
|
assignee_id=user.id,
|
|
priority="medium",
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
|
|
# Create a new unassigned task
|
|
new_task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
title="New Task for Assignment",
|
|
status_id=test_statuses[0].id,
|
|
created_by=test_user.id,
|
|
priority="medium",
|
|
)
|
|
db.add(new_task)
|
|
db.commit()
|
|
db.refresh(new_task)
|
|
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "least_loaded"}
|
|
context = {"current_user": test_user}
|
|
|
|
result = action.execute(db, new_task, config, context)
|
|
db.commit()
|
|
|
|
assert result["status"] == "success"
|
|
assert result["strategy"] == "least_loaded"
|
|
# Should assign to user with fewest tasks
|
|
assert result["new_assignee_id"] is not None
|
|
|
|
def test_execute_no_eligible_members(self, db, test_user, test_space):
|
|
"""Test assignment when no eligible members are found."""
|
|
# Create project without department (only owner as member)
|
|
project = Project(
|
|
id=str(uuid.uuid4()),
|
|
space_id=test_space.id,
|
|
title="Solo Project",
|
|
owner_id=test_user.id,
|
|
department_id=None, # No department
|
|
)
|
|
db.add(project)
|
|
|
|
status = TaskStatus(
|
|
id=str(uuid.uuid4()),
|
|
project_id=project.id,
|
|
name="To Do",
|
|
color="#808080",
|
|
position=0,
|
|
)
|
|
db.add(status)
|
|
db.commit()
|
|
|
|
task = Task(
|
|
id=str(uuid.uuid4()),
|
|
project_id=project.id,
|
|
title="Solo Task",
|
|
status_id=status.id,
|
|
created_by=test_user.id,
|
|
priority="medium",
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
|
|
action = AutoAssignAction()
|
|
config = {"strategy": "round_robin"}
|
|
context = {"current_user": test_user}
|
|
|
|
result = action.execute(db, task, config, context)
|
|
|
|
# Should still work - assigns to owner
|
|
assert result["status"] == "success"
|
|
assert task.assignee_id == test_user.id
|
|
|
|
|
|
class TestActionExecutor:
|
|
"""Tests for ActionExecutor registry and execution."""
|
|
|
|
def test_get_supported_actions(self):
|
|
"""Test getting supported action types."""
|
|
actions = ActionExecutor.get_supported_actions()
|
|
assert "update_field" in actions
|
|
assert "auto_assign" in actions
|
|
|
|
def test_validate_action_update_field(self, db):
|
|
"""Test validating update_field action via executor."""
|
|
action = {
|
|
"type": "update_field",
|
|
"config": {"field": "priority", "value": "high"},
|
|
}
|
|
# Should not raise
|
|
ActionExecutor.validate_action(action, db)
|
|
|
|
def test_validate_action_auto_assign(self, db, test_user):
|
|
"""Test validating auto_assign action via executor."""
|
|
action = {
|
|
"type": "auto_assign",
|
|
"config": {"strategy": "specific_user", "user_id": test_user.id},
|
|
}
|
|
# Should not raise
|
|
ActionExecutor.validate_action(action, db)
|
|
|
|
def test_validate_action_unknown_type(self, db):
|
|
"""Test that unknown action types pass through (for extensibility)."""
|
|
action = {"type": "unknown_action"}
|
|
# Should not raise - allows other handlers to process
|
|
ActionExecutor.validate_action(action, db)
|
|
|
|
def test_execute_action_update_field(self, db, test_task, test_user):
|
|
"""Test executing update_field action via executor."""
|
|
action = {
|
|
"type": "update_field",
|
|
"config": {"field": "priority", "value": "urgent"},
|
|
}
|
|
context = {"current_user": test_user}
|
|
|
|
result = ActionExecutor.execute_action(db, test_task, action, context)
|
|
db.commit()
|
|
|
|
assert result is not None
|
|
assert result["status"] == "success"
|
|
assert test_task.priority == "urgent"
|
|
|
|
def test_execute_action_auto_assign(self, db, test_task, test_user, additional_users):
|
|
"""Test executing auto_assign action via executor."""
|
|
target_user = additional_users[1]
|
|
action = {
|
|
"type": "auto_assign",
|
|
"config": {"strategy": "specific_user", "user_id": target_user.id},
|
|
}
|
|
context = {"current_user": test_user}
|
|
|
|
result = ActionExecutor.execute_action(db, test_task, action, context)
|
|
db.commit()
|
|
|
|
assert result is not None
|
|
assert result["status"] == "success"
|
|
assert test_task.assignee_id == target_user.id
|
|
|
|
def test_execute_action_unknown_type_returns_none(self, db, test_task, test_user):
|
|
"""Test that unknown action types return None."""
|
|
action = {"type": "unknown_action"}
|
|
context = {"current_user": test_user}
|
|
|
|
result = ActionExecutor.execute_action(db, test_task, action, context)
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestTriggerServiceIntegration:
|
|
"""Tests for TriggerService integration with new actions."""
|
|
|
|
def test_validate_actions_update_field(self, db):
|
|
"""Test TriggerService validates update_field actions."""
|
|
actions = [{
|
|
"type": "update_field",
|
|
"config": {"field": "priority", "value": "high"},
|
|
}]
|
|
# Should not raise
|
|
TriggerService.validate_actions(actions, db)
|
|
|
|
def test_validate_actions_auto_assign(self, db, test_user):
|
|
"""Test TriggerService validates auto_assign actions."""
|
|
actions = [{
|
|
"type": "auto_assign",
|
|
"config": {"strategy": "specific_user", "user_id": test_user.id},
|
|
}]
|
|
# Should not raise
|
|
TriggerService.validate_actions(actions, db)
|
|
|
|
def test_validate_actions_invalid_action_type(self, db):
|
|
"""Test TriggerService rejects invalid action types."""
|
|
actions = [{"type": "invalid_action"}]
|
|
with pytest.raises(ActionValidationError) as exc:
|
|
TriggerService.validate_actions(actions, db)
|
|
assert "Invalid action type" in str(exc.value)
|
|
|
|
def test_get_supported_action_types(self):
|
|
"""Test getting all supported action types."""
|
|
types = TriggerService.get_supported_action_types()
|
|
assert "notify" in types
|
|
assert "update_field" in types
|
|
assert "auto_assign" in types
|
|
|
|
def test_evaluate_triggers_with_update_field_action(
|
|
self, db, test_task, test_user, test_project, test_statuses
|
|
):
|
|
"""Test trigger evaluation with update_field action."""
|
|
# Create trigger that updates priority when status changes
|
|
trigger = Trigger(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="Status Change Priority Trigger",
|
|
trigger_type="field_change",
|
|
conditions={
|
|
"field": "status_id",
|
|
"operator": "changed_to",
|
|
"value": test_statuses[1].id, # In Progress
|
|
},
|
|
actions=[{
|
|
"type": "update_field",
|
|
"config": {"field": "priority", "value": "high"},
|
|
}],
|
|
is_active=True,
|
|
created_by=test_user.id,
|
|
)
|
|
db.add(trigger)
|
|
db.commit()
|
|
|
|
old_values = {"status_id": test_statuses[0].id}
|
|
new_values = {"status_id": test_statuses[1].id}
|
|
|
|
logs = TriggerService.evaluate_triggers(db, test_task, old_values, new_values, test_user)
|
|
db.commit()
|
|
|
|
assert len(logs) == 1
|
|
assert logs[0].status == "success"
|
|
assert test_task.priority == "high"
|
|
|
|
def test_evaluate_triggers_with_auto_assign_action(
|
|
self, db, test_task, test_user, test_project, test_statuses, additional_users
|
|
):
|
|
"""Test trigger evaluation with auto_assign action."""
|
|
target_user = additional_users[0]
|
|
|
|
# Create trigger that auto-assigns when priority changes to urgent
|
|
trigger = Trigger(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="Priority Auto Assign Trigger",
|
|
trigger_type="field_change",
|
|
conditions={
|
|
"field": "priority",
|
|
"operator": "changed_to",
|
|
"value": "urgent",
|
|
},
|
|
actions=[{
|
|
"type": "auto_assign",
|
|
"config": {"strategy": "specific_user", "user_id": target_user.id},
|
|
}],
|
|
is_active=True,
|
|
created_by=test_user.id,
|
|
)
|
|
db.add(trigger)
|
|
db.commit()
|
|
|
|
old_values = {"priority": "medium"}
|
|
new_values = {"priority": "urgent"}
|
|
|
|
logs = TriggerService.evaluate_triggers(db, test_task, old_values, new_values, test_user)
|
|
db.commit()
|
|
|
|
assert len(logs) == 1
|
|
assert logs[0].status == "success"
|
|
assert test_task.assignee_id == target_user.id
|
|
|
|
def test_evaluate_triggers_with_multiple_actions(
|
|
self, db, test_task, test_user, test_project, test_statuses, additional_users
|
|
):
|
|
"""Test trigger evaluation with multiple actions."""
|
|
target_user = additional_users[0]
|
|
|
|
# Create trigger with both update_field and auto_assign
|
|
trigger = Trigger(
|
|
id=str(uuid.uuid4()),
|
|
project_id=test_project.id,
|
|
name="Multi Action Trigger",
|
|
trigger_type="field_change",
|
|
conditions={
|
|
"field": "status_id",
|
|
"operator": "changed_to",
|
|
"value": test_statuses[1].id,
|
|
},
|
|
actions=[
|
|
{
|
|
"type": "update_field",
|
|
"config": {"field": "priority", "value": "urgent"},
|
|
},
|
|
{
|
|
"type": "auto_assign",
|
|
"config": {"strategy": "specific_user", "user_id": target_user.id},
|
|
},
|
|
],
|
|
is_active=True,
|
|
created_by=test_user.id,
|
|
)
|
|
db.add(trigger)
|
|
db.commit()
|
|
|
|
old_values = {"status_id": test_statuses[0].id}
|
|
new_values = {"status_id": test_statuses[1].id}
|
|
|
|
logs = TriggerService.evaluate_triggers(db, test_task, old_values, new_values, test_user)
|
|
db.commit()
|
|
|
|
assert len(logs) == 1
|
|
assert logs[0].status == "success"
|
|
assert test_task.priority == "urgent"
|
|
assert test_task.assignee_id == target_user.id
|
|
|
|
# Check that both actions are logged
|
|
details = logs[0].details
|
|
assert len(details["actions_executed"]) == 2
|