fishy/Dynavera
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Added extra code path tests and edge case checks

Browse source
This commit is contained in:
Viswamedha Nalabotu 2026-03-08 12:55:28 +00:00
parent ddae68b433
commit 00d60b3f4f
5 changed files with 773 additions and 82 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

213
apps/accounts/tests/test_api.py
View file

@ -1,5 +1,5 @@
from django.test import TestCase
from rest_framework import status
from rest_framework.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_204_NO_CONTENT, HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND
from rest_framework.test import APIClient
from apps.accounts.models import Invite, Organization, Role, User
@ -40,34 +40,34 @@ class AccountsApiTests(TestCase):
def test_user_list_path(self):
response = self.client.get('/api/user/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_user_retrieve_path(self):
response = self.client.get(f'/api/user/{self.manager.uuid}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_user_login_path(self):
response = self.client.post('/api/user/login/', {
'email_address': 'manager@example.com',
'password': 'pass1234',
})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
self.assertTrue(response.json().get('success'))
def test_user_logout_path(self):
self.client.force_authenticate(self.manager)
response = self.client.post('/api/user/logout/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_user_me_path(self):
self.client.force_authenticate(self.member)
response = self.client.get('/api/user/me/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
self.assertEqual(response.json()['email_address'], 'member@example.com')
def test_user_session_path(self):
response = self.client.get('/api/user/session/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
self.assertIn('isAuthenticated', response.json())
def test_user_signup_path(self):
@ -80,7 +80,7 @@ class AccountsApiTests(TestCase):
'date_of_birth': '1995-05-05',
'manager': False,
}, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.status_code, HTTP_201_CREATED)
def test_user_change_password_path(self):
self.client.force_authenticate(self.member)
@ -89,12 +89,12 @@ class AccountsApiTests(TestCase):
'password': 'newpass123',
'confirm_password': 'newpass123',
}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_list_path(self):
self.client.force_authenticate(self.manager)
response = self.client.get('/api/organization/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_create_path(self):
self.client.force_authenticate(self.manager)
@ -102,12 +102,12 @@ class AccountsApiTests(TestCase):
'name': 'Team Beta',
'description': 'Second team',
}, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.status_code, HTTP_201_CREATED)
def test_organization_retrieve_path(self):
self.client.force_authenticate(self.member)
response = self.client.get(f'/api/organization/{self.organization.uuid}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_update_path(self):
self.client.force_authenticate(self.manager)
@ -116,7 +116,7 @@ class AccountsApiTests(TestCase):
{'name': 'Team Alpha Updated', 'description': 'Updated'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_partial_update_path(self):
self.client.force_authenticate(self.manager)
@ -125,81 +125,222 @@ class AccountsApiTests(TestCase):
{'description': 'Patched'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_delete_path(self):
self.client.force_authenticate(self.manager)
org = Organization.objects.create(name='Delete Me', owner=self.manager)
org.members.add(self.manager)
response = self.client.delete(f'/api/organization/{org.uuid}/')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(response.status_code, HTTP_204_NO_CONTENT)
def test_organization_invite_list_path(self):
self.client.force_authenticate(self.manager)
Invite.objects.create(organization=self.organization, created_by=self.manager)
response = self.client.get(f'/api/organization/{self.organization.uuid}/invite/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.get(f'/api/invite/?organization_uuid={self.organization.uuid}')
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_create_invite_path(self):
self.client.force_authenticate(self.manager)
response = self.client.post(f'/api/organization/{self.organization.uuid}/create-invite/?max_uses=2', {}, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.post(f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=2', {}, format='json')
self.assertEqual(response.status_code, HTTP_201_CREATED)
self.assertIn('uuid', response.json())
def test_organization_revoke_invite_path(self):
self.client.force_authenticate(self.manager)
invite = Invite.objects.create(organization=self.organization, created_by=self.manager)
response = self.client.delete(f'/api/organization/{self.organization.uuid}/revoke-invite/{invite.uuid}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.delete(f'/api/invite/{invite.uuid}/?organization_uuid={self.organization.uuid}')
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_join_path(self):
self.client.force_authenticate(self.other)
invite = Invite.objects.create(organization=self.organization, created_by=self.manager)
response = self.client.post(f'/api/organization/join/{invite.uuid}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.post(f'/api/invite/join/?invite_uuid={invite.uuid}')
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_leave_path(self):
self.client.force_authenticate(self.member)
self.role.members.add(self.member)
response = self.client.post(f'/api/organization/{self.organization.uuid}/leave/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
self.organization.refresh_from_db()
self.assertFalse(self.organization.members.filter(uuid=self.member.uuid).exists())
self.assertFalse(self.role.members.filter(uuid=self.member.uuid).exists())
def test_organization_leave_owner_blocked_when_other_members_exist(self):
self.client.force_authenticate(self.manager)
response = self.client.post(f'/api/organization/{self.organization.uuid}/leave/')
self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST)
self.assertIn('Owner cannot leave while other members/managers exist', response.json().get('error', ''))
self.assertTrue(Organization.objects.filter(uuid=self.organization.uuid).exists())
def test_organization_leave_owner_deletes_org_when_no_other_members(self):
solo_org = Organization.objects.create(
name='Solo Owner Org',
description='Only owner remains',
owner=self.manager,
)
solo_org.members.add(self.manager)
self.client.force_authenticate(self.manager)
response = self.client.post(f'/api/organization/{solo_org.uuid}/leave/')
self.assertEqual(response.status_code, HTTP_200_OK)
self.assertFalse(Organization.objects.filter(uuid=solo_org.uuid).exists())
def test_organization_members_path(self):
self.client.force_authenticate(self.manager)
response = self.client.get(f'/api/organization/{self.organization.uuid}/members/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_remove_member_path(self):
self.client.force_authenticate(self.manager)
response = self.client.post(f'/api/organization/{self.organization.uuid}/member/{self.member.uuid}/remove/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_roles_get_path(self):
self.client.force_authenticate(self.manager)
response = self.client.get(f'/api/organization/{self.organization.uuid}/role/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.get(f'/api/role/?organization_uuid={self.organization.uuid}')
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_roles_post_path(self):
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/organization/{self.organization.uuid}/role/',
f'/api/role/?organization_uuid={self.organization.uuid}',
{'name': 'Designer', 'description': 'Design role'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.status_code, HTTP_201_CREATED)
def test_organization_my_roles_path(self):
self.client.force_authenticate(self.member)
self.role.members.add(self.member)
response = self.client.get('/api/organization/role/mine/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.get('/api/role/mine/')
self.assertEqual(response.status_code, HTTP_200_OK)
def test_organization_delete_role_path(self):
self.client.force_authenticate(self.manager)
delete_role = Role.objects.create(name='DeleteRole', organization=self.organization)
response = self.client.delete(f'/api/organization/{self.organization.uuid}/role/{delete_role.uuid}/')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
response = self.client.delete(f'/api/role/{delete_role.uuid}/?organization_uuid={self.organization.uuid}')
self.assertEqual(response.status_code, HTTP_204_NO_CONTENT)
def test_organization_join_role_path(self):
self.client.force_authenticate(self.member)
response = self.client.post(f'/api/organization/{self.organization.uuid}/role/{self.role.uuid}/join/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
response = self.client.post(f'/api/role/{self.role.uuid}/join/?organization_uuid={self.organization.uuid}')
self.assertEqual(response.status_code, HTTP_200_OK)
def test_invite_create_rejects_non_integer_max_uses(self):
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=abc',
{},
format='json',
)
self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST)
self.assertIn('max_uses', response.json())
def test_invite_create_rejects_out_of_bounds_max_uses(self):
self.client.force_authenticate(self.manager)
low = self.client.post(
f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=0',
{},
format='json',
)
high = self.client.post(
f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=1001',
{},
format='json',
)
self.assertEqual(low.status_code, HTTP_400_BAD_REQUEST)
self.assertEqual(high.status_code, HTTP_400_BAD_REQUEST)
def test_invite_create_accepts_max_uses_boundaries(self):
self.client.force_authenticate(self.manager)
low = self.client.post(
f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=1',
{},
format='json',
)
high = self.client.post(
f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=1000',
{},
format='json',
)
self.assertEqual(low.status_code, HTTP_201_CREATED)
self.assertEqual(high.status_code, HTTP_201_CREATED)
def test_invite_join_fails_after_max_uses_reached(self):
other_2 = User.objects.create_user(
email_address='other2@example.com',
password='pass1234',
first_name='Other',
last_name='Two',
date_of_birth='1994-04-04',
)
invite = Invite.objects.create(
organization=self.organization,
created_by=self.manager,
max_uses=1,
)
self.client.force_authenticate(self.other)
first = self.client.post(f'/api/invite/join/?invite_uuid={invite.uuid}')
self.assertEqual(first.status_code, HTTP_200_OK)
self.client.force_authenticate(other_2)
second = self.client.post(f'/api/invite/join/?invite_uuid={invite.uuid}')
self.assertEqual(second.status_code, HTTP_400_BAD_REQUEST)
self.assertIn('Invalid or expired invitation', second.json().get('error', ''))
def test_non_manager_member_cannot_create_invite(self):
self.client.force_authenticate(self.member)
response = self.client.post(
f'/api/invite/?organization_uuid={self.organization.uuid}&max_uses=2',
{},
format='json',
)
self.assertEqual(response.status_code, HTTP_403_FORBIDDEN)
def test_owner_cannot_be_removed_as_member(self):
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/organization/{self.organization.uuid}/member/{self.manager.uuid}/remove/'
)
self.assertEqual(response.status_code, HTTP_403_FORBIDDEN)
def test_remove_member_not_found(self):
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/organization/{self.organization.uuid}/member/{self.other.uuid}/remove/'
)
self.assertEqual(response.status_code, HTTP_404_NOT_FOUND)
def test_role_name_can_repeat_across_organizations(self):
second_org = Organization.objects.create(
name='Team Beta Scoped Role',
description='Second org',
owner=self.manager,
)
second_org.members.add(self.manager)
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/role/?organization_uuid={second_org.uuid}',
{'name': 'Developer', 'description': 'Same role name in different org'},
format='json',
)
self.assertEqual(response.status_code, HTTP_201_CREATED)
def test_role_name_must_be_unique_within_organization(self):
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/role/?organization_uuid={self.organization.uuid}',
{'name': 'Developer', 'description': 'Duplicate role in same org'},
format='json',
)
self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST)
self.assertIn('name', response.json())

31
apps/accounts/tests/test_models.py
View file

@ -1,4 +1,6 @@
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.db import IntegrityError, transaction
from django.test import TestCase
from django.utils import timezone
@ -95,3 +97,32 @@ class AccountsModelTests(TestCase):
self.assertIsNotNone(role.updated_at)
self.assertEqual(str(role), 'Engineer (Org C)')
def test_owner_is_added_to_members_on_create(self):
org = Organization.objects.create(name='Org Owner Membership', owner=self.owner)
self.assertTrue(org.members.filter(id=self.owner.id).exists())
def test_owner_cannot_be_removed_from_members(self):
org = Organization.objects.create(name='Org Owner Locked', owner=self.owner)
org.members.add(self.owner)
with self.assertRaises(ValidationError):
org.members.remove(self.owner)
def test_owner_remains_member_after_clear(self):
org = Organization.objects.create(name='Org Clear Members', owner=self.owner)
org.members.add(self.owner, self.member)
org.members.clear()
self.assertTrue(org.members.filter(id=self.owner.id).exists())
self.assertFalse(org.members.filter(id=self.member.id).exists())
def test_role_name_unique_per_organization(self):
org = Organization.objects.create(name='Org Role Scoped', owner=self.owner)
other_org = Organization.objects.create(name='Org Role Scoped 2', owner=self.member)
Role.objects.create(name='Analyst', organization=org)
Role.objects.create(name='Analyst', organization=other_org)
with self.assertRaises(IntegrityError):
with transaction.atomic():
Role.objects.create(name='Analyst', organization=org)

73
apps/knowledge/tests/test_api.py
View file

@ -2,7 +2,7 @@ from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from django.db.models.signals import post_save
from django.test import TestCase
from rest_framework import status
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
@ -63,22 +63,22 @@ class KnowledgeApiTests(TestCase):
def test_training_file_list_path(self):
self.client.force_authenticate(self.member)
response = self.client.get('/api/training-file/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_training_file_create_path(self):
self.client.force_authenticate(self.member)
uploaded = SimpleUploadedFile('new.txt', b'new body', content_type='text/plain')
response = self.client.post('/api/training-file/', {
'role': str(self.role.uuid),
'role_uuid': str(self.role.uuid),
'description': 'new file',
'file': uploaded,
})
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST)
def test_training_file_retrieve_path(self):
self.client.force_authenticate(self.member)
response = self.client.get(f'/api/training-file/{self.training_file.uuid}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_training_file_update_path(self):
self.client.force_authenticate(self.owner)
@ -89,7 +89,7 @@ class KnowledgeApiTests(TestCase):
'file': SimpleUploadedFile('replace.txt', b'updated', content_type='text/plain'),
},
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST)
def test_training_file_partial_update_path(self):
self.client.force_authenticate(self.owner)
@ -98,19 +98,72 @@ class KnowledgeApiTests(TestCase):
{'description': 'patched desc'},
format='multipart',
)
self.assertIn(response.status_code, (status.HTTP_200_OK, status.HTTP_400_BAD_REQUEST))
self.assertIn(response.status_code, (HTTP_200_OK, HTTP_400_BAD_REQUEST))
def test_training_file_destroy_path(self):
self.client.force_authenticate(self.owner)
response = self.client.delete(f'/api/training-file/{self.training_file.uuid}/')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(response.status_code, HTTP_204_NO_CONTENT)
def test_role_rag_document_list_path(self):
self.client.force_authenticate(self.member)
response = self.client.get('/api/role-rag-document/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_role_rag_document_retrieve_path(self):
self.client.force_authenticate(self.member)
response = self.client.get(f'/api/role-rag-document/{self.rag_doc.uuid}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_training_file_list_for_non_member_returns_empty(self):
outsider = User.objects.create_user(
email_address='outsider-k@example.com',
password='pass1234',
first_name='Out',
last_name='Sider',
date_of_birth='1994-04-04',
)
self.client.force_authenticate(outsider)
response = self.client.get('/api/training-file/')
self.assertEqual(response.status_code, HTTP_200_OK)
self.assertEqual(len(response.json()), 0)
def test_training_file_create_requires_role_uuid(self):
self.client.force_authenticate(self.owner)
uploaded = SimpleUploadedFile('new.txt', b'new body', content_type='text/plain')
response = self.client.post('/api/training-file/', {
'file': uploaded,
'file_name': 'new.txt',
})
self.assertEqual(response.status_code, HTTP_400_BAD_REQUEST)
self.assertIn('role_uuid', response.json())
def test_training_file_create_by_owner_succeeds(self):
self.client.force_authenticate(self.owner)
uploaded = SimpleUploadedFile('owner-ok.txt', b'owner file', content_type='text/plain')
response = self.client.post('/api/training-file/', {
'role_uuid': str(self.role.uuid),
'file': uploaded,
'file_name': 'owner-ok.txt',
})
self.assertEqual(response.status_code, HTTP_201_CREATED)
def test_training_file_destroy_forbidden_for_regular_member(self):
self.client.force_authenticate(self.member)
response = self.client.delete(f'/api/training-file/{self.training_file.uuid}/')
self.assertEqual(response.status_code, HTTP_403_FORBIDDEN)
def test_training_file_destroy_allowed_for_org_manager_member(self):
manager_member = User.objects.create_user(
email_address='manager-member-k@example.com',
password='pass1234',
first_name='Manager',
last_name='Member',
date_of_birth='1995-05-05',
is_manager=True,
)
self.org.members.add(manager_member)
self.client.force_authenticate(manager_member)
response = self.client.delete(f'/api/training-file/{self.training_file.uuid}/')
self.assertEqual(response.status_code, HTTP_204_NO_CONTENT)

449
apps/onboarding/tests/test_api.py
View file

@ -1,10 +1,13 @@
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.test import TestCase
from rest_framework import status
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()
@ -59,32 +62,30 @@ class OnboardingApiTests(TestCase):
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)
self.assertEqual(response.status_code, 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),
'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, status.HTTP_500_INTERNAL_SERVER_ERROR)
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, status.HTTP_200_OK)
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}/',
{
'organization': str(self.org.uuid),
'name': 'Coordinator Knowledge Updated',
'agent_type': 'knowledge',
'system_prompt': 'Updated',
@ -93,7 +94,7 @@ class OnboardingApiTests(TestCase):
},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_agent_config_partial_update_path(self):
self.client.force_authenticate(self.manager)
@ -102,7 +103,7 @@ class OnboardingApiTests(TestCase):
{'name': 'Coordinator Knowledge Patched'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_agent_config_destroy_path(self):
self.client.force_authenticate(self.manager)
@ -112,28 +113,27 @@ class OnboardingApiTests(TestCase):
agent_type='monitor',
)
response = self.client.delete(f'/api/agent-config/{deletable.uuid}/')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
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, status.HTTP_200_OK)
self.assertEqual(response.status_code, 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),
'role_uuid': str(self.role.uuid),
'structure': [],
'is_active': True,
}, format='json')
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
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, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_onboarding_flow_update_path(self):
self.client.force_authenticate(self.manager)
@ -141,13 +141,12 @@ class OnboardingApiTests(TestCase):
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)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_onboarding_flow_partial_update_path(self):
self.client.force_authenticate(self.manager)
@ -156,27 +155,56 @@ class OnboardingApiTests(TestCase):
{'title': 'Coordinator Flow Patched'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
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, status.HTTP_204_NO_CONTENT)
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, (status.HTTP_200_OK, status.HTTP_201_CREATED))
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, status.HTTP_200_OK)
self.assertEqual(response.status_code, 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),
@ -184,12 +212,12 @@ class OnboardingApiTests(TestCase):
'state': {'progress': 0},
'active_configs': {},
}, format='json')
self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
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, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_onboarding_session_update_path(self):
self.client.force_authenticate(self.member)
@ -204,7 +232,7 @@ class OnboardingApiTests(TestCase):
},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_onboarding_session_partial_update_path(self):
self.client.force_authenticate(self.member)
@ -213,7 +241,7 @@ class OnboardingApiTests(TestCase):
{'status': 'paused'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.status_code, HTTP_200_OK)
def test_onboarding_session_destroy_path(self):
self.client.force_authenticate(self.member)
@ -224,7 +252,7 @@ class OnboardingApiTests(TestCase):
active_configs={},
)
response = self.client.delete(f'/api/onboarding-session/{deletable.uuid}/')
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(response.status_code, HTTP_204_NO_CONTENT)
def test_onboarding_session_interact_path(self):
self.client.force_authenticate(self.member)
@ -237,24 +265,373 @@ class OnboardingApiTests(TestCase):
},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
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, 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)
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, status.HTTP_200_OK)
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, status.HTTP_200_OK)
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)

