Added orgs app, tests, api and viewsets

This commit is contained in:
Viswamedha Nalabotu 2026-01-17 15:51:11 +00:00
parent 5f631b9504
commit 44ad993b55
13 changed files with 778 additions and 0 deletions

0
apps/orgs/__init__.py Normal file
View file

53
apps/orgs/admin.py Normal file
View file

@ -0,0 +1,53 @@
from django.contrib.admin import ModelAdmin, TabularInline, register
from apps.orgs.models import Organization, OrganizationInvitation, OrganizationMembership, Role, RoleMembership
class OrganizationMembershipInline(TabularInline):
model = OrganizationMembership
extra = 0
raw_id_fields = ('user',)
class RoleInline(TabularInline):
model = Role
extra = 0
class RoleMembershipInline(TabularInline):
model = RoleMembership
extra = 0
raw_id_fields = ('user',)
@register(Organization)
class OrganizationAdmin(ModelAdmin):
list_display = ('id', 'uuid', 'name', 'owner', 'created_at', 'updated_at')
search_fields = ('name', 'owner__email_address')
list_filter = ('created_at',)
inlines = (OrganizationMembershipInline, RoleInline)
raw_id_fields = ('owner',)
readonly_fields = ('uuid', 'created_at', 'updated_at')
@register(OrganizationMembership)
class OrganizationMembershipAdmin(ModelAdmin):
list_display = ('id', 'user', 'organization', 'is_manager')
search_fields = ('user__email_address', 'organization__name')
list_filter = ('is_manager',)
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')
search_fields = ('token', 'organization__name', 'created_by__email_address')
list_filter = ('is_active',)
raw_id_fields = ('organization', 'created_by', 'used_by')
readonly_fields = ('token', 'created_at')
@register(Role)
class RoleAdmin(ModelAdmin):
list_display = ('id', 'name', 'organization', 'uuid')
search_fields = ('name', 'organization__name')
raw_id_fields = ('organization',)
inlines = (RoleMembershipInline,)
readonly_fields = ('uuid',)
@register(RoleMembership)
class RoleMembershipAdmin(ModelAdmin):
list_display = ('id', 'user', 'role')
raw_id_fields = ('user', 'role')

5
apps/orgs/apps.py Normal file
View file

@ -0,0 +1,5 @@
from django.apps import AppConfig
class OrgsConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'apps.orgs'

View file

