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

Added model and api tests

Browse source
This commit is contained in:
Viswamedha Nalabotu 2026-02-27 12:12:26 +00:00
parent 445227a7ef
commit 5ffdb8ba27
9 changed files with 887 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
apps/accounts/tests/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Accounts tests package."""

208
apps/accounts/tests/test_api.py Normal file
View file

@ -0,0 +1,208 @@
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 Invite, Organization, Role
User = get_user_model()
class AccountsApiTests(TestCase):
def setUp(self):
self.client: APIClient = APIClient()
self.manager = User.objects.create_user(
email_address='manager@example.com',
password='pass1234',
first_name='Manager',
last_name='User',
date_of_birth='1990-01-01',
is_manager=True,
)
self.member = User.objects.create_user(
email_address='member@example.com',
password='pass1234',
first_name='Member',
last_name='User',
date_of_birth='1992-02-02',
)
self.other = User.objects.create_user(
email_address='other@example.com',
password='pass1234',
first_name='Other',
last_name='User',
date_of_birth='1993-03-03',
)
self.organization = Organization.objects.create(
name='Team Alpha',
description='Main team',
owner=self.manager,
)
self.organization.members.add(self.manager, self.member)
self.role = Role.objects.create(name='Developer', organization=self.organization)
def test_user_list_path(self):
response = self.client.get('/api/user/')
self.assertEqual(response.status_code, status.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)
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.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)
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.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.assertIn('isAuthenticated', response.json())
def test_user_signup_path(self):
response = self.client.post('/api/user/signup/', {
'email_address': 'signup@example.com',
'password': 'newpass123',
'confirm_password': 'newpass123',
'first_name': 'Sign',
'last_name': 'Up',
'date_of_birth': '1995-05-05',
'manager': False,
}, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
def test_user_change_password_path(self):
self.client.force_authenticate(self.member)
response = self.client.post('/api/user/change_password/', {
'old_password': 'pass1234',
'password': 'newpass123',
'confirm_password': 'newpass123',
}, format='json')
self.assertEqual(response.status_code, status.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)
def test_organization_create_path(self):
self.client.force_authenticate(self.manager)
response = self.client.post('/api/organization/', {
'name': 'Team Beta',
'description': 'Second team',
}, format='json')
self.assertEqual(response.status_code, status.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)
def test_organization_update_path(self):
self.client.force_authenticate(self.manager)
response = self.client.put(
f'/api/organization/{self.organization.uuid}/',
{'name': 'Team Alpha Updated', 'description': 'Updated'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_organization_partial_update_path(self):
self.client.force_authenticate(self.manager)
response = self.client.patch(
f'/api/organization/{self.organization.uuid}/',
{'description': 'Patched'},
format='json',
)
self.assertEqual(response.status_code, status.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)
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)
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)
self.assertIn('token', 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.token}/')
self.assertEqual(response.status_code, status.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.token}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_organization_leave_path(self):
self.client.force_authenticate(self.member)
response = self.client.post(f'/api/organization/{self.organization.uuid}/leave/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
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)
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.id}/remove/')
self.assertEqual(response.status_code, status.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)
def test_organization_roles_post_path(self):
self.client.force_authenticate(self.manager)
response = self.client.post(
f'/api/organization/{self.organization.uuid}/role/',
{'name': 'Designer', 'description': 'Design role'},
format='json',
)
self.assertEqual(response.status_code, status.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)
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)
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)

97
apps/accounts/tests/test_models.py Normal file
View file

@ -0,0 +1,97 @@
from django.contrib.auth import get_user_model
from django.test import TestCase
from django.utils import timezone
from apps.accounts.models import Invite, Organization, Role
User = get_user_model()
class AccountsModelTests(TestCase):
def setUp(self):
self.owner = User.objects.create_user(
email_address='owner@example.com',
password='pass1234',
first_name='Owner',
last_name='User',
date_of_birth='1990-01-01',
is_manager=True,
)
self.member = User.objects.create_user(
email_address='member@example.com',
password='pass1234',
first_name='Member',
last_name='User',
date_of_birth='1992-02-02',
)
def test_user_fields_and_defaults(self):
self.assertEqual(self.owner.email_address, 'owner@example.com')
self.assertEqual(self.owner.first_name, 'Owner')
self.assertEqual(self.owner.last_name, 'User')
self.assertEqual(str(self.owner.date_of_birth), '1990-01-01')
self.assertTrue(self.owner.is_active)
self.assertFalse(self.owner.is_staff)
self.assertTrue(self.owner.is_manager)
self.assertIsNotNone(self.owner.id)
self.assertIsNotNone(self.owner.uuid)
self.assertIsNotNone(self.owner.created_at)
self.assertIsNotNone(self.owner.updated_at)
self.assertEqual(self.owner.full_name, 'Owner User')
self.assertEqual(str(self.owner), 'Owner User')
def test_organization_fields_and_relations(self):
org = Organization.objects.create(
name='Acme',
description='Primary org',
owner=self.owner,
)
org.members.add(self.owner, self.member)
self.assertEqual(org.name, 'Acme')
self.assertEqual(org.description, 'Primary org')
self.assertEqual(org.owner, self.owner)
self.assertEqual(org.members.count(), 2)
self.assertIsNotNone(org.id)
self.assertIsNotNone(org.uuid)
self.assertIsNotNone(org.created_at)
self.assertIsNotNone(org.updated_at)
self.assertEqual(str(org), 'Acme')
def test_invite_fields_defaults_and_validity(self):
org = Organization.objects.create(name='Org B', owner=self.owner)
invite = Invite.objects.create(organization=org, created_by=self.owner)
self.assertIsNotNone(invite.token)
self.assertEqual(invite.organization, org)
self.assertEqual(invite.created_by, self.owner)
self.assertEqual(invite.uses, 0)
self.assertEqual(invite.max_uses, 1)
self.assertTrue(invite.is_active)
self.assertGreater(invite.expires_at, timezone.now())
self.assertTrue(invite.is_valid())
invite.uses = 1
invite.save(update_fields=['uses'])
self.assertFalse(invite.is_valid())
def test_role_fields_relations_and_string(self):
org = Organization.objects.create(name='Org C', owner=self.owner)
role = Role.objects.create(name='Engineer', description='Builds things', organization=org)
role.members.add(self.member)
self.assertEqual(role.name, 'Engineer')
self.assertEqual(role.description, 'Builds things')
self.assertEqual(role.organization, org)
self.assertEqual(role.members.count(), 1)
self.assertIsNotNone(role.id)
self.assertIsNotNone(role.uuid)
self.assertIsNotNone(role.created_at)
self.assertIsNotNone(role.updated_at)
self.assertEqual(str(role), 'Engineer (Org C)')

1
apps/knowledge/tests/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Knowledge tests package."""

