import pytest import uuid from datetime import datetime from app.models import ( User, Space, Project, Task, TaskStatus, Trigger, TriggerLog, Notification, CustomField, ProjectMember, Department, Role ) from app.services.trigger_service import TriggerService from app.services.custom_value_service import CustomValueService from app.schemas.task import CustomValueInput @pytest.fixture def test_user(db): """Create a test user.""" user = User( id=str(uuid.uuid4()), email="testuser@example.com", name="Test User", 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(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_user_csrf_token(test_user): """Generate a CSRF token for the test user.""" from app.core.security import generate_csrf_token return generate_csrf_token(test_user.id) @pytest.fixture def test_user_auth_headers(test_user_token, test_user_csrf_token): """Get complete auth headers for test user.""" return { "Authorization": f"Bearer {test_user_token}", "X-CSRF-Token": test_user_csrf_token, } @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 triggers", owner_id=test_user.id, ) db.add(space) db.commit() return space @pytest.fixture def test_project(db, test_space, test_user): """Create a test project.""" project = Project( id=str(uuid.uuid4()), space_id=test_space.id, title="Test Project", description="Test project for triggers", owner_id=test_user.id, ) db.add(project) db.commit() return project @pytest.fixture def test_status(db, test_project): """Create test task statuses.""" status1 = TaskStatus( id=str(uuid.uuid4()), project_id=test_project.id, name="To Do", color="#808080", position=0, ) status2 = TaskStatus( id=str(uuid.uuid4()), project_id=test_project.id, name="In Progress", color="#0000FF", position=1, ) db.add(status1) db.add(status2) db.commit() return status1, status2 @pytest.fixture def test_task(db, test_project, test_user, test_status): """Create a test task.""" task = Task( id=str(uuid.uuid4()), project_id=test_project.id, title="Test Task", description="Test task for triggers", status_id=test_status[0].id, created_by=test_user.id, assignee_id=test_user.id, ) db.add(task) db.commit() return task @pytest.fixture def test_trigger(db, test_project, test_user, test_status): """Create a test trigger.""" trigger = Trigger( id=str(uuid.uuid4()), project_id=test_project.id, name="Status Change Trigger", description="Notify when status changes to In Progress", trigger_type="field_change", conditions={ "field": "status_id", "operator": "changed_to", "value": test_status[1].id, }, actions=[{ "type": "notify", "target": "assignee", "template": "Task {task_title} status changed", }], is_active=True, created_by=test_user.id, ) db.add(trigger) db.commit() return trigger class TestTriggerService: """Tests for TriggerService.""" def test_check_conditions_changed_to(self, db, test_status): """Test changed_to condition.""" conditions = { "field": "status_id", "operator": "changed_to", "value": test_status[1].id, } old_values = {"status_id": test_status[0].id} new_values = {"status_id": test_status[1].id} result = TriggerService._check_conditions(conditions, old_values, new_values) assert result is True def test_check_conditions_changed_to_no_match(self, db, test_status): """Test changed_to condition when value doesn't match.""" conditions = { "field": "status_id", "operator": "changed_to", "value": test_status[1].id, } old_values = {"status_id": test_status[1].id} new_values = {"status_id": test_status[0].id} result = TriggerService._check_conditions(conditions, old_values, new_values) assert result is False def test_check_conditions_equals(self, db, test_status): """Test equals condition.""" conditions = { "field": "status_id", "operator": "equals", "value": test_status[1].id, } old_values = {"status_id": test_status[0].id} new_values = {"status_id": test_status[1].id} result = TriggerService._check_conditions(conditions, old_values, new_values) assert result is True def test_check_conditions_not_equals(self, db, test_status): """Test not_equals condition.""" conditions = { "field": "status_id", "operator": "not_equals", "value": test_status[0].id, } old_values = {"status_id": test_status[0].id} new_values = {"status_id": test_status[1].id} result = TriggerService._check_conditions(conditions, old_values, new_values) assert result is True def test_check_conditions_composite_and(self, db, test_status): """Test composite AND conditions with one unchanged rule.""" conditions = { "logic": "and", "rules": [ {"field": "status_id", "operator": "changed_to", "value": test_status[1].id}, {"field": "priority", "operator": "equals", "value": "high"}, ], } old_values = {"status_id": test_status[0].id, "priority": "high"} new_values = {"status_id": test_status[1].id, "priority": "high"} result = TriggerService._check_conditions(conditions, old_values, new_values) assert result is True def test_check_conditions_due_date_in_range_inclusive(self, db): """Test due_date in range operator is inclusive.""" conditions = { "logic": "and", "rules": [ { "field": "due_date", "operator": "in", "value": {"start": "2024-01-01", "end": "2024-01-15"}, } ], } old_values = {"due_date": datetime(2024, 1, 10)} new_values = {"due_date": datetime(2024, 1, 15)} result = TriggerService._check_conditions(conditions, old_values, new_values) assert result is True def test_evaluate_triggers_creates_notification(self, db, test_task, test_trigger, test_user, test_status): """Test that evaluate_triggers creates notification when conditions match.""" # Create another user to receive notification other_user = User( id=str(uuid.uuid4()), email="other@example.com", name="Other User", role_id="00000000-0000-0000-0000-000000000003", is_active=True, ) db.add(other_user) test_task.assignee_id = other_user.id db.commit() old_values = {"status_id": test_status[0].id} new_values = {"status_id": test_status[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" # Check notification was created notifications = db.query(Notification).filter( Notification.user_id == other_user.id ).all() assert len(notifications) == 1 def test_evaluate_triggers_inactive_trigger_not_executed(self, db, test_task, test_trigger, test_user, test_status): """Test that inactive triggers are not executed.""" test_trigger.is_active = False db.commit() old_values = {"status_id": test_status[0].id} new_values = {"status_id": test_status[1].id} logs = TriggerService.evaluate_triggers(db, test_task, old_values, new_values, test_user) assert len(logs) == 0 def test_custom_field_formula_condition(self, db, test_task, test_project, test_user): """Test formula custom field conditions are evaluated.""" number_field = CustomField( id=str(uuid.uuid4()), project_id=test_project.id, name="Points", field_type="number", position=0, ) formula_field = CustomField( id=str(uuid.uuid4()), project_id=test_project.id, name="Double Points", field_type="formula", formula="{Points} * 2", position=1, ) db.add_all([number_field, formula_field]) db.commit() CustomValueService.save_custom_values( db, test_task, [CustomValueInput(field_id=number_field.id, value=3)], ) db.commit() old_custom_values = { cv.field_id: cv.value for cv in CustomValueService.get_custom_values_for_task(db, test_task) } CustomValueService.save_custom_values( db, test_task, [CustomValueInput(field_id=number_field.id, value=4)], ) db.commit() new_custom_values = { cv.field_id: cv.value for cv in CustomValueService.get_custom_values_for_task(db, test_task) } trigger = Trigger( id=str(uuid.uuid4()), project_id=test_project.id, name="Formula Trigger", description="Notify when formula changes to 8", trigger_type="field_change", conditions={ "field": "custom_fields", "field_id": formula_field.id, "operator": "changed_to", "value": "8", }, actions=[{"type": "notify", "target": f"user:{test_user.id}"}], is_active=True, created_by=test_user.id, ) db.add(trigger) db.commit() logs = TriggerService.evaluate_triggers( db, test_task, {"custom_fields": old_custom_values}, {"custom_fields": new_custom_values}, test_user, ) db.commit() assert len(logs) == 1 assert logs[0].status == "success" class TestTriggerNotifications: """Tests for trigger notification target resolution.""" def test_notify_project_members_excludes_triggerer(self, db, test_task, test_project, test_user, test_status): member_user = User( id=str(uuid.uuid4()), email="member@example.com", name="Member User", role_id="00000000-0000-0000-0000-000000000003", is_active=True, ) other_member = User( id=str(uuid.uuid4()), email="member2@example.com", name="Other Member", role_id="00000000-0000-0000-0000-000000000003", is_active=True, ) db.add_all([member_user, other_member]) db.commit() db.add_all([ ProjectMember( id=str(uuid.uuid4()), project_id=test_project.id, user_id=member_user.id, role="member", added_by=test_user.id, ), ProjectMember( id=str(uuid.uuid4()), project_id=test_project.id, user_id=other_member.id, role="member", added_by=test_user.id, ), ProjectMember( id=str(uuid.uuid4()), project_id=test_project.id, user_id=test_user.id, role="member", added_by=test_user.id, ), ]) db.commit() trigger = Trigger( id=str(uuid.uuid4()), project_id=test_project.id, name="Project Members Trigger", description="Notify all project members", trigger_type="field_change", conditions={ "field": "status_id", "operator": "changed_to", "value": test_status[1].id, }, actions=[{"type": "notify", "target": "project_members"}], is_active=True, created_by=test_user.id, ) db.add(trigger) db.commit() logs = TriggerService.evaluate_triggers( db, test_task, {"status_id": test_status[0].id}, {"status_id": test_status[1].id}, member_user, ) db.commit() assert len(logs) == 1 assert db.query(Notification).filter(Notification.user_id == member_user.id).count() == 0 assert db.query(Notification).filter(Notification.user_id == other_member.id).count() == 1 assert db.query(Notification).filter(Notification.user_id == test_user.id).count() == 1 def test_notify_department_and_role_targets(self, db, test_task, test_project, test_user, test_status): department = Department( id=str(uuid.uuid4()), name="QA Department", ) qa_role = Role( id=str(uuid.uuid4()), name="qa", permissions={}, is_system_role=False, ) db.add_all([department, qa_role]) db.commit() triggerer = User( id=str(uuid.uuid4()), email="qa_lead@example.com", name="QA Lead", role_id=qa_role.id, department_id=department.id, is_active=True, ) dept_user = User( id=str(uuid.uuid4()), email="dept_user@example.com", name="Dept User", role_id="00000000-0000-0000-0000-000000000003", department_id=department.id, is_active=True, ) role_user = User( id=str(uuid.uuid4()), email="role_user@example.com", name="Role User", role_id=qa_role.id, department_id=None, is_active=True, ) db.add_all([triggerer, dept_user, role_user]) db.commit() dept_trigger = Trigger( id=str(uuid.uuid4()), project_id=test_project.id, name="Department Trigger", description="Notify department", trigger_type="field_change", conditions={ "field": "status_id", "operator": "changed_to", "value": test_status[1].id, }, actions=[{"type": "notify", "target": f"department:{department.id}"}], is_active=True, created_by=test_user.id, ) role_trigger = Trigger( id=str(uuid.uuid4()), project_id=test_project.id, name="Role Trigger", description="Notify role", trigger_type="field_change", conditions={ "field": "status_id", "operator": "changed_to", "value": test_status[1].id, }, actions=[{"type": "notify", "target": f"role:{qa_role.name}"}], is_active=True, created_by=test_user.id, ) db.add_all([dept_trigger, role_trigger]) db.commit() TriggerService.evaluate_triggers( db, test_task, {"status_id": test_status[0].id}, {"status_id": test_status[1].id}, triggerer, ) db.commit() assert db.query(Notification).filter(Notification.user_id == triggerer.id).count() == 0 assert db.query(Notification).filter(Notification.user_id == dept_user.id).count() == 1 assert db.query(Notification).filter(Notification.user_id == role_user.id).count() == 1 class TestTriggerAPI: """Tests for Trigger API endpoints.""" def test_create_trigger(self, client, test_user_auth_headers, test_project, test_status): """Test creating a trigger.""" response = client.post( f"/api/projects/{test_project.id}/triggers", headers=test_user_auth_headers, json={ "name": "New Trigger", "description": "Test trigger", "trigger_type": "field_change", "conditions": { "field": "status_id", "operator": "changed_to", "value": test_status[1].id, }, "actions": [{ "type": "notify", "target": "assignee", }], }, ) assert response.status_code == 201 data = response.json() assert data["name"] == "New Trigger" assert data["is_active"] is True def test_list_triggers(self, client, test_user_token, test_project, test_trigger): """Test listing triggers.""" response = client.get( f"/api/projects/{test_project.id}/triggers", headers={"Authorization": f"Bearer {test_user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["total"] >= 1 assert len(data["triggers"]) >= 1 def test_get_trigger(self, client, test_user_token, test_trigger): """Test getting a specific trigger.""" response = client.get( f"/api/triggers/{test_trigger.id}", headers={"Authorization": f"Bearer {test_user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["id"] == test_trigger.id assert data["name"] == test_trigger.name def test_update_trigger(self, client, test_user_auth_headers, test_trigger): """Test updating a trigger.""" response = client.put( f"/api/triggers/{test_trigger.id}", headers=test_user_auth_headers, json={ "name": "Updated Trigger", "is_active": False, }, ) assert response.status_code == 200 data = response.json() assert data["name"] == "Updated Trigger" assert data["is_active"] is False def test_delete_trigger(self, client, test_user_auth_headers, test_trigger, test_user_token): """Test deleting a trigger.""" response = client.delete( f"/api/triggers/{test_trigger.id}", headers=test_user_auth_headers, ) assert response.status_code == 204 # Verify deletion response = client.get( f"/api/triggers/{test_trigger.id}", headers={"Authorization": f"Bearer {test_user_token}"}, ) assert response.status_code == 404 def test_get_trigger_logs(self, client, test_user_token, test_trigger, db): """Test getting trigger logs.""" # Create a log entry log = TriggerLog( id=str(uuid.uuid4()), trigger_id=test_trigger.id, status="success", details={"test": True}, ) db.add(log) db.commit() response = client.get( f"/api/triggers/{test_trigger.id}/logs", headers={"Authorization": f"Bearer {test_user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["total"] >= 1 def test_create_trigger_invalid_field(self, client, test_user_auth_headers, test_project): """Test creating a trigger with invalid field.""" response = client.post( f"/api/projects/{test_project.id}/triggers", headers=test_user_auth_headers, json={ "name": "Invalid Trigger", "trigger_type": "field_change", "conditions": { "field": "invalid_field", "operator": "equals", "value": "test", }, "actions": [{"type": "notify", "target": "assignee"}], }, ) assert response.status_code == 400 assert "Invalid condition field" in response.json()["detail"] def test_create_trigger_invalid_operator(self, client, test_user_auth_headers, test_project): """Test creating a trigger with invalid operator.""" response = client.post( f"/api/projects/{test_project.id}/triggers", headers=test_user_auth_headers, json={ "name": "Invalid Trigger", "trigger_type": "field_change", "conditions": { "field": "status_id", "operator": "invalid_op", "value": "test", }, "actions": [{"type": "notify", "target": "assignee"}], }, ) 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"