@ -0,0 +1,106 @@
import django.db.models.deletion
import uuid
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Organization',
fields=[
('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)),
('uuid', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
('name', models.CharField(max_length=255, unique=True)),
('description', models.TextField(blank=True, default='')),
('owner', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='owned_organizations', to=settings.AUTH_USER_MODEL)),
],
options={
'verbose_name': 'Organization',
'verbose_name_plural': 'Organizations',
},
),
migrations.CreateModel(
name='OrganizationInvitation',
fields=[
('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)),
('token', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
('expires_at', models.DateTimeField()),
('used_at', models.DateTimeField(blank=True, null=True)),
('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)),
],
options={
'verbose_name': 'Invite Token',
'verbose_name_plural': 'Invite Tokens',
},
),
migrations.CreateModel(
name='OrganizationMembership',
fields=[
('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)),
],
options={
'verbose_name': 'Organization Membership',
'verbose_name_plural': 'Organization Memberships',
'unique_together': {('user', 'organization')},
},
),
migrations.AddField(
model_name='organization',
name='members',
field=models.ManyToManyField(related_name='organizations', through='orgs.OrganizationMembership', to=settings.AUTH_USER_MODEL),
),
migrations.CreateModel(
name='Role',
fields=[
('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)),
('name', models.CharField(max_length=100, unique=True)),
('uuid', models.UUIDField(default=uuid.uuid4, editable=False, unique=True)),
('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='roles', to='orgs.organization')),
],
options={
'verbose_name': 'Role',
'verbose_name_plural': 'Roles',
},
),
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')),
('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)),
],
options={
'verbose_name': 'Role Membership',
'verbose_name_plural': 'Role Memberships',
'unique_together': {('user', 'role')},
},
),
migrations.AddField(
model_name='role',
name='members',
field=models.ManyToManyField(related_name='roles', through='orgs.RoleMembership', to=settings.AUTH_USER_MODEL),
),
]

View file

94
apps/orgs/models.py Normal file
View file

@ -0,0 +1,94 @@
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 apps.users.mixins import TimeStampMixin
from apps.users.models import User
class Organization(TimeStampMixin, Model):
id = BigAutoField(primary_key = True)
uuid = UUIDField(default = uuid4, unique = True, editable = False)
name = CharField(max_length = 255, unique = True)
description = TextField(blank = True, default = '')
owner = ForeignKey(User, on_delete = CASCADE, related_name = 'owned_organizations')
members = ManyToManyField(User, through = 'OrganizationMembership', related_name = 'organizations')
class Meta:
verbose_name = _('Organization')
verbose_name_plural = _('Organizations')
def __str__(self) -> str:
return self.name
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')
verbose_name_plural = _('Organization Memberships')
unique_together = [['user', 'organization']]
def __str__(self) -> str:
return f'{self.user.full_name} - {self.organization.name} ({self.is_manager})'
class OrganizationInvitation(TimeStampMixin, Model):
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)
is_active = BooleanField(default = True)
class Meta:
verbose_name = _("Invite Token")
verbose_name_plural = _("Invite Tokens")
def save(self, *args, **kwargs):
if not self.expires_at:
self.expires_at = timezone.now() + timedelta(days=7)
super().save(*args, **kwargs)
def is_valid(self):
return self.is_active and not self.used_by 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})"
class Role(TimeStampMixin, Model):
id = BigAutoField(primary_key = True)
name = CharField(max_length = 100, unique = True)
uuid = UUIDField(default = uuid4, editable = False, unique = True)
organization = ForeignKey(Organization, on_delete = CASCADE, related_name = "roles")
members = ManyToManyField(User, through = "RoleMembership", related_name = "roles")
class Meta:
verbose_name = _('Role')
verbose_name_plural = _('Roles')
def __str__(self) -> str:
return self.name
class RoleMembership(TimeStampMixin, Model):
user = ForeignKey(User, on_delete = CASCADE, related_name = "role_memberships")
role = ForeignKey(Role, on_delete = CASCADE, related_name = "memberships")
class Meta:
verbose_name = _("Role Membership")
verbose_name_plural = _("Role Memberships")
unique_together = [["user", "role"]]
def __str__(self) -> str:
return f"{self.user.full_name} - {self.role.name}"

75
apps/orgs/serializers.py Normal file
View file

@ -0,0 +1,75 @@
from rest_framework.serializers import ModelSerializer, SerializerMethodField, IntegerField
from apps.orgs.models import Organization, OrganizationMembership, OrganizationInvitation, Role, RoleMembership
from apps.users.serializers import UserSerializer
class OrganizationSerializer(ModelSerializer):
owner = UserSerializer(read_only = True)
member_count = SerializerMethodField()
role_count = SerializerMethodField()
class Meta:
model = Organization
fields = ['id', 'uuid', 'name', 'description', 'owner', 'created_at', 'updated_at', 'member_count', 'role_count']
read_only_fields = ['uuid', 'owner', 'created_at', 'updated_at']
def get_member_count(self, obj):
return obj.memberships.count()
def get_role_count(self, obj):
return obj.roles.count()
class OrganizationMembershipSerializer(ModelSerializer):
user = UserSerializer(read_only = True)
user_id = IntegerField(write_only = True, required = False)
class Meta:
model = OrganizationMembership
fields = ['id', 'user', 'user_id', 'organization', 'is_manager', 'created_at']
read_only_fields = ['organization', 'created_at']
def create(self, validated_data):
user_id = validated_data.pop('user_id', None)
if user_id:
validated_data['user_id'] = user_id
return super().create(validated_data)
class OrganizationInvitationSerializer(ModelSerializer):
created_by = UserSerializer(read_only = True)
used_by = UserSerializer(read_only = 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']
def get_invite_url(self, obj):
request = self.context.get('request')
if request:
return request.build_absolute_uri(f'/invite/{obj.token}')
return f'/invite/{obj.token}'
def get_is_valid(self, obj):
return obj.is_valid()
class RoleMembershipSerializer(ModelSerializer):
user = UserSerializer(read_only = True)
class Meta:
model = RoleMembership
fields = ['id', 'user', 'role', 'created_at']
read_only_fields = ['created_at']
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']
def get_member_count(self, obj):
return obj.memberships.count()

View file

228
apps/orgs/tests/test_api.py Normal file
View file

@ -0,0 +1,228 @@
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
User = get_user_model()
class OrganizationAPITests(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.user = User.objects.create_user(email_address='apiuser@example.com', password='pass')
def test_create_organization_creates_membership(self):
data = {'name': 'API Org', 'description': 'Created via API'}
view = OrganizationViewSet.as_view({'post': 'create'})
request = self.factory.post('/', data)
force_authenticate(request, user=self.user)
response = view(request)
self.assertIn(response.status_code, (HTTP_201_CREATED, HTTP_200_OK))
org = Organization.objects.get(name='API Org')
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'})
request = self.factory.post('/', {})
force_authenticate(request, user=self.user)
response = org_view(request, uuid=str(org.uuid))
self.assertEqual(response.status_code, HTTP_201_CREATED)
token = response.data.get('token')
other = User.objects.create_user(email_address='other@example.com', password='pass')
invite_view = InviteViewSet.as_view({'post': 'accept'})
req2 = self.factory.post('/', {})
force_authenticate(req2, user=other)
resp2 = invite_view(req2, token=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)
member = User.objects.create_user(email_address='member@example.com', password='pass')
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
members_view = OrganizationViewSet.as_view({'get': 'members'})
req = self.factory.get('/')
force_authenticate(req, user=self.user)
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.refresh_from_db()
self.assertTrue(OrganizationMembership.objects.get(organization=org, user=member).is_manager)
remove_view = OrganizationViewSet.as_view({'delete': 'remove_member'})
req3 = self.factory.delete('/')
force_authenticate(req3, user=self.user)
resp3 = remove_view(req3, uuid=str(org.uuid), user_id=str(org.owner.id))
self.assertEqual(resp3.status_code, HTTP_400_BAD_REQUEST)
invites_view = OrganizationViewSet.as_view({'post': 'invites', 'get': 'invites'})
req4 = self.factory.post('/')
force_authenticate(req4, user=self.user)
resp4 = invites_view(req4, uuid=str(org.uuid))
self.assertEqual(resp4.status_code, HTTP_201_CREATED)
token = resp4.data.get('token')
req5 = self.factory.get('/')
force_authenticate(req5, user=self.user)
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)
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'})
req = self.factory.post('/')
force_authenticate(req, user=self.user)
resp = view(req, uuid=str(org.uuid))
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)
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)
view = OrganizationViewSet.as_view({'get': 'retrieve'})
req = self.factory.get('/')
force_authenticate(req, user=other)
resp = view(req, uuid=str(org.uuid))
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)
view = OrganizationViewSet.as_view({'get': 'list'})
req = self.factory.get('/')
force_authenticate(req, user=self.user)
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)
view = OrganizationViewSet.as_view({'get': 'list'})
req = self.factory.get('/')
force_authenticate(req, user=other)
resp = view(req)
self.assertEqual(resp.status_code, HTTP_200_OK)
self.assertTrue(any(o['name'] == 'MemberListOrg' for o in resp.data))
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)
view = OrganizationViewSet.as_view({'get': 'list'})
req = self.factory.get('/')
force_authenticate(req, user=outsider)
resp = view(req)
self.assertEqual(resp.status_code, HTTP_200_OK)
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
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)
role = org.roles.create(name='Tester')
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)
member = User.objects.create_user(email_address='monly@example.com', password='pass')
OrganizationMembership.objects.create(organization=org, user=member, is_manager=False)
members_view = OrganizationViewSet.as_view({'get': '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)
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)
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)
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'})
req = self.factory.post('/')
force_authenticate(req, user=other)
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):
org = Organization.objects.create(name='RemoveForbidOrg', owner=self.user)
OrganizationMembership.objects.create(organization=org, user=self.user, is_manager=False)
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('/')
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)
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')
force_authenticate(req, user=self.user)
resp = update_view(req, uuid=str(org.uuid), user_id=str(member.id))
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)
token = OrganizationInvitation.objects.create(organization=org, created_by=self.user)
revoke_view = OrganizationViewSet.as_view({'delete': 'revoke_invite'})
req = self.factory.delete('/')
force_authenticate(req, user=self.user)
resp = revoke_view(req, uuid=str(org.uuid), token=str(token.token))
self.assertEqual(resp.status_code, HTTP_403_FORBIDDEN)

