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

Fixed error cases, tests, viewsets and api paths

Browse source
This commit is contained in:
Viswamedha Nalabotu 2026-01-20 02:59:22 +00:00
parent 0a07e408c5
commit e99c07ef19
9 changed files with 94 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
apps/mlstore/migrations/0001_initial.py
View file

@ -1,5 +1,3 @@
# Generated by Django 5.2.10 on 2026-01-17 16:12
import django.db.models.deletion
import uuid
from django.conf import settings

4
apps/orgs/admin.py
View file

@ -33,10 +33,10 @@ class OrganizationMembershipAdmin(ModelAdmin):
@register(OrganizationInvitation)
class OrganizationInvitationAdmin(ModelAdmin):
list_display = ('id', 'token', 'organization', 'created_by', 'is_active', 'expires_at', 'max_uses', 'created_at')
list_display = ('id', 'token', 'organization', 'created_by', 'is_active', 'expires_at', 'max_uses', 'created_at', 'uses')
search_fields = ('token', 'organization__name', 'created_by__email_address')
list_filter = ('is_active',)
raw_id_fields = ('organization', 'created_by', 'used_by')
raw_id_fields = ('organization', 'created_by')
readonly_fields = ('token', 'created_at')
@register(Role)

2
apps/orgs/migrations/0001_initial.py
View file

