from datetime import timedelta from unittest.mock import patch from urllib.parse import quote from fsm.state_choices import AnnotationStateChoices, ProjectStateChoices, TaskStateChoices from fsm.state_manager import get_state_manager from fsm.state_models import AnnotationState, ProjectState, TaskState from fsm.tests.factories import AnnotationStateFactory, ProjectStateFactory, TaskStateFactory from projects.tests.factories import ProjectFactory from rest_framework.test import APITestCase from tasks.tests.factories import AnnotationFactory, TaskFactory class FSMEntityHistoryAPITests(APITestCase): @classmethod def setUpTestData(cls): cls.project = ProjectFactory() cls.user = cls.project.created_by ProjectState.objects.all().delete() # Clean everything just in case cls.task = TaskFactory(project=cls.project) TaskState.objects.all().delete() # Clean everything just in case cls.annotation = AnnotationFactory(task=cls.task, completed_by=cls.user) AnnotationState.objects.all().delete() # Clean everything just in case def test_invalid_entity_name(self): self.client.force_authenticate(user=self.user) response = self.client.get('/api/fsm/entities/invalid/1/history') assert response.status_code == 404 def test_project_not_found(self): self.client.force_authenticate(user=self.user) response = self.client.get('/api/fsm/entities/project/999999/history') assert response.status_code == 404 def test_empty_project_history(self): self.client.force_authenticate(user=self.user) response = self.client.get(f'/api/fsm/entities/project/{self.project.id}/history') assert response.status_code == 200 assert response.json()['results'] == [] def test_project_history(self): state_1 = ProjectStateFactory(project=self.project, state=ProjectStateChoices.CREATED) state_1.created_at = state_1.created_at - timedelta(seconds=10) state_1.save() state_2 = ProjectStateFactory( project=self.project, state=ProjectStateChoices.IN_PROGRESS, previous_state=ProjectStateChoices.CREATED, triggered_by=self.user, reason='Project started by user', ) state_3 = ProjectStateFactory( project=self.project, state=ProjectStateChoices.COMPLETED, previous_state=ProjectStateChoices.IN_PROGRESS, transition_name='complete_project', reason='All tasks completed', ) self.client.force_authenticate(user=self.user) response = self.client.get(f'/api/fsm/entities/project/{self.project.id}/history') assert response.status_code == 200 results = response.json()['results'] assert len(results) == 3 assert results[0]['id'] == str(state_3.id) assert results[1]['id'] == str(state_2.id) assert results[2]['id'] == str(state_1.id) # Test that reason is returned as a top-level field (not nested in context_data) assert 'reason' in results[0] assert results[0]['reason'] == 'All tasks completed' assert results[1]['reason'] == 'Project started by user' # Test ordering response = self.client.get(f'/api/fsm/entities/project/{self.project.id}/history?ordering=id') assert response.status_code == 200 assert len(response.json()['results']) == 3 assert response.json()['results'][0]['id'] == str(state_1.id) assert response.json()['results'][1]['id'] == str(state_2.id) assert response.json()['results'][2]['id'] == str(state_3.id) # Test state filtering response = self.client.get( f'/api/fsm/entities/project/{self.project.id}/history?state={ProjectStateChoices.COMPLETED}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_3.id) # Test previous_state filtering response = self.client.get( f'/api/fsm/entities/project/{self.project.id}/history?previous_state={ProjectStateChoices.IN_PROGRESS}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_3.id) # Test transition_name filtering response = self.client.get( f'/api/fsm/entities/project/{self.project.id}/history?transition_name=complete_project' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_3.id) # Test triggered_by filtering response = self.client.get(f'/api/fsm/entities/project/{self.project.id}/history?triggered_by={self.user.id}') assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) assert response.json()['results'][0]['triggered_by']['id'] == self.user.id # Test date filtering created_at_from = (state_2.created_at - timedelta(seconds=1)).isoformat() created_at_to = state_3.created_at.isoformat() response = self.client.get( f'/api/fsm/entities/project/{self.project.id}/history?created_at_from={quote(created_at_from)}&created_at_to={quote(created_at_to)}' ) assert response.status_code == 200 assert len(response.json()['results']) == 2 assert response.json()['results'][0]['id'] == str(state_3.id) assert response.json()['results'][1]['id'] == str(state_2.id) def test_task_not_found(self): self.client.force_authenticate(user=self.user) response = self.client.get('/api/fsm/entities/task/999999/history') assert response.status_code == 404 def test_empty_task_history(self): self.client.force_authenticate(user=self.user) response = self.client.get(f'/api/fsm/entities/task/{self.task.id}/history') assert response.status_code == 200 assert response.json()['results'] == [] def test_task_history(self): state_1 = TaskStateFactory(task=self.task, state=TaskStateChoices.CREATED) state_1.created_at = state_1.created_at - timedelta(seconds=10) state_1.save() state_2 = TaskStateFactory( task=self.task, state=TaskStateChoices.IN_PROGRESS, previous_state=TaskStateChoices.CREATED, triggered_by=self.user, ) state_3 = TaskStateFactory( task=self.task, state=TaskStateChoices.COMPLETED, previous_state=TaskStateChoices.IN_PROGRESS, transition_name='complete_task', ) self.client.force_authenticate(user=self.user) response = self.client.get(f'/api/fsm/entities/task/{self.task.id}/history') assert response.status_code == 200 assert len(response.json()['results']) == 3 assert response.json()['results'][0]['id'] == str(state_3.id) assert response.json()['results'][1]['id'] == str(state_2.id) assert response.json()['results'][2]['id'] == str(state_1.id) # Test ordering response = self.client.get(f'/api/fsm/entities/task/{self.task.id}/history?ordering=id') assert response.status_code == 200 assert len(response.json()['results']) == 3 assert response.json()['results'][0]['id'] == str(state_1.id) assert response.json()['results'][1]['id'] == str(state_2.id) assert response.json()['results'][2]['id'] == str(state_3.id) # Test state filtering response = self.client.get(f'/api/fsm/entities/task/{self.task.id}/history?state={TaskStateChoices.COMPLETED}') assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_3.id) # Test previous_state filtering response = self.client.get( f'/api/fsm/entities/task/{self.task.id}/history?previous_state={TaskStateChoices.IN_PROGRESS}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_3.id) # Test transition_name filtering response = self.client.get(f'/api/fsm/entities/task/{self.task.id}/history?transition_name=complete_task') assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_3.id) # Test triggered_by filtering response = self.client.get(f'/api/fsm/entities/task/{self.task.id}/history?triggered_by={self.user.id}') assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) assert response.json()['results'][0]['triggered_by']['id'] == self.user.id # Test date filtering created_at_from = (state_2.created_at - timedelta(seconds=1)).isoformat() created_at_to = state_3.created_at.isoformat() response = self.client.get( f'/api/fsm/entities/task/{self.task.id}/history?created_at_from={quote(created_at_from)}&created_at_to={quote(created_at_to)}' ) assert response.status_code == 200 assert len(response.json()['results']) == 2 assert response.json()['results'][0]['id'] == str(state_3.id) assert response.json()['results'][1]['id'] == str(state_2.id) def test_annotation_not_found(self): self.client.force_authenticate(user=self.user) response = self.client.get('/api/fsm/entities/annotation/999999/history') assert response.status_code == 404 def test_empty_annotation_history(self): self.client.force_authenticate(user=self.user) response = self.client.get(f'/api/fsm/entities/annotation/{self.annotation.id}/history') assert response.status_code == 200 assert response.json()['results'] == [] def test_annotation_history(self): state_1 = AnnotationStateFactory(annotation=self.annotation, state=AnnotationStateChoices.SUBMITTED) state_1.created_at = state_1.created_at - timedelta(seconds=10) state_1.save() state_2 = AnnotationStateFactory( annotation=self.annotation, state=AnnotationStateChoices.COMPLETED, previous_state=AnnotationStateChoices.SUBMITTED, triggered_by=self.user, transition_name='complete_annotation', ) self.client.force_authenticate(user=self.user) response = self.client.get(f'/api/fsm/entities/annotation/{self.annotation.id}/history') assert response.status_code == 200 assert len(response.json()['results']) == 2 assert response.json()['results'][0]['id'] == str(state_2.id) assert response.json()['results'][1]['id'] == str(state_1.id) # Test ordering response = self.client.get(f'/api/fsm/entities/annotation/{self.annotation.id}/history?ordering=id') assert response.status_code == 200 assert len(response.json()['results']) == 2 assert response.json()['results'][0]['id'] == str(state_1.id) assert response.json()['results'][1]['id'] == str(state_2.id) # Test state filtering response = self.client.get( f'/api/fsm/entities/annotation/{self.annotation.id}/history?state={AnnotationStateChoices.COMPLETED}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) # Test previous_state filtering response = self.client.get( f'/api/fsm/entities/annotation/{self.annotation.id}/history?previous_state={AnnotationStateChoices.SUBMITTED}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) # Test transition_name filtering response = self.client.get( f'/api/fsm/entities/annotation/{self.annotation.id}/history?transition_name=complete_annotation' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) # Test triggered_by filtering response = self.client.get( f'/api/fsm/entities/annotation/{self.annotation.id}/history?triggered_by={self.user.id}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) assert response.json()['results'][0]['triggered_by']['id'] == self.user.id # Test date filtering created_at_from = (state_2.created_at - timedelta(seconds=1)).isoformat() created_at_to = state_2.created_at.isoformat() response = self.client.get( f'/api/fsm/entities/annotation/{self.annotation.id}/history?created_at_from={quote(created_at_from)}&created_at_to={quote(created_at_to)}' ) assert response.status_code == 200 assert len(response.json()['results']) == 1 assert response.json()['results'][0]['id'] == str(state_2.id) class FSMEntityTransitionAPITests(APITestCase): @classmethod def setUpTestData(cls): cls.project = ProjectFactory() cls.user = cls.project.created_by cls.task = TaskFactory(project=cls.project) cls.annotation = AnnotationFactory(task=cls.task, completed_by=cls.user) # Clean any pre-existing FSM state to have a known baseline ProjectState.objects.all().delete() TaskState.objects.all().delete() AnnotationState.objects.all().delete() def setUp(self): self.client.force_authenticate(user=self.user) self.StateManager = get_state_manager() @patch('fsm.state_manager.flag_set', return_value=True) def test_success_task_manual_transition(self, _mock_flag): response = self.client.post( f'/api/fsm/entities/task/{self.task.id}/transition/', data={'transition_name': 'task_completed', 'transition_data': {'reason': 'test complete'}}, format='json', ) assert response.status_code == 200 data = response.json() assert data['success'] is True assert data['new_state'] == TaskStateChoices.COMPLETED assert data['state_record']['triggered_by']['id'] == self.user.id # Ensure a state record exists current_state = self.StateManager.get_current_state_value(self.task) assert current_state == TaskStateChoices.COMPLETED @patch('fsm.state_manager.flag_set', return_value=True) def test_success_project_manual_transition(self, _mock_flag): response = self.client.post( f'/api/fsm/entities/project/{self.project.id}/transition/', data={'transition_name': 'project_in_progress'}, format='json', ) assert response.status_code == 200 data = response.json() assert data['success'] is True assert data['new_state'] == ProjectStateChoices.IN_PROGRESS assert data['state_record']['triggered_by']['id'] == self.user.id @patch('fsm.state_manager.flag_set', return_value=True) def test_request_body_validation_missing_transition_name(self, _mock_flag): response = self.client.post( f'/api/fsm/entities/task/{self.task.id}/transition/', data={}, format='json', ) assert response.status_code == 400 body = response.json() assert body.get('detail') == 'Validation error' assert 'validation_errors' in body assert 'transition_name' in body['validation_errors'] @patch('fsm.state_manager.flag_set', return_value=True) def test_returns_detailed_error_messages_on_failed_transition(self, _mock_flag): # Use an unknown transition to trigger a detailed validation error response response = self.client.post( f'/api/fsm/entities/task/{self.task.id}/transition/', data={'transition_name': 'does_not_exist', 'transition_data': {}}, format='json', ) assert response.status_code == 400 body = response.json() assert 'detail' in body @patch('fsm.state_manager.flag_set', return_value=True) def test_cannot_trigger_auto_triggered_transitions_manually(self, _mock_flag): # 'annotation_submitted' is auto-triggered on create response = self.client.post( f'/api/fsm/entities/annotation/{self.annotation.id}/transition/', data={'transition_name': 'annotation_submitted'}, format='json', ) assert response.status_code == 400 body = response.json() assert body.get('detail') == 'Validation error' assert 'validation_errors' in body assert 'transition_name' in body['validation_errors'] @patch('fsm.state_manager.flag_set', return_value=True) def test_audit_trail_captures_triggered_by(self, _mock_flag): response = self.client.post( f'/api/fsm/entities/project/{self.project.id}/transition/', data={'transition_name': 'project_in_progress'}, format='json', ) assert response.status_code == 200 body = response.json() assert body['state_record']['triggered_by']['id'] == self.user.id @patch('fsm.state_manager.flag_set', return_value=True) def test_unknown_transition_returns_400(self, _mock_flag): response = self.client.post( f'/api/fsm/entities/task/{self.task.id}/transition/', data={'transition_name': 'does_not_exist', 'transition_data': {}}, format='json', ) assert response.status_code == 400 body = response.json() assert 'detail' in body class LsoFSMEntityTransitionAPITests(FSMEntityTransitionAPITests, APITestCase): """Tests for LSO only that should not be inherited in LSE""" @patch('fsm.state_manager.flag_set', return_value=False) def test_feature_flag_respected_no_state_record_created(self, _mock_flag): """LSE State manager infers missing states, LSO does not""" # Execute a manual transition with FSM disabled response = self.client.post( f'/api/fsm/entities/task/{self.task.id}/transition/', data={'transition_name': 'task_completed'}, format='json', ) # Endpoint should still respond; state should not be created assert response.status_code == 200 current_state = self.StateManager.get_current_state_value(self.task) assert current_state is None