View file

@ -0,0 +1,76 @@
from django.test import TestCase
from django.utils import timezone
from django.contrib.auth import get_user_model
from datetime import timedelta
from apps.orgs.models import Organization, OrganizationMembership, OrganizationInvitation, Role, RoleMembership
User = get_user_model()
class OrganizationModelTests(TestCase):
def setUp(self):
self.user = User.objects.create_user(email_address='u@example.com', password='pass')
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)
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)
invite = OrganizationInvitation.objects.create(organization=org, created_by=self.user)
self.assertIsNotNone(invite.expires_at)
self.assertTrue(invite.is_valid())
invite.used_by = self.user
invite.save()
self.assertFalse(invite.is_valid())
invite.used_by = None
invite.expires_at = timezone.now() - timedelta(days=1)
invite.save()
self.assertFalse(invite.is_valid())
def test_role_and_role_membership(self):
org = Organization.objects.create(name='RoleOrg', owner=self.user)
role = Role.objects.create(name='Admin', organization=org)
rm = RoleMembership.objects.create(role=role, user=self.user)
self.assertIn(role, org.roles.all())
self.assertIn(self.user, role.members.all())
def test_unique_organization_name(self):
Organization.objects.create(name='UniqueOrg', owner=self.user)
with self.assertRaises(Exception):
Organization.objects.create(name='UniqueOrg', owner=self.user)
def test_membership_unique_together(self):
org = Organization.objects.create(name='UTOrg', owner=self.user)
OrganizationMembership.objects.create(organization=org, user=self.user)
with self.assertRaises(Exception):
OrganizationMembership.objects.create(organization=org, user=self.user)
def test_invite_default_expiry_is_seven_days(self):
org = Organization.objects.create(name='ExpiryOrg', owner=self.user)
invite = OrganizationInvitation.objects.create(organization=org, created_by=self.user)
delta = invite.expires_at - invite.created_at
self.assertTrue(6 <= delta.days <= 8)
def test_invite_str_contains_org_name(self):
org = Organization.objects.create(name='StrOrg', owner=self.user)
invite = OrganizationInvitation.objects.create(organization=org, created_by=self.user)
self.assertIn('StrOrg', str(invite))
def test_role_uuid_and_unique(self):
org = Organization.objects.create(name='RoleUuidOrg', owner=self.user)
r1 = Role.objects.create(name='R1', organization=org)
r2 = Role.objects.create(name='R2', organization=org)
self.assertNotEqual(r1.uuid, r2.uuid)
def test_str_methods(self):
org = Organization.objects.create(name='StrTestOrg', owner=self.user)
m = OrganizationMembership.objects.create(organization=org, user=self.user)
self.assertIn(org.name, str(m))

