feat: complete LOW priority code quality improvements
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>
This commit is contained in:
beabigegg
700
backend/tests/test_action_executor.py
Normal file
700
backend/tests/test_action_executor.py
Normal file
@@ -0,0 +1,700 @@
|
||||
"""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
|
||||
@@ -273,3 +273,145 @@ class TestEncryptionKeyValidation:
|
||||
}):
|
||||
settings = Settings()
|
||||
assert settings.ENCRYPTION_MASTER_KEY is None
|
||||
|
||||
|
||||
class TestConfidentialProjectUpload:
|
||||
"""Tests for file upload on confidential projects."""
|
||||
|
||||
@pytest.fixture
|
||||
def test_user(self, db):
|
||||
"""Create a test user."""
|
||||
from app.models import User
|
||||
import uuid as uuid_module
|
||||
|
||||
user = User(
|
||||
id=str(uuid_module.uuid4()),
|
||||
email="testuser_enc@example.com",
|
||||
name="Test User Encryption",
|
||||
role_id="00000000-0000-0000-0000-000000000003",
|
||||
is_active=True,
|
||||
is_system_admin=False,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
return user
|
||||
|
||||
@pytest.fixture
|
||||
def test_user_token(self, client, mock_redis, test_user):
|
||||
"""Get a token for test user."""
|
||||
from app.core.security import create_access_token, create_token_payload
|
||||
|
||||
token_data = create_token_payload(
|
||||
user_id=test_user.id,
|
||||
email=test_user.email,
|
||||
role="engineer",
|
||||
department_id=None,
|
||||
is_system_admin=False,
|
||||
)
|
||||
token = create_access_token(token_data)
|
||||
mock_redis.setex(f"session:{test_user.id}", 900, token)
|
||||
return token
|
||||
|
||||
@pytest.fixture
|
||||
def test_space(self, db, test_user):
|
||||
"""Create a test space."""
|
||||
from app.models import Space
|
||||
import uuid as uuid_module
|
||||
|
||||
space = Space(
|
||||
id=str(uuid_module.uuid4()),
|
||||
name="Test Space Encryption",
|
||||
description="Test space for encryption tests",
|
||||
owner_id=test_user.id,
|
||||
)
|
||||
db.add(space)
|
||||
db.commit()
|
||||
return space
|
||||
|
||||
@pytest.fixture
|
||||
def confidential_project(self, db, test_space, test_user):
|
||||
"""Create a confidential test project."""
|
||||
from app.models import Project
|
||||
import uuid as uuid_module
|
||||
|
||||
project = Project(
|
||||
id=str(uuid_module.uuid4()),
|
||||
space_id=test_space.id,
|
||||
title="Confidential Project",
|
||||
description="Test confidential project",
|
||||
owner_id=test_user.id,
|
||||
security_level="confidential",
|
||||
)
|
||||
db.add(project)
|
||||
db.commit()
|
||||
return project
|
||||
|
||||
@pytest.fixture
|
||||
def test_task(self, db, confidential_project, test_user):
|
||||
"""Create a test task in confidential project."""
|
||||
from app.models import Task
|
||||
import uuid as uuid_module
|
||||
|
||||
task = Task(
|
||||
id=str(uuid_module.uuid4()),
|
||||
project_id=confidential_project.id,
|
||||
title="Test Task Encryption",
|
||||
description="Test task for encryption tests",
|
||||
created_by=test_user.id,
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
return task
|
||||
|
||||
def test_upload_confidential_project_encryption_unavailable(
|
||||
self, client, test_user_token, test_task, db
|
||||
):
|
||||
"""Test that uploading to confidential project returns 400 when encryption is unavailable."""
|
||||
from io import BytesIO
|
||||
|
||||
# Mock encryption service to return False for is_encryption_available
|
||||
with patch('app.api.attachments.router.encryption_service') as mock_enc_service:
|
||||
mock_enc_service.is_encryption_available.return_value = False
|
||||
|
||||
content = b"Test file content"
|
||||
files = {"file": ("test.pdf", BytesIO(content), "application/pdf")}
|
||||
|
||||
response = client.post(
|
||||
f"/api/tasks/{test_task.id}/attachments",
|
||||
headers={"Authorization": f"Bearer {test_user_token}"},
|
||||
files=files,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "ENCRYPTION_MASTER_KEY" in response.json()["detail"]
|
||||
assert "environment variable" in response.json()["detail"]
|
||||
|
||||
def test_upload_confidential_project_no_active_key(
|
||||
self, client, test_user_token, test_task, db
|
||||
):
|
||||
"""Test that uploading to confidential project returns 400 when no active encryption key exists."""
|
||||
from io import BytesIO
|
||||
from app.models import EncryptionKey
|
||||
|
||||
# Make sure no active encryption keys exist
|
||||
db.query(EncryptionKey).filter(EncryptionKey.is_active == True).update(
|
||||
{"is_active": False}
|
||||
)
|
||||
db.commit()
|
||||
|
||||
# Mock encryption service to return True for is_encryption_available
|
||||
with patch('app.api.attachments.router.encryption_service') as mock_enc_service:
|
||||
mock_enc_service.is_encryption_available.return_value = True
|
||||
|
||||
content = b"Test file content"
|
||||
files = {"file": ("test.pdf", BytesIO(content), "application/pdf")}
|
||||
|
||||
response = client.post(
|
||||
f"/api/tasks/{test_task.id}/attachments",
|
||||
headers={"Authorization": f"Bearer {test_user_token}"},
|
||||
files=files,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "active encryption key" in response.json()["detail"]
|
||||
assert "create" in response.json()["detail"].lower()
|
||||
|
||||
@@ -375,3 +375,119 @@ class TestTriggerAPI:
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "Invalid operator" in response.json()["detail"]
|
||||
|
||||
|
||||
class TestTriggerActionRollback:
|
||||
"""Tests for trigger action rollback mechanism."""
|
||||
|
||||
def test_multi_action_rollback_on_failure(self, db, test_task, test_project, test_user, test_status):
|
||||
"""Test that when one action fails, all previous actions are rolled back.
|
||||
|
||||
Scenario:
|
||||
1. Create a trigger with 2 actions: update_field (priority) + update_field (invalid status_id)
|
||||
2. The first action should update priority to 'high'
|
||||
3. The second action should fail because of invalid status_id
|
||||
4. The first action's change should be rolled back
|
||||
"""
|
||||
# Record original priority
|
||||
original_priority = test_task.priority
|
||||
|
||||
# Create a trigger with multiple actions where the second one will fail
|
||||
trigger = Trigger(
|
||||
id=str(uuid.uuid4()),
|
||||
project_id=test_project.id,
|
||||
name="Multi-Action Trigger",
|
||||
description="Test rollback on failure",
|
||||
trigger_type="field_change",
|
||||
conditions={
|
||||
"field": "status_id",
|
||||
"operator": "changed_to",
|
||||
"value": test_status[1].id,
|
||||
},
|
||||
actions=[
|
||||
# First action: update priority (should succeed)
|
||||
{
|
||||
"type": "update_field",
|
||||
"field": "priority",
|
||||
"value": "high",
|
||||
},
|
||||
# Second action: update to invalid status (should fail)
|
||||
{
|
||||
"type": "update_field",
|
||||
"field": "status_id",
|
||||
"value": "non-existent-status-id",
|
||||
},
|
||||
],
|
||||
is_active=True,
|
||||
created_by=test_user.id,
|
||||
)
|
||||
db.add(trigger)
|
||||
db.commit()
|
||||
|
||||
old_values = {"status_id": test_status[0].id}
|
||||
new_values = {"status_id": test_status[1].id}
|
||||
|
||||
# Execute trigger - second action should fail
|
||||
logs = TriggerService.evaluate_triggers(db, test_task, old_values, new_values, test_user)
|
||||
db.commit()
|
||||
|
||||
# Verify trigger execution failed
|
||||
assert len(logs) == 1
|
||||
assert logs[0].status == "failed"
|
||||
assert logs[0].error_message is not None
|
||||
assert "not found" in logs[0].error_message.lower()
|
||||
|
||||
# Refresh task from database to get the actual state
|
||||
db.refresh(test_task)
|
||||
|
||||
# Verify that the first action's change was rolled back
|
||||
# Priority should remain unchanged (not 'high')
|
||||
assert test_task.priority == original_priority, (
|
||||
f"Expected priority to be rolled back to '{original_priority}', "
|
||||
f"but got '{test_task.priority}'"
|
||||
)
|
||||
|
||||
def test_all_actions_succeed_committed(self, db, test_task, test_project, test_user, test_status):
|
||||
"""Test that when all actions succeed, changes are committed."""
|
||||
# Create a trigger with multiple successful actions
|
||||
trigger = Trigger(
|
||||
id=str(uuid.uuid4()),
|
||||
project_id=test_project.id,
|
||||
name="Success Trigger",
|
||||
description="Test successful commit",
|
||||
trigger_type="field_change",
|
||||
conditions={
|
||||
"field": "status_id",
|
||||
"operator": "changed_to",
|
||||
"value": test_status[1].id,
|
||||
},
|
||||
actions=[
|
||||
# First action: update priority
|
||||
{
|
||||
"type": "update_field",
|
||||
"field": "priority",
|
||||
"value": "urgent",
|
||||
},
|
||||
],
|
||||
is_active=True,
|
||||
created_by=test_user.id,
|
||||
)
|
||||
db.add(trigger)
|
||||
db.commit()
|
||||
|
||||
old_values = {"status_id": test_status[0].id}
|
||||
new_values = {"status_id": test_status[1].id}
|
||||
|
||||
# Execute trigger
|
||||
logs = TriggerService.evaluate_triggers(db, test_task, old_values, new_values, test_user)
|
||||
db.commit()
|
||||
|
||||
# Verify trigger execution succeeded
|
||||
assert len(logs) == 1
|
||||
assert logs[0].status == "success"
|
||||
|
||||
# Refresh task from database
|
||||
db.refresh(test_task)
|
||||
|
||||
# Verify that the change was committed
|
||||
assert test_task.priority == "urgent"
|
||||
|
||||
Reference in New Issue
Block a user