from unittest.mock import patch from django.contrib.auth import get_user_model from django.test import TestCase from rest_framework.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_204_NO_CONTENT, HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN from rest_framework.test import APIClient from apps.accounts.models import Organization, Role from apps.onboarding.models import AgentConfig, AgentInteractionLog, OnboardingFlow, OnboardingSession from apps.onboarding.viewsets import OnboardingSessionViewSet User = get_user_model() class OnboardingApiTests(TestCase): def setUp(self): self.client: APIClient = APIClient() self.manager = User.objects.create_user( email_address='manager-o@example.com', password='pass1234', first_name='Manager', last_name='O', date_of_birth='1990-01-01', is_manager=True, ) self.member = User.objects.create_user( email_address='member-o@example.com', password='pass1234', first_name='Member', last_name='O', date_of_birth='1993-03-03', ) self.org = Organization.objects.create(name='Onboarding API Org', owner=self.manager) self.org.members.add(self.manager, self.member) self.role = Role.objects.create(name='Coordinator', organization=self.org) self.agent_config = AgentConfig.objects.create( organization=self.org, name='Coordinator Knowledge', agent_type='knowledge', system_prompt='Assist', ) self.flow = OnboardingFlow.objects.create( title='Coordinator Flow', role=self.role, structure=[{'uuid': 'page-1'}], ) self.session = OnboardingSession.objects.create( user=self.member, role=self.role, state={'progress': 10}, active_configs={}, ) self.log = AgentInteractionLog.objects.create( session=self.session, agent_config=self.agent_config, sender_type='user', content='hello', tool_call_metadata={}, ) def test_agent_config_list_path(self): self.client.force_authenticate(self.member) response = self.client.get('/api/agent-config/') self.assertEqual(response.status_code, HTTP_200_OK) def test_agent_config_create_path(self): self.client.force_authenticate(self.manager) response = self.client.post('/api/agent-config/', { 'organization_uuid': str(self.org.uuid), 'name': 'Coordinator Monitor', 'agent_type': 'monitor', 'system_prompt': 'Monitor progress', 'llm_config': {'model': 'local'}, 'tool_permissions': ['read'], }, format='json') self.assertEqual(response.status_code, HTTP_201_CREATED) def test_agent_config_retrieve_path(self): self.client.force_authenticate(self.member) response = self.client.get(f'/api/agent-config/{self.agent_config.uuid}/') self.assertEqual(response.status_code, HTTP_200_OK) def test_agent_config_update_path(self): self.client.force_authenticate(self.manager) response = self.client.put( f'/api/agent-config/{self.agent_config.uuid}/', { 'name': 'Coordinator Knowledge Updated', 'agent_type': 'knowledge', 'system_prompt': 'Updated', 'llm_config': {'model': 'local'}, 'tool_permissions': ['read'], }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_agent_config_partial_update_path(self): self.client.force_authenticate(self.manager) response = self.client.patch( f'/api/agent-config/{self.agent_config.uuid}/', {'name': 'Coordinator Knowledge Patched'}, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_agent_config_destroy_path(self): self.client.force_authenticate(self.manager) deletable = AgentConfig.objects.create( organization=self.org, name='Delete Config', agent_type='monitor', ) response = self.client.delete(f'/api/agent-config/{deletable.uuid}/') self.assertEqual(response.status_code, HTTP_204_NO_CONTENT) def test_onboarding_flow_list_path(self): self.client.force_authenticate(self.member) response = self.client.get('/api/onboarding-flow/') self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_flow_create_path(self): self.client.force_authenticate(self.manager) response = self.client.post('/api/onboarding-flow/', { 'title': 'New Flow', 'role_uuid': str(self.role.uuid), 'structure': [], 'is_active': True, }, format='json') self.assertEqual(response.status_code, HTTP_201_CREATED) def test_onboarding_flow_retrieve_path(self): self.client.force_authenticate(self.member) response = self.client.get(f'/api/onboarding-flow/{self.flow.uuid}/') self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_flow_update_path(self): self.client.force_authenticate(self.manager) response = self.client.put( f'/api/onboarding-flow/{self.flow.uuid}/', { 'title': 'Coordinator Flow Updated', 'structure': [{'uuid': 'page-2'}], 'is_active': True, }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_flow_partial_update_path(self): self.client.force_authenticate(self.manager) response = self.client.patch( f'/api/onboarding-flow/{self.flow.uuid}/', {'title': 'Coordinator Flow Patched'}, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_flow_destroy_path(self): self.client.force_authenticate(self.manager) delete_flow = OnboardingFlow.objects.create(title='Delete Flow', role=self.role, structure=[]) response = self.client.delete(f'/api/onboarding-flow/{delete_flow.uuid}/') self.assertEqual(response.status_code, HTTP_204_NO_CONTENT) def test_onboarding_flow_start_session_path(self): self.role.members.add(self.member) self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-flow/{self.flow.uuid}/start-session/') self.assertIn(response.status_code, (HTTP_200_OK, HTTP_201_CREATED)) def test_onboarding_flow_start_session_requires_role_membership_for_regular_users(self): self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-flow/{self.flow.uuid}/start-session/') self.assertEqual(response.status_code, HTTP_403_FORBIDDEN) self.assertIn('Join this role before starting onboarding.', response.data.get('error', '')) def test_onboarding_flow_start_session_allows_manager_without_role_membership(self): self.client.force_authenticate(self.manager) response = self.client.post(f'/api/onboarding-flow/{self.flow.uuid}/start-session/') self.assertIn(response.status_code, (HTTP_200_OK, HTTP_201_CREATED)) def test_onboarding_flow_start_session_is_user_specific(self): self.role.members.add(self.member) self.client.force_authenticate(self.manager) response = self.client.post(f'/api/onboarding-flow/{self.flow.uuid}/start-session/') self.assertIn(response.status_code, (HTTP_200_OK, HTTP_201_CREATED)) self.assertNotEqual(response.data.get('uuid'), str(self.session.uuid)) self.assertEqual(response.data.get('user', {}).get('uuid'), str(self.manager.uuid)) def test_onboarding_flow_start_session_reuses_existing_user_session(self): self.role.members.add(self.member) self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-flow/{self.flow.uuid}/start-session/') self.assertEqual(response.status_code, HTTP_200_OK) self.assertEqual(response.data.get('uuid'), str(self.session.uuid)) self.assertEqual(response.data.get('user', {}).get('uuid'), str(self.member.uuid)) def test_onboarding_session_list_path(self): self.client.force_authenticate(self.member) response = self.client.get('/api/onboarding-session/') self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_session_create_path(self): self.client.force_authenticate(self.member) response = self.client.post('/api/onboarding-session/', { 'user': str(self.member.uuid), 'role': str(self.role.uuid), 'status': 'active', 'state': {'progress': 0}, 'active_configs': {}, }, format='json') self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) def test_onboarding_session_retrieve_path(self): self.client.force_authenticate(self.member) response = self.client.get(f'/api/onboarding-session/{self.session.uuid}/') self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_session_update_path(self): self.client.force_authenticate(self.member) response = self.client.put( f'/api/onboarding-session/{self.session.uuid}/', { 'user': str(self.member.uuid), 'role': str(self.role.uuid), 'status': 'paused', 'state': {'progress': 20}, 'active_configs': {'knowledge': 'enabled'}, }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_session_partial_update_path(self): self.client.force_authenticate(self.member) response = self.client.patch( f'/api/onboarding-session/{self.session.uuid}/', {'status': 'paused'}, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_session_destroy_path(self): self.client.force_authenticate(self.member) deletable = OnboardingSession.objects.create( user=self.member, role=self.role, state={}, active_configs={}, ) response = self.client.delete(f'/api/onboarding-session/{deletable.uuid}/') self.assertEqual(response.status_code, HTTP_204_NO_CONTENT) def test_onboarding_session_interact_path(self): self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/interact/', { 'message': 'my answer', 'page_uuid': 'page-1', 'responses': {'q1': 'yes'}, }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_session_interact_tracks_page_visit_without_response_log(self): self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'fields': [], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.session.state = {'flow_uuid': str(self.flow.uuid)} self.session.save(update_fields=['state', 'updated_at']) existing_log_count = AgentInteractionLog.objects.filter(session=self.session).count() self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/interact/', { 'page_uuid': 'page-1', }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) self.assertEqual(response.data['session_state'].get('last_page_uuid'), 'page-1') self.assertIn('page-1', response.data['session_state'].get('visited_pages', [])) self.assertEqual(AgentInteractionLog.objects.filter(session=self.session).count(), existing_log_count) def test_onboarding_session_interact_stores_page_responses_without_assessment(self): self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'fields': [ { 'uuid': 'field-1', 'key': 'q1', 'label': 'Question 1', 'field_type': 'select', 'required': True, 'options': ['A', 'B', 'C'], 'validation': {'correct_option': 'B', 'explanation': 'B is correct'}, } ], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/interact/', { 'page_uuid': 'page-1', 'responses': {'q1': 'B'}, }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) self.assertNotIn('assessment', response.data) self.assertIn('session_state', response.data) def test_onboarding_session_interact_does_not_mark_final_quiz_completed(self): self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'fields': [], }, { 'uuid': 'quiz-1', 'title': 'Final Assessment Quiz', 'meta': {'page_type': 'final_quiz', 'pass_mark': 80}, 'fields': [ { 'uuid': 'field-1', 'key': 'q1', 'label': 'Question 1', 'field_type': 'select', 'required': True, 'options': ['A', 'B', 'C'], 'validation': {'correct_option': 'B', 'explanation': 'B is correct'}, } ], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.session.state = {'flow_uuid': str(self.flow.uuid), 'completed_modules': ['page-1']} self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/interact/', { 'page_uuid': 'quiz-1', 'responses': {'q1': 'B'}, }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) completed = response.data['session_state'].get('completed_modules', []) self.assertNotIn('quiz-1', completed) def test_onboarding_session_ask_ka_path(self): self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'body': 'Base onboarding content', 'fields': [], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.session.state = {'flow_uuid': str(self.flow.uuid)} self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/ask-ka/', { 'page_uuid': 'page-1', 'message': 'Can you explain this section in simpler terms?', 'mode': 'separate', }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) self.assertEqual(response.data['status'], 'ok') self.assertTrue(bool(response.data['answer'])) self.assertIn('page_help', response.data['session_state']) self.assertIn('page-1', response.data['session_state']['page_help']) def test_onboarding_session_ask_ka_update_page_rewrites_body(self): original_body = "## Intro\n\nOriginal onboarding content" revised_body = "## Intro\n\nRevised onboarding content with integrated clarification" self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'body': original_body, 'fields': [], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.session.state = {'flow_uuid': str(self.flow.uuid)} self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) with patch.object(OnboardingSessionViewSet, '_run_ka_page_revision', return_value=revised_body): response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/ask-ka/', { 'page_uuid': 'page-1', 'message': 'Please make this clearer for beginners.', 'mode': 'update_page', }, format='json', ) self.assertEqual(response.status_code, HTTP_200_OK) self.assertTrue(response.data['updated_page']) self.flow.refresh_from_db() page = self.flow.structure[0] self.assertEqual(page.get('body'), revised_body) self.assertNotIn('### Clarification', str(page.get('body') or '')) def test_onboarding_session_complete_blocks_when_quiz_score_below_pass_mark(self): self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'fields': [], }, { 'uuid': 'quiz-1', 'title': 'Final Assessment Quiz', 'meta': {'page_type': 'final_quiz', 'pass_mark': 80}, 'fields': [ { 'uuid': 'field-1', 'key': 'q1', 'label': 'Question 1', 'field_type': 'select', 'required': True, 'options': ['A', 'B', 'C'], 'validation': {'correct_option': 'B', 'explanation': 'B is correct'}, } ], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.session.state = { 'flow_uuid': str(self.flow.uuid), 'responses': { 'quiz-1': {'q1': 'A'} }, } self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/complete/', format='json', ) self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) self.assertEqual(response.data['quiz_result']['score_percentage'], 0) self.assertEqual(response.data['quiz_result']['pass_mark'], 80) self.assertFalse(response.data['quiz_result']['passed']) def test_onboarding_session_complete_path(self): self.flow.structure = [ { 'uuid': 'page-1', 'title': 'Module 1', 'fields': [], }, { 'uuid': 'quiz-1', 'title': 'Final Assessment Quiz', 'meta': {'page_type': 'final_quiz', 'pass_mark': 80}, 'fields': [ { 'uuid': 'field-1', 'key': 'q1', 'label': 'Question 1', 'field_type': 'select', 'required': True, 'options': ['A', 'B', 'C'], 'validation': {'correct_option': 'B', 'explanation': 'B is correct'}, } ], } ] self.flow.save(update_fields=['structure', 'updated_at']) self.session.state = { 'flow_uuid': str(self.flow.uuid), 'responses': { 'quiz-1': {'q1': 'B'} }, } self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-session/{self.session.uuid}/complete/') self.assertEqual(response.status_code, HTTP_200_OK) self.assertEqual(response.data['quiz_result']['score_percentage'], 100) self.assertTrue(response.data['quiz_result']['passed']) def test_onboarding_session_history_path(self): self.client.force_authenticate(self.member) response = self.client.get(f'/api/onboarding-session/{self.session.uuid}/history/') self.assertEqual(response.status_code, HTTP_200_OK) def test_agent_interaction_log_list_path(self): self.client.force_authenticate(self.member) response = self.client.get('/api/agent-interaction-log/') self.assertEqual(response.status_code, HTTP_200_OK) def test_agent_interaction_log_retrieve_path(self): self.client.force_authenticate(self.member) response = self.client.get(f'/api/agent-interaction-log/{self.log.uuid}/') self.assertEqual(response.status_code, HTTP_200_OK) def test_onboarding_flow_create_rejects_invalid_is_active(self): self.client.force_authenticate(self.manager) response = self.client.post('/api/onboarding-flow/', { 'title': 'Bad Bool Flow', 'role_uuid': str(self.role.uuid), 'structure': [], 'is_active': 'not-a-bool', }, format='json') self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) self.assertIn('is_active', response.json()) def test_onboarding_flow_create_accepts_false_string(self): self.client.force_authenticate(self.manager) response = self.client.post('/api/onboarding-flow/', { 'title': 'Disabled Flow', 'role_uuid': str(self.role.uuid), 'structure': [], 'is_active': 'false', }, format='json') self.assertEqual(response.status_code, HTTP_201_CREATED) self.assertFalse(response.data.get('is_active')) def test_onboarding_flow_partial_update_rejects_invalid_structure(self): self.client.force_authenticate(self.manager) response = self.client.patch( f'/api/onboarding-flow/{self.flow.uuid}/', {'structure': 'not-a-list'}, format='json', ) self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) self.assertIn('structure', response.data) def test_onboarding_flow_destroy_removes_sessions_for_role(self): self.client.force_authenticate(self.manager) flow_to_delete = OnboardingFlow.objects.create( title='Delete Me and Sessions', role=self.role, structure=[], ) session = OnboardingSession.objects.create( user=self.member, role=self.role, state={'flow_uuid': str(flow_to_delete.uuid)}, active_configs={}, ) response = self.client.delete(f'/api/onboarding-flow/{flow_to_delete.uuid}/') self.assertEqual(response.status_code, HTTP_204_NO_CONTENT) self.assertFalse(OnboardingSession.objects.filter(uuid=session.uuid).exists()) def test_onboarding_flow_start_session_updates_flow_uuid_on_existing_session(self): self.role.members.add(self.member) replacement_flow = OnboardingFlow.objects.create( title='Replacement Flow', role=self.role, structure=[{'uuid': 'page-x'}], ) self.session.state = {'flow_uuid': str(self.flow.uuid)} self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-flow/{replacement_flow.uuid}/start-session/') self.assertEqual(response.status_code, HTTP_200_OK) self.session.refresh_from_db() self.assertEqual(self.session.state.get('flow_uuid'), str(replacement_flow.uuid)) def test_agent_config_create_requires_organization_uuid(self): self.client.force_authenticate(self.manager) response = self.client.post('/api/agent-config/', { 'name': 'No Org Config', 'agent_type': 'monitor', }, format='json') self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) self.assertIn('organization_uuid', response.data) def test_agent_config_create_forbidden_for_non_manager_member(self): self.client.force_authenticate(self.member) response = self.client.post('/api/agent-config/', { 'organization_uuid': str(self.org.uuid), 'name': 'Member Cannot Create', 'agent_type': 'monitor', }, format='json') self.assertEqual(response.status_code, HTTP_403_FORBIDDEN) def test_onboarding_session_ask_ka_requires_page_uuid_and_message(self): self.client.force_authenticate(self.member) response = self.client.post( f'/api/onboarding-session/{self.session.uuid}/ask-ka/', {'page_uuid': 'page-1'}, format='json', ) self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) def test_onboarding_session_complete_returns_error_when_no_flow_exists(self): OnboardingFlow.objects.filter(role=self.role).delete() self.session.state = {'flow_uuid': str(self.flow.uuid), 'responses': {}} self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-session/{self.session.uuid}/complete/') self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST) self.assertIn('error', response.data) def test_onboarding_session_progress_overview_returns_user_flow_matrix(self): self.role.members.add(self.member) extra_flow = OnboardingFlow.objects.create( title='Second Flow', role=self.role, structure=[], is_active=True, ) self.session.state = {'flow_uuid': str(self.flow.uuid), 'progress_percentage': 35} self.session.save(update_fields=['state', 'updated_at']) self.client.force_authenticate(self.manager) response = self.client.get('/api/onboarding-session/progress-overview/') self.assertEqual(response.status_code, HTTP_200_OK) rows = response.json() self.assertTrue(isinstance(rows, list)) self.assertTrue(any(row.get('flow', {}).get('uuid') == str(self.flow.uuid) for row in rows)) self.assertTrue(any(row.get('flow', {}).get('uuid') == str(extra_flow.uuid) for row in rows)) first_flow_row = next(row for row in rows if row.get('flow', {}).get('uuid') == str(self.flow.uuid)) self.assertEqual(first_flow_row.get('latest_status'), self.session.status) self.assertEqual(first_flow_row.get('progress'), 35) def test_onboarding_session_list_filters_by_user_and_flow_for_manager(self): self.role.members.add(self.member, self.manager) self.session.state = {'flow_uuid': str(self.flow.uuid), 'progress_percentage': 50} self.session.save(update_fields=['state', 'updated_at']) manager_session = OnboardingSession.objects.create( user=self.manager, role=self.role, state={'flow_uuid': str(self.flow.uuid), 'progress_percentage': 10}, active_configs={}, ) self.client.force_authenticate(self.manager) response = self.client.get( '/api/onboarding-session/', { 'role_uuid': str(self.role.uuid), 'user_uuid': str(self.member.uuid), 'flow_uuid': str(self.flow.uuid), }, ) self.assertEqual(response.status_code, HTTP_200_OK) uuids = {str(item.get('uuid')) for item in response.json()} self.assertIn(str(self.session.uuid), uuids) self.assertNotIn(str(manager_session.uuid), uuids)