136
apps/orgs/viewsets.py Normal file
View file

@ -0,0 +1,136 @@
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
from apps.orgs.serializers import OrganizationSerializer, OrganizationMembershipSerializer, OrganizationInvitationSerializer, RoleSerializer
class OrganizationViewSet(ModelViewSet):
queryset = Organization.objects.all()
serializer_class = OrganizationSerializer
permission_classes = [IsAuthenticated]
lookup_field = 'uuid'
def get_queryset(self):
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)
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:
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)
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)
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=['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)
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)
@action(detail=True, methods=['get', 'post'])
def invites(self, request, uuid=None):
org = self.get_object()
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)
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)
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)
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()

View file

@ -1,8 +1,12 @@
from rest_framework.routers import DefaultRouter
from apps.orgs.viewsets import OrganizationViewSet, InviteViewSet, RoleViewSet
from apps.users.viewsets import UserViewSet
router = DefaultRouter()
router.register(r'user', UserViewSet, basename='user')
router.register(r'organization', OrganizationViewSet, basename='organization')
router.register(r'invite', InviteViewSet, basename='invite')
router.register(r'role', RoleViewSet, basename='role')
urlpatterns = router.urls

View file

@ -65,6 +65,7 @@ THIRD_PARTY_APPS = [
]
LOCAL_APPS = [
'apps.users',
'apps.orgs',
]
INSTALLED_APPS = OVERRIDE_APPS + DJANGO_APPS + THIRD_PARTY_APPS + LOCAL_APPS