116
apps/knowledge/tests/test_api.py Normal file
View file

@ -0,0 +1,116 @@
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.test import APIClient
from apps.accounts.models import Organization, Role
from apps.knowledge.models import RoleRagDocument, TrainingFile, trigger_ingestion
User = get_user_model()
class KnowledgeApiTests(TestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
post_save.disconnect(trigger_ingestion, sender=TrainingFile)
@classmethod
def tearDownClass(cls):
post_save.connect(trigger_ingestion, sender=TrainingFile)
super().tearDownClass()
def setUp(self):
self.client: APIClient = APIClient()
self.owner = User.objects.create_user(
email_address='owner-k@example.com',
password='pass1234',
first_name='Owner',
last_name='K',
date_of_birth='1990-01-01',
is_manager=True,
)
self.member = User.objects.create_user(
email_address='member-k@example.com',
password='pass1234',
first_name='Member',
last_name='K',
date_of_birth='1992-02-02',
)
self.org = Organization.objects.create(name='Knowledge API Org', owner=self.owner)
self.org.members.add(self.owner, self.member)
self.role = Role.objects.create(name='Researcher', organization=self.org)
self.training_file = TrainingFile.objects.create(
role=self.role,
uploaded_by=self.owner,
file=SimpleUploadedFile('doc.txt', b'content', content_type='text/plain'),
file_name='doc.txt',
file_size=7,
file_type='text/plain',
)
self.rag_doc = RoleRagDocument.objects.create(
role=self.role,
training_file=self.training_file,
content='chunk body',
content_hash='b' * 64,
metadata={'k': 'v'},
chunk_index=0,
)
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)
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),
'description': 'new file',
'file': uploaded,
})
self.assertEqual(response.status_code, status.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)
def test_training_file_update_path(self):
self.client.force_authenticate(self.owner)
response = self.client.put(
f'/api/training-file/{self.training_file.uuid}/',
{
'description': 'updated desc',
'file': SimpleUploadedFile('replace.txt', b'updated', content_type='text/plain'),
},
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_training_file_partial_update_path(self):
self.client.force_authenticate(self.owner)
response = self.client.patch(
f'/api/training-file/{self.training_file.uuid}/',
{'description': 'patched desc'},
format='multipart',
)
self.assertIn(response.status_code, (status.HTTP_200_OK, status.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)
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)
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)