89
apps/onboarding/tests/test_consumers.py Normal file
View file

@ -0,0 +1,89 @@
from asgiref.sync import async_to_sync
from django.contrib.auth import get_user_model
from django.test import TestCase
from apps.accounts.models import Organization, Role
from apps.onboarding.consumers import OnboardingConsumer
from apps.onboarding.models import AgentConfig
User = get_user_model()
class OnboardingConsumerConfigSelectionTests(TestCase):
def setUp(self):
self.user = User.objects.create_user(
email_address='consumer-test@example.com',
password='pass1234',
first_name='Consumer',
last_name='Tester',
date_of_birth='1992-02-02',
is_manager=True,
)
self.org = Organization.objects.create(name='Consumer Test Org', owner=self.user)
self.org.members.add(self.user)
self.quant_role = Role.objects.create(name='Quant Role Consumer', organization=self.org)
self.ux_role = Role.objects.create(name='UX Role Consumer', organization=self.org)
self.consumer = OnboardingConsumer()
def test_get_config_by_type_prefers_exact_role(self):
quant_cfg = AgentConfig.objects.create(
organization=self.org,
role=self.quant_role,
name='Quant Curriculum Override',
agent_type='curriculum',
system_prompt='Quant-specific prompt',
)
AgentConfig.objects.create(
organization=self.org,
role=self.ux_role,
name='UX Curriculum Override',
agent_type='curriculum',
system_prompt='UX-specific prompt',
)
selected = async_to_sync(self.consumer.get_config_by_type)(str(self.quant_role.uuid), 'curriculum')
self.assertIsNotNone(selected)
self.assertEqual(selected.uuid, quant_cfg.uuid)
self.assertEqual(selected.role_id, self.quant_role.id)
def test_get_config_by_type_falls_back_to_org_default(self):
AgentConfig.objects.filter(role=self.quant_role, agent_type='monitor').delete()
org_default = AgentConfig.objects.create(
organization=self.org,
role=None,
name='Org Monitor Default',
agent_type='monitor',
system_prompt='Organization-level monitor prompt',
)
selected = async_to_sync(self.consumer.get_config_by_type)(str(self.quant_role.uuid), 'monitor')
self.assertIsNotNone(selected)
self.assertEqual(selected.uuid, org_default.uuid)
self.assertIsNone(selected.role)
def test_extract_json_list_supports_wrapped_questions_payload(self):
payload = (
"Here is your quiz output:\n"
"```json\n"
'{"questions": [{"key": "q1", "label": "Question?", "field_type": "select", "options": ["A", "B"], "required": true, "validation": {"correct_option": "A", "explanation": "A"}}]}\n'
"```"
)
extracted = self.consumer._extract_json_list(payload)
self.assertIsInstance(extracted, list)
self.assertEqual(len(extracted), 1)
self.assertEqual(extracted[0]['key'], 'q1')
def test_build_fallback_quiz_fields_generates_eight_valid_questions(self):
fallback = self.consumer._build_fallback_quiz_fields(['Topic A', 'Topic B'])
self.assertEqual(len(fallback), 8)
self.assertTrue(all(item.get('field_type') == 'select' for item in fallback))
self.assertTrue(all(len(item.get('options', [])) >= 4 for item in fallback))
self.assertTrue(all(item.get('validation', {}).get('correct_option') in item.get('options', []) for item in fallback))