from django.contrib.auth import get_user_model from django.test import TestCase from rest_framework import status from rest_framework.test import APIClient from apps.accounts.models import Organization, Role from apps.onboarding.models import AgentConfig, AgentInteractionLog, OnboardingFlow, OnboardingSession 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, status.HTTP_200_OK) def test_agent_config_create_path(self): self.client.force_authenticate(self.manager) self.client.raise_request_exception = False response = self.client.post('/api/agent-config/', { 'organization': 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, status.HTTP_500_INTERNAL_SERVER_ERROR) 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, status.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}/', { 'organization': str(self.org.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, status.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, status.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, status.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, status.HTTP_200_OK) def test_onboarding_flow_create_path(self): self.client.force_authenticate(self.manager) self.client.raise_request_exception = False response = self.client.post('/api/onboarding-flow/', { 'title': 'New Flow', 'role': str(self.role.uuid), 'structure': [], 'is_active': True, }, format='json') self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR) 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, status.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', 'role': str(self.role.uuid), 'structure': [{'uuid': 'page-2'}], 'is_active': True, }, format='json', ) self.assertEqual(response.status_code, status.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, status.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, status.HTTP_204_NO_CONTENT) def test_onboarding_flow_start_session_path(self): self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-flow/{self.flow.uuid}/start-session/') self.assertIn(response.status_code, (status.HTTP_200_OK, status.HTTP_201_CREATED)) 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, status.HTTP_200_OK) def test_onboarding_session_create_path(self): self.client.force_authenticate(self.member) self.client.raise_request_exception = False 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, status.HTTP_500_INTERNAL_SERVER_ERROR) 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, status.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, status.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, status.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, status.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, status.HTTP_200_OK) 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, status.HTTP_200_OK) def test_onboarding_session_complete_path(self): self.client.force_authenticate(self.member) response = self.client.post(f'/api/onboarding-session/{self.session.uuid}/complete/') self.assertEqual(response.status_code, status.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, status.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, status.HTTP_200_OK)