95
apps/knowledge/tests/test_models.py Normal file
View file

@ -0,0 +1,95 @@
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 apps.accounts.models import Organization, Role
from apps.knowledge.models import RoleRagDocument, TrainingFile, trigger_ingestion
User = get_user_model()
class KnowledgeModelTests(TestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
post_save.disconnect(trigger_ingestion, sender=TrainingFile)
@classmethod
def tearDownClass(cls):
post_save.connect(trigger_ingestion, sender=TrainingFile)
super().tearDownClass()
def setUp(self):
self.user = User.objects.create_user(
email_address='knowledge@example.com',
password='pass1234',
first_name='Knowledge',
last_name='User',
date_of_birth='1991-01-01',
)
self.org = Organization.objects.create(name='Knowledge Org', owner=self.user)
self.org.members.add(self.user)
self.role = Role.objects.create(name='Analyst', organization=self.org)
def test_training_file_fields_and_defaults(self):
uploaded = SimpleUploadedFile('training.txt', b'hello world', content_type='text/plain')
training_file = TrainingFile.objects.create(
role=self.role,
uploaded_by=self.user,
file=uploaded,
file_name='training.txt',
file_size=11,
file_type='text/plain',
description='sample',
)
self.assertEqual(training_file.role, self.role)
self.assertEqual(training_file.uploaded_by, self.user)
self.assertEqual(training_file.file_name, 'training.txt')
self.assertEqual(training_file.file_size, 11)
self.assertEqual(training_file.file_type, 'text/plain')
self.assertEqual(training_file.description, 'sample')
self.assertEqual(training_file.status, 'ingesting')
self.assertFalse(training_file.is_processed)
self.assertIsNotNone(training_file.id)
self.assertIsNotNone(training_file.uuid)
self.assertIsNotNone(training_file.created_at)
self.assertIsNotNone(training_file.updated_at)
self.assertIn('training.txt (Analyst)', str(training_file))
def test_role_rag_document_fields_and_defaults(self):
uploaded = SimpleUploadedFile('base.txt', b'base', content_type='text/plain')
training_file = TrainingFile.objects.create(
role=self.role,
uploaded_by=self.user,
file=uploaded,
file_name='base.txt',
file_size=4,
file_type='text/plain',
)
document = RoleRagDocument.objects.create(
role=self.role,
training_file=training_file,
content='Chunk content',
content_hash='a' * 64,
metadata={'source': 'base.txt'},
chunk_index=2,
is_active=True,
)
self.assertEqual(document.role, self.role)
self.assertEqual(document.training_file, training_file)
self.assertEqual(document.content, 'Chunk content')
self.assertEqual(document.content_hash, 'a' * 64)
self.assertEqual(document.metadata, {'source': 'base.txt'})
self.assertEqual(document.chunk_index, 2)
self.assertTrue(document.is_active)
self.assertIsNotNone(document.id)
self.assertIsNotNone(document.uuid)
self.assertIsNotNone(document.created_at)
self.assertIsNotNone(document.updated_at)
self.assertEqual(str(document), 'Analyst - Chunk 2')

0
apps/onboarding/tests/__init__.py Normal file
View file

260
apps/onboarding/tests/test_api.py Normal file
View file

@ -0,0 +1,260 @@
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)

109
apps/onboarding/tests/test_models.py Normal file
View file

@ -0,0 +1,109 @@
from django.contrib.auth import get_user_model
from django.test import TestCase
from apps.accounts.models import Organization, Role
from apps.onboarding.models import AgentConfig, AgentInteractionLog, OnboardingFlow, OnboardingSession
User = get_user_model()
class OnboardingModelTests(TestCase):
def setUp(self):
self.user = User.objects.create_user(
email_address='onboard@example.com',
password='pass1234',
first_name='Onboard',
last_name='User',
date_of_birth='1994-04-04',
is_manager=True,
)
self.org = Organization.objects.create(name='Onboard Org', owner=self.user)
self.org.members.add(self.user)
self.role = Role.objects.create(name='Operator', organization=self.org)
def test_agent_config_fields(self):
config = AgentConfig.objects.create(
organization=self.org,
name='Operator Knowledge Agent',
agent_type='knowledge',
llm_config={'model_id': 'x'},
system_prompt='Assist user',
tool_permissions=['search'],
)
self.assertEqual(config.organization, self.org)
self.assertEqual(config.name, 'Operator Knowledge Agent')
self.assertEqual(config.agent_type, 'knowledge')
self.assertEqual(config.llm_config, {'model_id': 'x'})
self.assertEqual(config.system_prompt, 'Assist user')
self.assertEqual(config.tool_permissions, ['search'])
self.assertIsNotNone(config.id)
self.assertIsNotNone(config.uuid)
self.assertIsNotNone(config.created_at)
self.assertIsNotNone(config.updated_at)
def test_onboarding_session_fields(self):
session = OnboardingSession.objects.create(
user=self.user,
role=self.role,
status='active',
state={'progress': 25},
active_configs={'knowledge': 'enabled'},
)
self.assertEqual(session.user, self.user)
self.assertEqual(session.role, self.role)
self.assertEqual(session.status, 'active')
self.assertEqual(session.state, {'progress': 25})
self.assertEqual(session.active_configs, {'knowledge': 'enabled'})
self.assertIsNone(session.completed_at)
self.assertIsNotNone(session.id)
self.assertIsNotNone(session.uuid)
self.assertIsNotNone(session.created_at)
self.assertIsNotNone(session.updated_at)
def test_agent_interaction_log_fields(self):
config = AgentConfig.objects.create(
organization=self.org,
name='Operator Assistant',
agent_type='assessment',
)
session = OnboardingSession.objects.create(
user=self.user,
role=self.role,
state={},
active_configs={},
)
log = AgentInteractionLog.objects.create(
session=session,
agent_config=config,
sender_type='user',
content='Hello',
tool_call_metadata={'page_uuid': 'abc'},
)
self.assertEqual(log.session, session)
self.assertEqual(log.agent_config, config)
self.assertEqual(log.sender_type, 'user')
self.assertEqual(log.content, 'Hello')
self.assertEqual(log.tool_call_metadata, {'page_uuid': 'abc'})
self.assertIsNotNone(log.id)
self.assertIsNotNone(log.uuid)
self.assertIsNotNone(log.created_at)
self.assertIsNotNone(log.updated_at)
def test_onboarding_flow_fields(self):
flow = OnboardingFlow.objects.create(
title='New Hire Flow',
role=self.role,
structure=[{'uuid': 'p1', 'title': 'Intro'}],
is_active=True,
)
self.assertEqual(flow.title, 'New Hire Flow')
self.assertEqual(flow.role, self.role)
self.assertEqual(flow.structure, [{'uuid': 'p1', 'title': 'Intro'}])
self.assertTrue(flow.is_active)
self.assertIsNotNone(flow.id)
self.assertIsNotNone(flow.uuid)
self.assertIsNotNone(flow.created_at)
self.assertIsNotNone(flow.updated_at)