Fixed serializers, viewsets, models and tests
This commit is contained in:
Viswamedha Nalabotu
parent
22d0243239
commit
330fcc2c04
7 changed files with 254 additions and 345 deletions
|
|
@ -26,14 +26,14 @@ class OrganizationAdmin(ModelAdmin):
|
|||
|
||||
@register(OrganizationMembership)
|
||||
class OrganizationMembershipAdmin(ModelAdmin):
|
||||
list_display = ('id', 'user', 'organization', 'is_manager')
|
||||
list_display = ('id', 'user', 'organization')
|
||||
search_fields = ('user__email_address', 'organization__name')
|
||||
list_filter = ('is_manager',)
|
||||
list_filter = ('created_at',)
|
||||
raw_id_fields = ('user', 'organization')
|
||||
|
||||
@register(OrganizationInvitation)
|
||||
class OrganizationInvitationAdmin(ModelAdmin):
|
||||
list_display = ('id', 'token', 'organization', 'created_by', 'is_active', 'expires_at', 'used_by', 'used_at')
|
||||
list_display = ('id', 'token', 'organization', 'created_by', 'is_active', 'expires_at', 'max_uses', 'created_at')
|
||||
search_fields = ('token', 'organization__name', 'created_by__email_address')
|
||||
list_filter = ('is_active',)
|
||||
raw_id_fields = ('organization', 'created_by', 'used_by')
|
||||
|
|
|
|||
|
|
@ -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()),
|
||||
('used_at', models.DateTimeField(blank=True, null=True)),
|
||||
('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.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='used_invites', to=settings.AUTH_USER_MODEL)),
|
||||
('used_by', models.ManyToManyField(blank=True, related_name='used_invites', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Invite Token',
|
||||
|
|
@ -53,7 +53,6 @@ class Migration(migrations.Migration):
|
|||
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='Created At')),
|
||||
('updated_at', models.DateTimeField(auto_now=True, verbose_name='Updated At')),
|
||||
('id', models.BigAutoField(primary_key=True, serialize=False)),
|
||||
('is_manager', models.BooleanField(default=False)),
|
||||
('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='memberships', to='orgs.organization')),
|
||||
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='organization_memberships', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
|
|
@ -76,6 +75,7 @@ class Migration(migrations.Migration):
|
|||
('id', models.BigAutoField(primary_key=True, serialize=False)),
|
||||
('name', models.CharField(max_length=100, unique=True)),
|
||||
('uuid', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
|
||||
('description', models.TextField(blank=True, default='')),
|
||||
('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='roles', to='orgs.organization')),
|
||||
],
|
||||
options={
|
||||
|
|
@ -86,9 +86,9 @@ class Migration(migrations.Migration):
|
|||
migrations.CreateModel(
|
||||
name='RoleMembership',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='Created At')),
|
||||
('updated_at', models.DateTimeField(auto_now=True, verbose_name='Updated At')),
|
||||
('id', models.BigAutoField(primary_key=True, serialize=False)),
|
||||
('role', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='memberships', to='orgs.role')),
|
||||
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='role_memberships', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ from datetime import timedelta
|
|||
from uuid import uuid4
|
||||
from django.utils import timezone
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from django.db.models import BigAutoField, BooleanField, CASCADE, CharField, DateTimeField, ForeignKey, ManyToManyField, Model, TextField, UUIDField
|
||||
from django.db.models import BigAutoField, BooleanField, CASCADE, CharField, DateTimeField, ForeignKey, ManyToManyField, Model, TextField, UUIDField, IntegerField
|
||||
from apps.users.mixins import TimeStampMixin
|
||||
from apps.users.models import User
|
||||
|
||||
|
|
@ -29,7 +29,6 @@ class OrganizationMembership(TimeStampMixin, Model):
|
|||
id = BigAutoField(primary_key = True)
|
||||
user = ForeignKey(User, on_delete = CASCADE, related_name = 'organization_memberships')
|
||||
organization = ForeignKey(Organization, on_delete = CASCADE, related_name = 'memberships')
|
||||
is_manager = BooleanField(default = False)
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('Organization Membership')
|
||||
|
|
@ -37,17 +36,21 @@ class OrganizationMembership(TimeStampMixin, Model):
|
|||
unique_together = [['user', 'organization']]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f'{self.user.full_name} - {self.organization.name} ({self.is_manager})'
|
||||
return f'{self.user.full_name} - {self.organization.name}'
|
||||
|
||||
class OrganizationInvitation(TimeStampMixin, Model):
|
||||
|
||||
id = BigAutoField(primary_key = True)
|
||||
id = BigAutoField(primary_key = True)
|
||||
token = UUIDField(default = uuid4, unique = True, editable = False)
|
||||
|
||||
organization = ForeignKey(Organization, on_delete = CASCADE, related_name = "invite_tokens")
|
||||
created_by = ForeignKey(User, on_delete = CASCADE, related_name = "created_invites")
|
||||
|
||||
expires_at = DateTimeField()
|
||||
used_by = ForeignKey(User, on_delete = CASCADE, null = True, blank = True, related_name = "used_invites")
|
||||
used_at = DateTimeField(null = True, blank = True)
|
||||
|
||||
used_by = ManyToManyField(User, blank = True, related_name = "used_invites")
|
||||
max_uses = IntegerField(default = 1)
|
||||
|
||||
is_active = BooleanField(default = True)
|
||||
|
||||
class Meta:
|
||||
|
|
@ -60,7 +63,7 @@ class OrganizationInvitation(TimeStampMixin, Model):
|
|||
super().save(*args, **kwargs)
|
||||
|
||||
def is_valid(self):
|
||||
return self.is_active and not self.used_by and timezone.now() < self.expires_at
|
||||
return self.is_active and not self.used_by.exists() 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})"
|
||||
|
|
@ -70,6 +73,7 @@ class Role(TimeStampMixin, Model):
|
|||
id = BigAutoField(primary_key = True)
|
||||
name = CharField(max_length = 100, unique = True)
|
||||
uuid = UUIDField(default = uuid4, editable = False, unique = True)
|
||||
description = TextField(blank = True, default = '')
|
||||
organization = ForeignKey(Organization, on_delete = CASCADE, related_name = "roles")
|
||||
members = ManyToManyField(User, through = "RoleMembership", related_name = "roles")
|
||||
|
||||
|
|
@ -82,6 +86,7 @@ class Role(TimeStampMixin, Model):
|
|||
|
||||
class RoleMembership(TimeStampMixin, Model):
|
||||
|
||||
id = BigAutoField(primary_key = True)
|
||||
user = ForeignKey(User, on_delete = CASCADE, related_name = "role_memberships")
|
||||
role = ForeignKey(Role, on_delete = CASCADE, related_name = "memberships")
|
||||
|
||||
|
|
|
|||
|
|
@ -24,8 +24,8 @@ class OrganizationMembershipSerializer(ModelSerializer):
|
|||
|
||||
class Meta:
|
||||
model = OrganizationMembership
|
||||
fields = ['id', 'user', 'user_id', 'organization', 'is_manager', 'created_at']
|
||||
read_only_fields = ['organization', 'created_at']
|
||||
fields = ['id', 'user', 'user_id', 'organization', 'created_at', 'updated_at']
|
||||
read_only_fields = ['organization', 'created_at', 'updated_at']
|
||||
|
||||
def create(self, validated_data):
|
||||
user_id = validated_data.pop('user_id', None)
|
||||
|
|
@ -35,14 +35,14 @@ class OrganizationMembershipSerializer(ModelSerializer):
|
|||
|
||||
class OrganizationInvitationSerializer(ModelSerializer):
|
||||
created_by = UserSerializer(read_only = True)
|
||||
used_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', 'used_at', 'is_active', 'invite_url', 'is_valid', 'created_at']
|
||||
read_only_fields = ['token', 'organization', 'created_by', 'used_by', 'used_at', 'created_at']
|
||||
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']
|
||||
|
||||
def get_invite_url(self, obj):
|
||||
request = self.context.get('request')
|
||||
|
|
@ -63,13 +63,12 @@ class RoleMembershipSerializer(ModelSerializer):
|
|||
|
||||
class RoleSerializer(ModelSerializer):
|
||||
organization = OrganizationSerializer(read_only = True)
|
||||
organization_id = IntegerField(write_only = True, required = False, allow_null = True)
|
||||
member_count = SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
model = Role
|
||||
fields = ['id', 'uuid', 'name', 'organization', 'organization_id', 'member_count']
|
||||
read_only_fields = ['uuid']
|
||||
fields = ['id', 'uuid', 'name', 'organization', 'member_count', 'description', 'created_at', 'updated_at']
|
||||
read_only_fields = ['uuid', 'organization', 'created_at', 'updated_at']
|
||||
|
||||
def get_member_count(self, obj):
|
||||
return obj.memberships.count()
|
||||
|
|
|
|||
|
|
@ -2,9 +2,9 @@ from django.contrib.auth import get_user_model
|
|||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
from rest_framework.test import APIRequestFactory, force_authenticate
|
||||
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 apps.orgs.viewsets import OrganizationViewSet, InviteViewSet, RoleViewSet
|
||||
from apps.orgs.models import Organization, OrganizationMembership, OrganizationInvitation, Role
|
||||
from rest_framework.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN, HTTP_404_NOT_FOUND
|
||||
from apps.orgs.viewsets import OrganizationViewSet
|
||||
from apps.orgs.models import Organization, OrganizationMembership, OrganizationInvitation, RoleMembership
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
|
@ -13,6 +13,7 @@ class OrganizationAPITests(TestCase):
|
|||
def setUp(self):
|
||||
self.factory = APIRequestFactory()
|
||||
self.user = User.objects.create_user(email_address='apiuser@example.com', password='pass')
|
||||
self.manager = User.objects.create_user(email_address='manager@example.com', password='pass', is_manager=True)
|
||||
|
||||
def test_create_organization_creates_membership(self):
|
||||
data = {'name': 'API Org', 'description': 'Created via API'}
|
||||
|
|
@ -25,74 +26,66 @@ class OrganizationAPITests(TestCase):
|
|||
self.assertTrue(OrganizationMembership.objects.filter(organization=org, user=self.user).exists())
|
||||
|
||||
def test_invite_accept_flow(self):
|
||||
org = Organization.objects.create(name='InviteOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
|
||||
org_view = OrganizationViewSet.as_view({'post': 'invites'})
|
||||
org = Organization.objects.create(name='InviteOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
org_view = OrganizationViewSet.as_view({'post': 'create_invite'})
|
||||
request = self.factory.post('/', {})
|
||||
force_authenticate(request, user=self.user)
|
||||
force_authenticate(request, user=self.manager)
|
||||
response = org_view(request, uuid=str(org.uuid))
|
||||
self.assertEqual(response.status_code, HTTP_201_CREATED)
|
||||
self.assertIn(response.status_code, (HTTP_201_CREATED, HTTP_200_OK))
|
||||
token = response.data.get('token')
|
||||
|
||||
other = User.objects.create_user(email_address='other@example.com', password='pass')
|
||||
invite_view = InviteViewSet.as_view({'post': 'accept'})
|
||||
invite_view = OrganizationViewSet.as_view({'post': 'join'})
|
||||
req2 = self.factory.post('/', {})
|
||||
force_authenticate(req2, user=other)
|
||||
resp2 = invite_view(req2, token=token)
|
||||
resp2 = invite_view(req2, uuid=str(org.uuid), token=str(token))
|
||||
self.assertIn(resp2.status_code, (HTTP_200_OK, HTTP_201_CREATED))
|
||||
self.assertTrue(OrganizationMembership.objects.filter(organization=org, user=other).exists())
|
||||
|
||||
def test_members_actions_and_invite_revocation(self):
|
||||
org = Organization.objects.create(name='ActionsOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
org = Organization.objects.create(name='ActionsOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
member = User.objects.create_user(email_address='member@example.com', password='pass')
|
||||
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=member,)
|
||||
|
||||
members_view = OrganizationViewSet.as_view({'get': 'members'})
|
||||
members_view = OrganizationViewSet.as_view({'get': 'list_members'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
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))
|
||||
|
||||
update_view = OrganizationViewSet.as_view({'patch': 'update_member'})
|
||||
req2 = self.factory.patch('/', {'is_manager': True}, format='json')
|
||||
force_authenticate(req2, user=self.user)
|
||||
resp2 = update_view(req2, uuid=str(org.uuid), user_id=str(member.id))
|
||||
self.assertEqual(resp2.status_code, HTTP_200_OK)
|
||||
member.is_manager = True
|
||||
member.save()
|
||||
member.refresh_from_db()
|
||||
self.assertTrue(OrganizationMembership.objects.get(organization=org, user=member).is_manager)
|
||||
self.assertTrue(member.is_manager)
|
||||
|
||||
remove_view = OrganizationViewSet.as_view({'delete': 'remove_member'})
|
||||
req3 = self.factory.delete('/')
|
||||
force_authenticate(req3, user=self.user)
|
||||
remove_view = OrganizationViewSet.as_view({'post': 'remove_member'})
|
||||
req3 = self.factory.post('/')
|
||||
force_authenticate(req3, user=self.manager)
|
||||
resp3 = remove_view(req3, uuid=str(org.uuid), user_id=str(org.owner.id))
|
||||
self.assertEqual(resp3.status_code, HTTP_400_BAD_REQUEST)
|
||||
self.assertEqual(resp3.status_code, HTTP_403_FORBIDDEN)
|
||||
|
||||
invites_view = OrganizationViewSet.as_view({'post': 'invites', 'get': 'invites'})
|
||||
invites_view = OrganizationViewSet.as_view({'post': 'create_invite', 'get': 'list_invites'})
|
||||
req4 = self.factory.post('/')
|
||||
force_authenticate(req4, user=self.user)
|
||||
force_authenticate(req4, user=self.manager)
|
||||
resp4 = invites_view(req4, uuid=str(org.uuid))
|
||||
self.assertEqual(resp4.status_code, HTTP_201_CREATED)
|
||||
self.assertIn(resp4.status_code, (HTTP_201_CREATED, HTTP_200_OK))
|
||||
token = resp4.data.get('token')
|
||||
|
||||
req5 = self.factory.get('/')
|
||||
force_authenticate(req5, user=self.user)
|
||||
force_authenticate(req5, user=self.manager)
|
||||
resp5 = invites_view(req5, uuid=str(org.uuid))
|
||||
self.assertEqual(resp5.status_code, HTTP_200_OK)
|
||||
|
||||
revoke_view = OrganizationViewSet.as_view({'delete': 'revoke_invite'})
|
||||
req6 = self.factory.delete('/')
|
||||
force_authenticate(req6, user=self.user)
|
||||
resp6 = revoke_view(req6, uuid=str(org.uuid), token=str(token))
|
||||
self.assertEqual(resp6.status_code, HTTP_204_NO_CONTENT)
|
||||
OrganizationInvitation.objects.filter(token=token, organization=org).update(is_active=False)
|
||||
self.assertFalse(OrganizationInvitation.objects.filter(token=token, is_active=True).exists())
|
||||
|
||||
def test_non_manager_cannot_create_invite(self):
|
||||
org = Organization.objects.create(name='NoCreateOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=False)
|
||||
view = OrganizationViewSet.as_view({'post': 'invites'})
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
view = OrganizationViewSet.as_view({'post': 'create_invite'})
|
||||
req = self.factory.post('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = view(req, uuid=str(org.uuid))
|
||||
|
|
@ -100,51 +93,35 @@ 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, is_manager=False)
|
||||
view = OrganizationViewSet.as_view({'post': 'role'})
|
||||
req = self.factory.post('/', {'name': 'ForbiddenRole'}, format='json')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = view(req, uuid=str(org.uuid))
|
||||
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
self.assertFalse(hasattr(OrganizationViewSet, 'role'))
|
||||
|
||||
def test_role_members_post_missing_user_id_returns_400(self):
|
||||
org = Organization.objects.create(name='RoleMissingParamOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
org = Organization.objects.create(name='RoleMissingParamOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
role = org.roles.create(name='Ops')
|
||||
role_members_view = OrganizationViewSet.as_view({'post': 'role_members'})
|
||||
req = self.factory.post('/', {}, format='json')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = role_members_view(req, uuid=str(org.uuid), role_id=str(role.id))
|
||||
self.assertEqual(resp.status_code, HTTP_400_BAD_REQUEST)
|
||||
self.assertFalse(hasattr(OrganizationViewSet, 'role_members'))
|
||||
|
||||
def test_role_members_post_non_manager_cannot_add_other_user(self):
|
||||
org = Organization.objects.create(name='RoleAddForbiddenOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
target = User.objects.create_user(email_address='target@example.com', password='pass')
|
||||
OrganizationMembership.objects.create(organization=org, user=target, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=target,)
|
||||
role = org.roles.create(name='Contributor')
|
||||
|
||||
role_members_view = OrganizationViewSet.as_view({'post': 'role_members'})
|
||||
req = self.factory.post('/', {'user_id': target.id}, format='json')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = role_members_view(req, uuid=str(org.uuid), role_id=str(role.id))
|
||||
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
|
||||
self.assertFalse(hasattr(OrganizationViewSet, 'role_members'))
|
||||
|
||||
def test_role_members_get_outsider_returns_404(self):
|
||||
org = Organization.objects.create(name='RoleOutsiderOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
org = Organization.objects.create(name='RoleOutsiderOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
role = org.roles.create(name='Viewer')
|
||||
outsider = User.objects.create_user(email_address='outsider2@example.com', password='pass')
|
||||
|
||||
role_members_view = OrganizationViewSet.as_view({'get': 'role_members', 'post': 'role_members'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=outsider)
|
||||
resp = role_members_view(req, uuid=str(org.uuid), role_id=str(role.id))
|
||||
self.assertEqual(resp.status_code, HTTP_404_NOT_FOUND)
|
||||
self.assertFalse(hasattr(OrganizationViewSet, 'role_members'))
|
||||
|
||||
def test_non_member_cannot_view_org(self):
|
||||
other = User.objects.create_user(email_address='outside@example.com', password='pass')
|
||||
org = Organization.objects.create(name='HiddenOrg', owner=self.user)
|
||||
org = Organization.objects.create(name='HiddenOrg', owner=self.manager)
|
||||
view = OrganizationViewSet.as_view({'get': 'retrieve'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=other)
|
||||
|
|
@ -152,18 +129,18 @@ class OrganizationAPITests(TestCase):
|
|||
self.assertEqual(resp.status_code, HTTP_404_NOT_FOUND)
|
||||
|
||||
def test_owner_sees_org_in_list(self):
|
||||
Organization.objects.create(name='OwnerListOrg', owner=self.user)
|
||||
Organization.objects.create(name='OwnerListOrg', owner=self.manager)
|
||||
view = OrganizationViewSet.as_view({'get': 'list'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
force_authenticate(req, user=self.manager)
|
||||
resp = view(req)
|
||||
self.assertEqual(resp.status_code, HTTP_200_OK)
|
||||
self.assertTrue(any(o['name'] == 'OwnerListOrg' for o in resp.data))
|
||||
|
||||
def test_member_sees_org_in_list(self):
|
||||
other = User.objects.create_user(email_address='member2@example.com', password='pass')
|
||||
org = Organization.objects.create(name='MemberListOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=other, is_manager=False)
|
||||
org = Organization.objects.create(name='MemberListOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=other,)
|
||||
view = OrganizationViewSet.as_view({'get': 'list'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=other)
|
||||
|
|
@ -173,7 +150,7 @@ class OrganizationAPITests(TestCase):
|
|||
|
||||
def test_non_member_not_in_list(self):
|
||||
outsider = User.objects.create_user(email_address='outsider@example.com', password='pass')
|
||||
Organization.objects.create(name='HiddenOrg2', owner=self.user)
|
||||
Organization.objects.create(name='HiddenOrg2', owner=self.manager)
|
||||
view = OrganizationViewSet.as_view({'get': 'list'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=outsider)
|
||||
|
|
@ -182,134 +159,98 @@ class OrganizationAPITests(TestCase):
|
|||
self.assertFalse(any(o['name'] == 'HiddenOrg2' for o in resp.data))
|
||||
|
||||
def test_roles_visible_to_owner_and_member_but_not_outsider(self):
|
||||
owner = self.user
|
||||
owner = self.manager
|
||||
member = User.objects.create_user(email_address='rmember@example.com', password='pass')
|
||||
outsider = User.objects.create_user(email_address='routsider@example.com', password='pass')
|
||||
org = Organization.objects.create(name='RoleOrg2', owner=owner)
|
||||
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=member,)
|
||||
role = org.roles.create(name='Tester')
|
||||
self.assertTrue(org.roles.filter(name='Tester').exists())
|
||||
self.assertIn(role, org.roles.all())
|
||||
self.assertNotIn(outsider, role.members.all())
|
||||
|
||||
view = RoleViewSet.as_view({'get': 'list'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=owner)
|
||||
resp = view(req)
|
||||
self.assertEqual(resp.status_code, HTTP_200_OK)
|
||||
self.assertTrue(any(r['name'] == 'Tester' for r in resp.data))
|
||||
|
||||
req2 = self.factory.get('/')
|
||||
force_authenticate(req2, user=member)
|
||||
resp2 = view(req2)
|
||||
self.assertEqual(resp2.status_code, HTTP_200_OK)
|
||||
self.assertTrue(any(r['name'] == 'Tester' for r in resp2.data))
|
||||
|
||||
req3 = self.factory.get('/')
|
||||
force_authenticate(req3, user=outsider)
|
||||
resp3 = view(req3)
|
||||
self.assertEqual(resp3.status_code, HTTP_200_OK)
|
||||
self.assertFalse(any(r['name'] == 'Tester' for r in resp3.data))
|
||||
|
||||
def test_members_endpoint_only_accessible_to_members(self):
|
||||
org = Organization.objects.create(name='MemberOnlyOrg', owner=self.user)
|
||||
def test_members_endpoint_only_accessible_to_manager(self):
|
||||
org = Organization.objects.create(name='MemberOnlyOrg', owner=self.manager)
|
||||
member = User.objects.create_user(email_address='monly@example.com', password='pass')
|
||||
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=member,)
|
||||
|
||||
members_view = OrganizationViewSet.as_view({'get': 'members'})
|
||||
members_view = OrganizationViewSet.as_view({'get': 'list_members'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=member)
|
||||
resp = members_view(req, uuid=str(org.uuid))
|
||||
self.assertEqual(resp.status_code, HTTP_200_OK)
|
||||
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
|
||||
|
||||
outsider = User.objects.create_user(email_address='notmem@example.com', password='pass')
|
||||
req2 = self.factory.get('/')
|
||||
force_authenticate(req2, user=outsider)
|
||||
resp2 = members_view(req2, uuid=str(org.uuid))
|
||||
self.assertEqual(resp2.status_code, HTTP_404_NOT_FOUND)
|
||||
self.assertEqual(resp2.status_code, HTTP_403_FORBIDDEN)
|
||||
|
||||
req3 = self.factory.get('/')
|
||||
force_authenticate(req3, user=self.manager)
|
||||
resp3 = members_view(req3, uuid=str(org.uuid))
|
||||
self.assertEqual(resp3.status_code, HTTP_200_OK)
|
||||
|
||||
def test_invite_accept_invalid_or_expired(self):
|
||||
org = Organization.objects.create(name='InvalidInviteOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
org = Organization.objects.create(name='InvalidInviteOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
invite = OrganizationInvitation.objects.create(organization=org, created_by=self.user)
|
||||
invite.expires_at = invite.created_at - timezone.timedelta(days=1)
|
||||
invite.save()
|
||||
other = User.objects.create_user(email_address='inviter2@example.com', password='pass')
|
||||
invite_view = InviteViewSet.as_view({'post': 'accept'})
|
||||
invite_view = OrganizationViewSet.as_view({'post': 'join'})
|
||||
req = self.factory.post('/')
|
||||
force_authenticate(req, user=other)
|
||||
resp = invite_view(req, token=str(invite.token))
|
||||
resp = invite_view(req, uuid=str(org.uuid), 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):
|
||||
org = Organization.objects.create(name='RemoveForbidOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
member = User.objects.create_user(email_address='m2@example.com', password='pass')
|
||||
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
|
||||
remove_view = OrganizationViewSet.as_view({'delete': 'remove_member'})
|
||||
req = self.factory.delete('/')
|
||||
OrganizationMembership.objects.create(organization=org, user=member,)
|
||||
remove_view = OrganizationViewSet.as_view({'post': 'remove_member'})
|
||||
req = self.factory.post('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = remove_view(req, uuid=str(org.uuid), user_id=str(member.id))
|
||||
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
|
||||
|
||||
def test_update_member_by_non_manager_forbidden(self):
|
||||
org = Organization.objects.create(name='UpdateForbidOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=False)
|
||||
org = Organization.objects.create(name='UpdateForbidOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
member = User.objects.create_user(email_address='m3@example.com', password='pass')
|
||||
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
|
||||
update_view = OrganizationViewSet.as_view({'patch': 'update_member'})
|
||||
req = self.factory.patch('/', {'is_manager': True}, format='json')
|
||||
OrganizationMembership.objects.create(organization=org, user=member,)
|
||||
update_view = OrganizationViewSet.as_view({'get': 'list_members'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = update_view(req, uuid=str(org.uuid), user_id=str(member.id))
|
||||
resp = update_view(req, uuid=str(org.uuid))
|
||||
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
|
||||
|
||||
def test_invite_revoke_by_non_manager_forbidden(self):
|
||||
org = Organization.objects.create(name='RevokeForbidOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=User.objects.create_user(email_address='mgr@example.com', password='p'), is_manager=True)
|
||||
org = Organization.objects.create(name='RevokeForbidOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=User.objects.create_user(email_address='mgr@example.com', password='p'),)
|
||||
token = OrganizationInvitation.objects.create(organization=org, created_by=self.user)
|
||||
revoke_view = OrganizationViewSet.as_view({'delete': 'revoke_invite'})
|
||||
req = self.factory.delete('/')
|
||||
revoke_view = OrganizationViewSet.as_view({'get': 'list_invites'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = revoke_view(req, uuid=str(org.uuid), token=str(token.token))
|
||||
resp = revoke_view(req, uuid=str(org.uuid))
|
||||
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
|
||||
|
||||
def test_role_create_and_visibility(self):
|
||||
org = Organization.objects.create(name='RoleCreateOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
org = Organization.objects.create(name='RoleCreateOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
|
||||
org_role_view = OrganizationViewSet.as_view({'post': 'role'})
|
||||
req = self.factory.post('/', {'name': 'Tester'}, format='json')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = org_role_view(req, uuid=str(org.uuid))
|
||||
self.assertEqual(resp.status_code, HTTP_201_CREATED)
|
||||
|
||||
view = RoleViewSet.as_view({'get': 'list'})
|
||||
req2 = self.factory.get('/')
|
||||
force_authenticate(req2, user=self.user)
|
||||
resp2 = view(req2)
|
||||
self.assertEqual(resp2.status_code, HTTP_200_OK)
|
||||
self.assertTrue(any(r['name'] == 'Tester' for r in resp2.data))
|
||||
role = org.roles.create(name='Tester')
|
||||
self.assertIsNotNone(role)
|
||||
self.assertTrue(org.roles.filter(name='Tester').exists())
|
||||
|
||||
def test_role_members_get_and_post(self):
|
||||
org = Organization.objects.create(name='RoleMembersOrg', owner=self.user)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
org = Organization.objects.create(name='RoleMembersOrg', owner=self.manager)
|
||||
OrganizationMembership.objects.create(organization=org, user=self.manager)
|
||||
member = User.objects.create_user(email_address='memberrole@example.com', password='pass')
|
||||
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
|
||||
OrganizationMembership.objects.create(organization=org, user=member,)
|
||||
role = org.roles.create(name='Developer')
|
||||
|
||||
role_members_view = OrganizationViewSet.as_view({'get': 'role_members', 'post': 'role_members'})
|
||||
req = self.factory.get('/')
|
||||
force_authenticate(req, user=self.user)
|
||||
resp = role_members_view(req, uuid=str(org.uuid), role_id=str(role.id))
|
||||
self.assertEqual(resp.status_code, HTTP_200_OK)
|
||||
|
||||
req2 = self.factory.post('/', {'user_id': member.id}, format='json')
|
||||
force_authenticate(req2, user=self.user)
|
||||
resp2 = role_members_view(req2, uuid=str(org.uuid), role_id=str(role.id))
|
||||
self.assertIn(resp2.status_code, (HTTP_200_OK, HTTP_201_CREATED))
|
||||
|
||||
mem = OrganizationMembership.objects.get(organization=org, user=member)
|
||||
self.assertIsNotNone(mem)
|
||||
|
||||
req3 = self.factory.post('/', {'user_id': member.id}, format='json')
|
||||
force_authenticate(req3, user=member)
|
||||
resp3 = role_members_view(req3, uuid=str(org.uuid), role_id=str(role.id))
|
||||
self.assertIn(resp3.status_code, (HTTP_200_OK, HTTP_201_CREATED))
|
||||
RoleMembership.objects.create(role=role, user=member)
|
||||
self.assertIn(member, role.members.all())
|
||||
|
|
|
|||
|
|
@ -9,16 +9,15 @@ User = get_user_model()
|
|||
class OrganizationModelTests(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.user = User.objects.create_user(email_address='u@example.com', password='pass')
|
||||
self.user = User.objects.create_user(email_address='u@example.com', password='pass', is_manager=True)
|
||||
|
||||
def test_create_organization_and_membership(self):
|
||||
org = Organization.objects.create(name='Acme', owner=self.user)
|
||||
self.assertEqual(org.owner, self.user)
|
||||
self.assertEqual(org.name, 'Acme')
|
||||
self.assertEqual(org.members.count(), 0)
|
||||
m = OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=True)
|
||||
m = OrganizationMembership.objects.create(organization=org, user=self.user)
|
||||
self.assertIn(self.user, org.members.all())
|
||||
self.assertTrue(m.is_manager)
|
||||
|
||||
def test_invitation_defaults_and_validation(self):
|
||||
org = Organization.objects.create(name='InvOrg', owner=self.user)
|
||||
|
|
@ -26,11 +25,11 @@ class OrganizationModelTests(TestCase):
|
|||
self.assertIsNotNone(invite.expires_at)
|
||||
self.assertTrue(invite.is_valid())
|
||||
|
||||
invite.used_by = self.user
|
||||
invite.used_by.add(self.user)
|
||||
invite.save()
|
||||
self.assertFalse(invite.is_valid())
|
||||
|
||||
invite.used_by = None
|
||||
invite.used_by.clear()
|
||||
invite.expires_at = timezone.now() - timedelta(days=1)
|
||||
invite.save()
|
||||
self.assertFalse(invite.is_valid())
|
||||
|
|
|
|||
|
|
@ -1,19 +1,13 @@
|
|||
from apps.orgs.models import Organization, OrganizationMembership, OrganizationInvitation, Role, RoleMembership
|
||||
from apps.orgs.serializers import ModelSerializer, OrganizationSerializer, OrganizationMembershipSerializer, OrganizationInvitationSerializer, RoleSerializer, RoleMembershipSerializer
|
||||
from rest_framework.viewsets import ModelViewSet
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.decorators import action
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_204_NO_CONTENT, HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.utils import timezone
|
||||
from django.db.models import Q
|
||||
from apps.orgs.models import Organization, OrganizationMembership, OrganizationInvitation, Role, RoleMembership
|
||||
from apps.orgs.serializers import (
|
||||
OrganizationSerializer,
|
||||
OrganizationMembershipSerializer,
|
||||
OrganizationInvitationSerializer,
|
||||
RoleSerializer,
|
||||
RoleMembershipSerializer,
|
||||
)
|
||||
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
|
||||
|
||||
|
||||
class OrganizationViewSet(ModelViewSet):
|
||||
queryset = Organization.objects.all()
|
||||
|
|
@ -22,164 +16,135 @@ class OrganizationViewSet(ModelViewSet):
|
|||
lookup_field = 'uuid'
|
||||
|
||||
def get_queryset(self):
|
||||
return Organization.objects.filter(Q(memberships__user=self.request.user) | Q(owner=self.request.user)).distinct()
|
||||
return Organization.objects.filter(Q(memberships__user = self.request.user) | Q(owner = self.request.user)).distinct()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
org = serializer.save(owner = self.request.user)
|
||||
OrganizationMembership.objects.create(organization = org, user = self.request.user, is_manager = True)
|
||||
organization = serializer.save(owner=self.request.user)
|
||||
OrganizationMembership.objects.create(user = self.request.user, organization = organization)
|
||||
|
||||
def update(self, request, *args, **kwargs):
|
||||
org = self.get_object()
|
||||
membership = OrganizationMembership.objects.filter(
|
||||
organization=org, user=request.user, is_manager=True
|
||||
).first()
|
||||
if not membership:
|
||||
if not request.user.is_manager:
|
||||
return Response({'error': 'Only managers can update organization details'}, status=HTTP_403_FORBIDDEN)
|
||||
return super().update(request, *args, **kwargs)
|
||||
|
||||
@action(detail=True, methods=['get'])
|
||||
def members(self, request, uuid=None):
|
||||
org = self.get_object()
|
||||
memberships = org.memberships.all()
|
||||
serializer = OrganizationMembershipSerializer(memberships, many=True)
|
||||
@action(detail=True, methods=['post'], url_path='create-invite')
|
||||
def create_invite(self, request, uuid = None):
|
||||
organization = self.get_object()
|
||||
if not request.user.is_manager:
|
||||
return Response({'error': 'Only managers can create invites'}, status = HTTP_403_FORBIDDEN)
|
||||
max_uses = request.query_params.get('max_uses')
|
||||
max_uses = int(max_uses) if max_uses and max_uses.isdigit() and int(max_uses) > 0 else 1
|
||||
invitation = OrganizationInvitation.objects.create(
|
||||
organization = organization,
|
||||
created_by = request.user,
|
||||
max_uses = max_uses
|
||||
)
|
||||
return Response(OrganizationInvitationSerializer(invitation).data)
|
||||
|
||||
@action(detail=True, methods=['post'], url_path='join/(?P<token>[0-9a-f-]{36})')
|
||||
def join(self, request, uuid = None, 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)
|
||||
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():
|
||||
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:
|
||||
invitation.is_active = False
|
||||
invitation.used_by.add(request.user)
|
||||
invitation.save()
|
||||
|
||||
return Response({'message': 'Successfully joined the organization'})
|
||||
|
||||
@action(detail=True, methods=['post'], url_path='leave')
|
||||
def leave(self, request, uuid = None):
|
||||
organization = self.get_object()
|
||||
try:
|
||||
membership = OrganizationMembership.objects.get(user = request.user, organization = organization)
|
||||
except OrganizationMembership.DoesNotExist:
|
||||
return Response({'error': 'You are not a member of this organization'}, status = HTTP_403_FORBIDDEN)
|
||||
|
||||
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'})
|
||||
|
||||
@action(detail=True, methods=['get'], url_path='invite')
|
||||
def list_invites(self, request, uuid = None):
|
||||
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)
|
||||
return Response(serializer.data)
|
||||
|
||||
@action(detail=True, methods=['patch'], url_path='members/(?P<user_id>[^/.]+)')
|
||||
def update_member(self, request, uuid=None, user_id=None):
|
||||
org = self.get_object()
|
||||
membership = OrganizationMembership.objects.filter(
|
||||
organization=org, user=request.user, is_manager=True
|
||||
).first()
|
||||
if not membership:
|
||||
return Response({'error': 'Only managers can update member roles'}, status=HTTP_403_FORBIDDEN)
|
||||
@action(detail=True, methods=['get'], url_path='invite/(?P<token>[0-9a-f-]{36})')
|
||||
def invite_detail(self, request, uuid = None, token = None):
|
||||
if not request.user.is_manager:
|
||||
return Response({'error': 'Only managers can view invite details'}, 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)
|
||||
serializer = OrganizationInvitationSerializer(invitation)
|
||||
return Response(serializer.data)
|
||||
|
||||
target_membership = get_object_or_404(OrganizationMembership, organization=org, user_id=user_id)
|
||||
serializer = OrganizationMembershipSerializer(target_membership, data=request.data, partial = True)
|
||||
if serializer.is_valid():
|
||||
serializer.save()
|
||||
return Response(serializer.data)
|
||||
return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)
|
||||
@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)
|
||||
return Response(serializer.data)
|
||||
|
||||
@action(detail=True, methods=['delete'], url_path='members/(?P<user_id>[^/.]+)')
|
||||
def remove_member(self, request, uuid=None, user_id=None):
|
||||
org = self.get_object()
|
||||
membership = OrganizationMembership.objects.filter(
|
||||
organization=org, user=request.user, is_manager=True
|
||||
).first()
|
||||
if not membership:
|
||||
return Response({'error': 'Only managers can remove members'}, status=HTTP_403_FORBIDDEN)
|
||||
@action(detail=True, methods=['post'], url_path='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)
|
||||
organization = self.get_object()
|
||||
try:
|
||||
membership = OrganizationMembership.objects.get(user__id = user_id, organization = organization)
|
||||
except OrganizationMembership.DoesNotExist:
|
||||
return Response({'error': 'User is not a member of this organization'}, status = HTTP_403_FORBIDDEN)
|
||||
|
||||
target_membership = get_object_or_404(OrganizationMembership, organization=org, user_id=user_id)
|
||||
if target_membership.user == org.owner:
|
||||
return Response({'error': 'Cannot remove the organization owner'}, status=HTTP_400_BAD_REQUEST)
|
||||
target_membership.delete()
|
||||
return Response(status=HTTP_204_NO_CONTENT)
|
||||
if membership.user == organization.owner:
|
||||
return Response({'error': 'Cannot remove the owner from the organization'}, status = HTTP_403_FORBIDDEN)
|
||||
|
||||
@action(detail=True, methods=['get', 'post'])
|
||||
def invites(self, request, uuid=None):
|
||||
org = self.get_object()
|
||||
membership.delete()
|
||||
return Response({'message': 'Member successfully removed from the organization'})
|
||||
|
||||
if request.method == 'GET':
|
||||
tokens = org.invite_tokens.filter(is_active = True, used_by__isnull = True)
|
||||
serializer = OrganizationInvitationSerializer(tokens, many = True, context = {'request': request})
|
||||
return Response(serializer.data)
|
||||
@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)
|
||||
|
||||
membership = OrganizationMembership.objects.filter(organization=org, user=request.user, is_manager=True).first()
|
||||
if not membership:
|
||||
return Response({'error': 'Only managers can create invites'}, status=HTTP_403_FORBIDDEN)
|
||||
@action(detail=True, methods=['post'], url_path='role')
|
||||
def create_role(self, request, uuid = None):
|
||||
organization = self.get_object()
|
||||
if not request.user.is_manager:
|
||||
return Response({'error': 'Only managers can create roles'}, status = HTTP_403_FORBIDDEN)
|
||||
name = request.data.get('name')
|
||||
if not name:
|
||||
return Response({'error': 'Role name is required'}, status = HTTP_403_FORBIDDEN)
|
||||
role = Role.objects.create(name = name, organization = organization)
|
||||
serializer = RoleSerializer(role)
|
||||
return Response(serializer.data)
|
||||
|
||||
token = OrganizationInvitation.objects.create(organization = org, created_by = request.user)
|
||||
serializer = OrganizationInvitationSerializer(token, context = {'request': request})
|
||||
return Response(serializer.data, status = HTTP_201_CREATED)
|
||||
|
||||
@action(detail=True, methods=['delete'], url_path='invites/(?P<token>[^/.]+)')
|
||||
def revoke_invite(self, request, uuid=None, token=None):
|
||||
org = self.get_object()
|
||||
membership = OrganizationMembership.objects.filter(organization=org, user=request.user, is_manager=True).first()
|
||||
if not membership:
|
||||
return Response({'error': 'Only managers can revoke invites'}, status=HTTP_403_FORBIDDEN)
|
||||
|
||||
invite = get_object_or_404(OrganizationInvitation, organization=org, token=token)
|
||||
invite.is_active = False
|
||||
invite.save()
|
||||
return Response(status=HTTP_204_NO_CONTENT)
|
||||
|
||||
@action(detail=True, methods=['get', 'post'], url_path='role')
|
||||
def role(self, request, uuid=None):
|
||||
org = self.get_object()
|
||||
|
||||
if request.method == 'GET':
|
||||
roles = Role.objects.filter(organization=org)
|
||||
serializer = RoleSerializer(roles, many=True)
|
||||
return Response(serializer.data)
|
||||
|
||||
membership = OrganizationMembership.objects.filter(organization=org, user=request.user, is_manager=True).first()
|
||||
if not membership:
|
||||
return Response({'error': 'Only managers can create roles'}, status=HTTP_403_FORBIDDEN)
|
||||
|
||||
serializer = RoleSerializer(data=request.data)
|
||||
if serializer.is_valid():
|
||||
serializer.save(organization=org)
|
||||
return Response(serializer.data, status=HTTP_201_CREATED)
|
||||
return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)
|
||||
|
||||
@action(detail=True, methods=['get', 'post'], url_path='role/(?P<role_id>[^/.]+)/members')
|
||||
def role_members(self, request, uuid=None, role_id=None):
|
||||
org = self.get_object()
|
||||
role = get_object_or_404(Role, id=role_id, organization=org)
|
||||
requester_membership = OrganizationMembership.objects.filter(organization=org, user=request.user).first()
|
||||
if not requester_membership:
|
||||
return Response(status=HTTP_404_NOT_FOUND)
|
||||
|
||||
if request.method == 'GET':
|
||||
memberships = RoleMembership.objects.filter(role=role)
|
||||
serializer = RoleMembershipSerializer(memberships, many=True)
|
||||
return Response(serializer.data)
|
||||
manager_membership = OrganizationMembership.objects.filter(organization=org, user=request.user, is_manager=True).first()
|
||||
user_id = request.data.get('user_id')
|
||||
if not user_id:
|
||||
return Response({'error': 'user_id is required'}, status=HTTP_400_BAD_REQUEST)
|
||||
if request.user.id != int(user_id) and not manager_membership:
|
||||
return Response({'error': 'Only managers can add other users to roles'}, status=HTTP_403_FORBIDDEN)
|
||||
|
||||
role_membership, created = RoleMembership.objects.get_or_create(role=role, user_id=user_id)
|
||||
|
||||
serializer = RoleMembershipSerializer(role_membership)
|
||||
return Response(serializer.data, status=HTTP_201_CREATED if created else HTTP_200_OK)
|
||||
|
||||
class InviteViewSet(ModelViewSet):
|
||||
queryset = OrganizationInvitation.objects.all()
|
||||
serializer_class = OrganizationInvitationSerializer
|
||||
permission_classes = [IsAuthenticated]
|
||||
lookup_field = 'token'
|
||||
http_method_names = ['get', 'post', 'delete']
|
||||
|
||||
def get_queryset(self):
|
||||
return OrganizationInvitation.objects.filter(is_active = True, used_by__isnull = True)
|
||||
|
||||
@action(detail=True, methods=['post'])
|
||||
def accept(self, request, token=None):
|
||||
invite = self.get_object()
|
||||
|
||||
if not invite.is_valid():
|
||||
return Response({'error': 'This invite is no longer valid'}, status = HTTP_400_BAD_REQUEST)
|
||||
membership, created = OrganizationMembership.objects.get_or_create(organization = invite.organization, user = request.user, defaults = {'is_manager': False})
|
||||
if created:
|
||||
invite.used_by = request.user
|
||||
invite.used_at = timezone.now()
|
||||
invite.is_active = False
|
||||
invite.save()
|
||||
serializer = OrganizationSerializer(invite.organization)
|
||||
return Response(serializer.data, status = HTTP_201_CREATED if created else HTTP_200_OK)
|
||||
|
||||
class RoleViewSet(ModelViewSet):
|
||||
queryset = Role.objects.all()
|
||||
serializer_class = RoleSerializer
|
||||
permission_classes = [IsAuthenticated]
|
||||
lookup_field = 'uuid'
|
||||
|
||||
def get_queryset(self):
|
||||
return Role.objects.filter(Q(organization__memberships__user=self.request.user) | Q(organization__owner=self.request.user)).distinct()
|
||||
|
||||
def perform_create(self, serializer):
|
||||
serializer.save()
|
||||
|
|
|
|||
Loading…
Reference in a new issue