124 lines
6.5 KiB
Python
124 lines
6.5 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from rest_framework.test import APIRequestFactory, force_authenticate
|
|
from rest_framework.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_403_FORBIDDEN
|
|
|
|
from apps.orgs.models import Organization, Role
|
|
from apps.mlstore.models import AgentModel, Agent
|
|
from apps.onboarding.models import OnboardingFlow, OnboardingPage, OnboardingSession
|
|
from apps.onboarding.viewsets import OnboardingFlowViewSet, OnboardingSessionViewSet
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class OnboardingAPITests(TestCase):
|
|
def setUp(self):
|
|
self.factory = APIRequestFactory()
|
|
self.user = User.objects.create_user(email_address='user@example.com', password='pass')
|
|
self.manager = User.objects.create_user(email_address='manager@example.com', password='pass', is_manager=True)
|
|
self.org = Organization.objects.create(name='Org', owner=self.manager)
|
|
self.role = Role.objects.create(name='Engineer', organization=self.org)
|
|
self.model = AgentModel.objects.create(name='test-model', version='v1', path='model.gguf')
|
|
self.agent = Agent.objects.create(model=self.model, organization=self.org)
|
|
|
|
def test_create_flow(self):
|
|
view = OnboardingFlowViewSet.as_view({'post': 'create'})
|
|
data = {
|
|
'role': str(self.role.uuid),
|
|
'agent': str(self.agent.uuid),
|
|
'title': 'Flow',
|
|
'description': 'Desc',
|
|
'status': 'draft',
|
|
}
|
|
request = self.factory.post('/', data)
|
|
force_authenticate(request, user=self.manager)
|
|
response = view(request)
|
|
self.assertIn(response.status_code, (HTTP_200_OK, HTTP_201_CREATED))
|
|
self.assertTrue(OnboardingFlow.objects.filter(title='Flow').exists())
|
|
|
|
def test_pages_action(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
OnboardingPage.objects.create(flow=flow, order=0, title='Page 1', body='Body')
|
|
view = OnboardingFlowViewSet.as_view({'get': 'pages'})
|
|
request = self.factory.get('/')
|
|
force_authenticate(request, user=self.manager)
|
|
response = view(request, uuid=str(flow.uuid))
|
|
self.assertEqual(response.status_code, HTTP_200_OK)
|
|
self.assertEqual(len(response.data.get('pages', [])), 1)
|
|
|
|
|
|
|
|
def test_create_session(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
view = OnboardingSessionViewSet.as_view({'post': 'create'})
|
|
request = self.factory.post('/', {'flow': str(flow.uuid)})
|
|
force_authenticate(request, user=self.user)
|
|
response = view(request)
|
|
self.assertIn(response.status_code, (HTTP_200_OK, HTTP_201_CREATED))
|
|
self.assertTrue(OnboardingSession.objects.filter(flow=flow, user=self.user).exists())
|
|
|
|
def test_submit_updates_session(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
page = OnboardingPage.objects.create(flow=flow, order=0, title='Page 1', body='Body')
|
|
session = OnboardingSession.objects.create(flow=flow, user=self.user)
|
|
|
|
view = OnboardingSessionViewSet.as_view({'post': 'submit'})
|
|
payload = {'page_uuid': str(page.uuid), 'responses': {'q1': 'a'}, 'mark_complete': True}
|
|
request = self.factory.post('/', payload, format='json')
|
|
force_authenticate(request, user=self.user)
|
|
response = view(request, uuid=str(session.uuid))
|
|
self.assertEqual(response.status_code, HTTP_200_OK)
|
|
session.refresh_from_db()
|
|
self.assertEqual(session.status, 'completed')
|
|
self.assertIn(str(page.uuid), session.responses)
|
|
|
|
def test_publish_flow_as_manager(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
self.assertEqual(flow.status, 'draft')
|
|
view = OnboardingFlowViewSet.as_view({'post': 'publish'})
|
|
request = self.factory.post('/')
|
|
force_authenticate(request, user=self.manager)
|
|
response = view(request, uuid=str(flow.uuid))
|
|
self.assertEqual(response.status_code, HTTP_200_OK)
|
|
flow.refresh_from_db()
|
|
self.assertEqual(flow.status, 'published')
|
|
|
|
def test_publish_flow_requires_manager(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
view = OnboardingFlowViewSet.as_view({'post': 'publish'})
|
|
request = self.factory.post('/')
|
|
force_authenticate(request, user=self.user)
|
|
response = view(request, uuid=str(flow.uuid))
|
|
self.assertEqual(response.status_code, HTTP_403_FORBIDDEN)
|
|
|
|
def test_get_or_create_session_creates_when_missing(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
view = OnboardingSessionViewSet.as_view({'post': 'get_or_create'})
|
|
request = self.factory.post('/', {'flow': str(flow.uuid)})
|
|
force_authenticate(request, user=self.user)
|
|
response = view(request)
|
|
self.assertEqual(response.status_code, HTTP_200_OK)
|
|
self.assertTrue(OnboardingSession.objects.filter(flow=flow, user=self.user).exists())
|
|
|
|
def test_get_or_create_session_reuses_active(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
existing = OnboardingSession.objects.create(flow=flow, user=self.user, current_page_order=1)
|
|
view = OnboardingSessionViewSet.as_view({'post': 'get_or_create'})
|
|
request = self.factory.post('/', {'flow': str(flow.uuid)})
|
|
force_authenticate(request, user=self.user)
|
|
response = view(request)
|
|
self.assertEqual(response.status_code, HTTP_200_OK)
|
|
self.assertEqual(response.data.get('uuid'), str(existing.uuid))
|
|
self.assertEqual(response.data.get('current_page_order'), 1)
|
|
|
|
def test_get_or_create_session_creates_after_completion(self):
|
|
flow = OnboardingFlow.objects.create(role=self.role, agent=self.agent, title='Flow', description='')
|
|
completed = OnboardingSession.objects.create(flow=flow, user=self.user, status='completed')
|
|
view = OnboardingSessionViewSet.as_view({'post': 'get_or_create'})
|
|
request = self.factory.post('/', {'flow': str(flow.uuid)})
|
|
force_authenticate(request, user=self.user)
|
|
response = view(request)
|
|
self.assertEqual(response.status_code, HTTP_200_OK)
|
|
self.assertNotEqual(response.data.get('uuid'), str(completed.uuid))
|
|
|
|
|