@ -36,11 +36,11 @@ class Migration(migrations.Migration):
('id', models.BigAutoField(primary_key=True, serialize=False)),
('token', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
('expires_at', models.DateTimeField()),
('uses', models.IntegerField(default=0)),
('max_uses', models.IntegerField(default=1)),
('is_active', models.BooleanField(default=True)),
('created_by', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='created_invites', to=settings.AUTH_USER_MODEL)),
('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='invite_tokens', to='orgs.organization')),
('used_by', models.ManyToManyField(blank=True, related_name='used_invites', to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'Invite Token',

4
apps/orgs/models.py
View file

@ -48,7 +48,7 @@ class OrganizationInvitation(TimeStampMixin, Model):
expires_at = DateTimeField()
used_by = ManyToManyField(User, blank = True, related_name = "used_invites")
uses = IntegerField(default = 0)
max_uses = IntegerField(default = 1)
is_active = BooleanField(default = True)
@ -63,7 +63,7 @@ class OrganizationInvitation(TimeStampMixin, Model):
super().save(*args, **kwargs)
def is_valid(self):
return self.is_active and not self.used_by.exists() and timezone.now() < self.expires_at
return self.is_active and self.uses < self.max_uses and timezone.now() < self.expires_at
def __str__(self) -> str:
return f"Invite for {self.organization.name} by {self.created_by.full_name} (expires {self.expires_at})"

5
apps/orgs/serializers.py
View file

@ -35,14 +35,13 @@ class OrganizationMembershipSerializer(ModelSerializer):
class OrganizationInvitationSerializer(ModelSerializer):
created_by = UserSerializer(read_only = True)
used_by = UserSerializer(read_only = True, many=True)
invite_url = SerializerMethodField()
is_valid = SerializerMethodField()
class Meta:
model = OrganizationInvitation
fields = ['id', 'token', 'organization', 'created_by', 'expires_at', 'used_by', 'max_uses', 'is_active', 'invite_url', 'is_valid', 'created_at', 'updated_at']
read_only_fields = ['token', 'organization', 'created_by', 'used_by', 'max_uses', 'created_at', 'updated_at']
fields = ['id', 'token', 'organization', 'created_by', 'expires_at', 'max_uses', 'is_active', 'invite_url', 'is_valid', 'created_at', 'updated_at', 'uses']
read_only_fields = ['token', 'organization', 'created_by', 'max_uses', 'created_at', 'updated_at', 'uses']
def get_invite_url(self, obj):
request = self.context.get('request')

8
apps/orgs/tests/test_api.py
View file

@ -39,7 +39,7 @@ class OrganizationAPITests(TestCase):
invite_view = OrganizationViewSet.as_view({'post': 'join'})
req2 = self.factory.post('/', {})
force_authenticate(req2, user=other)
resp2 = invite_view(req2, uuid=str(org.uuid), token=str(token))
resp2 = invite_view(req2, token=str(token))
self.assertIn(resp2.status_code, (HTTP_200_OK, HTTP_201_CREATED))
self.assertTrue(OrganizationMembership.objects.filter(organization=org, user=other).exists())
@ -54,7 +54,7 @@ class OrganizationAPITests(TestCase):
force_authenticate(req, user=self.manager)
resp = members_view(req, uuid=str(org.uuid))
self.assertEqual(resp.status_code, HTTP_200_OK)
self.assertTrue(any(m['user']['email_address'] == 'member@example.com' for m in resp.data))
self.assertTrue(any(m['email_address'] == 'member@example.com' for m in resp.data))
member.is_manager = True
member.save()
@ -94,7 +94,7 @@ class OrganizationAPITests(TestCase):
def test_role_create_forbidden_for_non_manager(self):
org = Organization.objects.create(name='RoleNoCreateOrg', owner=self.user)
OrganizationMembership.objects.create(organization=org, user=self.user)
self.assertFalse(hasattr(OrganizationViewSet, 'role'))
self.assertTrue(hasattr(OrganizationViewSet, 'role'))
def test_role_members_post_missing_user_id_returns_400(self):
org = Organization.objects.create(name='RoleMissingParamOrg', owner=self.manager)
@ -201,7 +201,7 @@ class OrganizationAPITests(TestCase):
invite_view = OrganizationViewSet.as_view({'post': 'join'})
req = self.factory.post('/')
force_authenticate(req, user=other)
resp = invite_view(req, uuid=str(org.uuid), token=str(invite.token))
resp = invite_view(req, token=str(invite.token))
self.assertIn(resp.status_code, (HTTP_400_BAD_REQUEST, HTTP_404_NOT_FOUND))
def test_remove_member_by_non_manager_forbidden(self):

4
apps/orgs/tests/test_models.py
View file

@ -25,11 +25,11 @@ class OrganizationModelTests(TestCase):
self.assertIsNotNone(invite.expires_at)
self.assertTrue(invite.is_valid())
invite.used_by.add(self.user)
invite.uses += 1
invite.save()
self.assertFalse(invite.is_valid())
invite.used_by.clear()
invite.uses = 0
invite.expires_at = timezone.now() - timedelta(days=1)
invite.save()
self.assertFalse(invite.is_valid())

104
apps/orgs/viewsets.py
View file

@ -7,6 +7,8 @@ from rest_framework.response import Response
from rest_framework.status import HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND, HTTP_400_BAD_REQUEST
from rest_framework.decorators import action
from django.utils import timezone
from apps.users.models import User
from apps.users.serializers import UserSerializer
class OrganizationViewSet(ModelViewSet):
@ -39,34 +41,38 @@ class OrganizationViewSet(ModelViewSet):
created_by = request.user,
max_uses = max_uses
)
return Response(OrganizationInvitationSerializer(invitation).data)
return Response(OrganizationInvitationSerializer(invitation, context={'request': request}).data)
@action(detail=True, methods=['post'], url_path='join/(?P<token>[0-9a-f-]{36})')
def join(self, request, uuid = None, token = None):
@action(detail=False, methods=['post'], url_path='join/(?P<token>[0-9a-f-]{36})')
def join(self, request, token = None):
try:
organization = Organization.objects.get(uuid=uuid)
except Organization.DoesNotExist:
return Response({'error': 'Organization not found'}, status=HTTP_403_FORBIDDEN)
try:
invitation = OrganizationInvitation.objects.get(token = token, organization = organization)
invitation = OrganizationInvitation.objects.get(token = token)
except OrganizationInvitation.DoesNotExist:
return Response({'error': 'Invalid invitation token'}, status = HTTP_404_NOT_FOUND)
if not invitation.is_active or invitation.expires_at < timezone.now():
return Response({'error': 'Invitation token is no longer valid'}, status = HTTP_400_BAD_REQUEST)
if OrganizationMembership.objects.filter(user = request.user, organization = organization).exists():
if invitation.uses >= invitation.max_uses:
invitation.is_active = False
invitation.save()
return Response({'error': 'Invitation token has reached its maximum number of uses'}, status = HTTP_400_BAD_REQUEST)
if OrganizationMembership.objects.filter(user = request.user, organization = invitation.organization).exists():
return Response({'error': 'You are already a member of this organization'}, status = HTTP_403_FORBIDDEN)
OrganizationMembership.objects.create(user = request.user, organization = organization)
invitation.max_uses -= 1
if invitation.max_uses <= 0:
OrganizationMembership.objects.create(user = request.user, organization = invitation.organization)
invitation.uses += 1
if invitation.uses >= invitation.max_uses:
invitation.is_active = False
invitation.used_by.add(request.user)
invitation.save()
organization_data = OrganizationSerializer(invitation.organization, context={'request': request}).data
organization_data['message'] = 'Successfully joined the organization'
organization_data['success'] = True
return Response({'message': 'Successfully joined the organization'})
return Response(organization_data)
@action(detail=True, methods=['post'], url_path='leave')
def leave(self, request, uuid = None):
@ -78,7 +84,7 @@ class OrganizationViewSet(ModelViewSet):
if organization.owner == request.user:
return Response({'error': 'The owner cannot leave the organization. Please transfer ownership or delete the organization.'}, status = HTTP_403_FORBIDDEN)
membership.delete()
return Response({'message': 'Successfully left the organization'})
@ -87,8 +93,8 @@ class OrganizationViewSet(ModelViewSet):
if not request.user.is_manager:
return Response({'error': 'Only managers can view invites'}, status = HTTP_403_FORBIDDEN)
organization = self.get_object()
invites = OrganizationInvitation.objects.filter(organization = organization)
serializer = OrganizationInvitationSerializer(invites, many = True)
invites = OrganizationInvitation.objects.filter(organization = organization, is_active = True)
serializer = OrganizationInvitationSerializer(invites, many = True, context={'request': request})
return Response(serializer.data)
@action(detail=True, methods=['get'], url_path='invite/(?P<token>[0-9a-f-]{36})')
@ -100,19 +106,33 @@ class OrganizationViewSet(ModelViewSet):
invitation = OrganizationInvitation.objects.get(token = token, organization = organization)
except OrganizationInvitation.DoesNotExist:
return Response({'error': 'Invalid invitation token'}, status = HTTP_403_FORBIDDEN)
serializer = OrganizationInvitationSerializer(invitation)
serializer = OrganizationInvitationSerializer(invitation, context={'request': request})
return Response(serializer.data)
@action(detail=True, methods=['post', 'delete'], url_path='invite/(?P<token>[0-9a-f-]{36})/revoke')
def revoke_invite(self, request, uuid = None, token = None):
if not request.user.is_manager:
return Response({'error': 'Only managers can revoke invites'}, status = HTTP_403_FORBIDDEN)
organization = self.get_object()
try:
invitation = OrganizationInvitation.objects.get(token = token, organization = organization)
except OrganizationInvitation.DoesNotExist:
return Response({'error': 'Invalid invitation token'}, status = HTTP_403_FORBIDDEN)
invitation.is_active = False
invitation.save()
return Response({'message': 'Invitation successfully revoked'})
@action(detail=True, methods=['get'], url_path='member')
def list_members(self, request, uuid = None):
if not request.user.is_manager:
return Response({'error': 'Only managers can view members'}, status = HTTP_403_FORBIDDEN)
organization = self.get_object()
memberships = OrganizationMembership.objects.filter(organization = organization)
serializer = OrganizationMembershipSerializer(memberships, many = True)
memberships = User.objects.filter(organization_memberships__organization = organization)
serializer = UserSerializer(memberships, many = True)
return Response(serializer.data)
@action(detail=True, methods=['post'], url_path='member/(?P<user_id>\d+)/remove')
@action(detail=True, methods=['post'], url_path=r'member/(?P<user_id>\d+)/remove')
def remove_member(self, request, uuid = None, user_id = None):
if not request.user.is_manager:
return Response({'error': 'Only managers can remove members'}, status = HTTP_403_FORBIDDEN)
@ -128,16 +148,14 @@ class OrganizationViewSet(ModelViewSet):
membership.delete()
return Response({'message': 'Member successfully removed from the organization'})
@action(detail=True, methods=['get'], url_path='role')
def list_roles(self, request, uuid = None):
organization = self.get_object()
roles = Role.objects.filter(organization = organization)
serializer = RoleSerializer(roles, many = True)
return Response(serializer.data)
@action(detail=True, methods=['post'], url_path='role')
def create_role(self, request, uuid = None):
@action(detail=True, methods=['get', 'post'], url_path='role')
def role(self, request, uuid = None):
organization = self.get_object()
if request.method == 'GET':
roles = Role.objects.filter(organization = organization)
serializer = RoleSerializer(roles, many = True)
return Response(serializer.data)
if not request.user.is_manager:
return Response({'error': 'Only managers can create roles'}, status = HTTP_403_FORBIDDEN)
name = request.data.get('name')
@ -147,4 +165,28 @@ class OrganizationViewSet(ModelViewSet):
serializer = RoleSerializer(role)
return Response(serializer.data)
@action(detail=True, methods=['post'], url_path='role/(?P<role_uuid>[0-9a-f-]{36})/delete')
def delete_role(self, request, uuid = None, role_uuid = None):
if not request.user.is_manager:
return Response({'error': 'Only managers can delete roles'}, status = HTTP_403_FORBIDDEN)
organization = self.get_object()
try:
role = Role.objects.get(uuid = role_uuid, organization = organization)
except Role.DoesNotExist:
return Response({'error': 'Role not found in this organization'}, status = HTTP_404_NOT_FOUND)
role.delete()
return Response({'message': 'Role successfully deleted'})
@action(detail=True, methods=['get'], url_path='role/(?P<role_uuid>[0-9a-f-]{36})/member')
def list_role_members(self, request, uuid = None, role_uuid = None):
organization = self.get_object()
try:
role = Role.objects.get(uuid = role_uuid, organization = organization)
except Role.DoesNotExist:
return Response({'error': 'Role not found in this organization'}, status = HTTP_404_NOT_FOUND)
memberships = RoleMembership.objects.filter(role = role)
serializer = RoleMembershipSerializer(memberships, many = True)
return Response(serializer.data)

14
apps/users/viewsets.py
View file

@ -55,13 +55,15 @@ class UserViewSet(ReadOnlyModelViewSet):
email_address = User.objects.normalize_email(email_address)
if User.objects.filter(email_address=email_address).exists():
return Response({'detail': 'Email address already exists.', 'success': False}, status=HTTP_400_BAD_REQUEST)
if not data.get('first_name') or not data.get('last_name'):
return Response({'detail': 'First and last name(s) must be provided.', 'success': False}, status=HTTP_400_BAD_REQUEST)
if not data.get('manager') or data.get('manager').lower() != 'true' and data.get('manager').lower() != 'false':
return Response({'detail': '"manager" field must be true or false.', 'success': False}, status=HTTP_400_BAD_REQUEST)
if type(manager:=data.get('manager')) is not bool:
if manager in ['true', 'True']:
manager = True
elif manager in ['false', 'False']:
manager = False
else:
return Response({'detail': '"manager" field must be a boolean value.', 'success': False}, status=HTTP_400_BAD_REQUEST)
if data.get('password') != data.get('confirm_password'):
return Response({'detail': 'Passwords do not match.', 'success': False}, status=HTTP_400_BAD_REQUEST)
try:
@ -71,7 +73,7 @@ class UserViewSet(ReadOnlyModelViewSet):
first_name=data.get('first_name'),
last_name=data.get('last_name'),
date_of_birth=data.get('date_of_birth'),
is_manager=data.get('manager').lower() == 'true'
is_manager=manager,
)
return Response({'detail': 'User account created successfully.', 'success': True}, status=HTTP_201_CREATED)
except Exception as e: