jjvijgen/OpenAssetManager
1
1
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Finish api for inviting people to your organization(s)

Browse Source
This commit is contained in:
Jeroen Vijgen
2025-07-12 13:51:56 +00:00
parent b65c292d83
commit 8b73307551
9 changed files with 685 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/asset_manager/src/migrations/models/0_20250709154845_init.py → api/asset_manager/src/migrations/models/3_20250710123814_None.py
+5 -4
View File
@@ -55,8 +55,8 @@ CREATE TABLE IF NOT EXISTS "membership" (
"modified_at" TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
"disabled_at" TIMESTAMP,
"acl_id" CHAR(36) NOT NULL REFERENCES "acl" ("id") ON DELETE CASCADE,
"organization_id" CHAR(36) NOT NULL REFERENCES "organization" ("id") ON DELETE CASCADE,
"user_id" CHAR(36) NOT NULL REFERENCES "user" ("id") ON DELETE CASCADE
"organization_id" CHAR(36) REFERENCES "organization" ("id") ON DELETE SET NULL,
"user_id" CHAR(36) REFERENCES "user" ("id") ON DELETE SET NULL
) /* Membership */;
CREATE TABLE IF NOT EXISTS "invite" (
"id" CHAR(36) NOT NULL PRIMARY KEY,
@@ -64,11 +64,12 @@ CREATE TABLE IF NOT EXISTS "invite" (
"sender" CHAR(36) NOT NULL,
"org_id" CHAR(36) NOT NULL,
"message" TEXT,
"accepted" INT NOT NULL,
"accepted" INT NOT NULL DEFAULT 0,
"disabled" INT NOT NULL DEFAULT 0,
"created_at" TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
"modified_at" TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
"disabled_at" TIMESTAMP
"disabled_at" TIMESTAMP,
"acl_id" CHAR(36) NOT NULL REFERENCES "acl" ("id") ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS "aerich" (
"id" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
api/asset_manager/src/modules/auth/router.py
+17 -8
View File
@@ -1,13 +1,12 @@
from datetime import datetime
from typing import Annotated
import uuid
from fastapi.security import OAuth2PasswordRequestForm
from fastapi.routing import APIRouter
import pytz
from modules.users.utils import get_current_active_user
from modules.users.models import User
from modules.auth.utils import create_jwt_tokens, get_tokens_from_logged_in_user
from modules.auth.models import Token
from modules.users.models import User
from fastapi import Depends, HTTPException, status
from tortoise.expressions import Q
from config import settings
@@ -31,23 +30,31 @@ async def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
Logs the user into our API, creates tokens and passes them back to User.
"""
user: User | None = await User.filter(Q(email=form.username) | Q(username=form.username)).first()
user: User | None = await User.filter(
Q(email=form.username) | Q(username=form.username)
).first()
if user is None:
raise HTTPException(status_code=401, detail=account_error)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail=account_error
)
if user.check_against_password(form.password) is False:
raise HTTPException(status_code=401, detail=account_error)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail=account_error
)
if user.disabled is True:
raise HTTPException(status_code=401, detail=account_error)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail=account_error
)
tokens = await create_jwt_tokens(user)
return {"jwt": tokens}
@router.get("/logout", status_code=204)
@router.get("/logout", status_code=status.HTTP_204_NO_CONTENT)
async def logout(user: Annotated[User, Depends(get_current_active_user)]):
"""
Logout
@@ -110,7 +117,9 @@ async def refresh_login(
return {"jwt": tokens}
@router.post("/register", status_code=201, response_model=user_model)
@router.post(
"/register", status_code=status.HTTP_201_CREATED, response_model=user_model
)
async def register(user: register_model):
# Prevent existing users from reapplying for our system.
existing_user: User | None = await User.filter(
api/asset_manager/src/modules/invitations/models.py
+16 -1
View File
@@ -1,6 +1,10 @@
from datetime import datetime
import uuid
import pytz
from tortoise import Model, fields
from modules.users.models import ACL
class Invite(Model):
id: uuid.UUID = fields.UUIDField(primary_key=True)
@@ -8,8 +12,19 @@ class Invite(Model):
sender: uuid.UUID = fields.UUIDField()
org_id: uuid.UUID = fields.UUIDField()
message: str | None = fields.TextField(null=True)
accepted: bool = fields.BooleanField()
acl: ACL = fields.ForeignKeyField("models.ACL")
accepted: bool = fields.BooleanField(default=False)
disabled: bool = fields.BooleanField(default=False)
created_at = fields.DatetimeField(null=True, auto_now_add=True)
modified_at = fields.DatetimeField(null=True, auto_now=True)
disabled_at = fields.DatetimeField(null=True)
async def delete(self, force: bool = False) -> None:
if force:
await Model.delete(self)
else:
self.disabled = True
self.disabled_at = datetime.now(tz=pytz.UTC)
await self.save()
api/asset_manager/src/modules/invitations/router.py
+173 -46
View File
@@ -1,36 +1,54 @@
from typing import Annotated, List
import uuid
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends, HTTPException, status
from modules.organizations.models import Organization
from modules.invitations.models import Invite
from modules.invitations.schemas import invitation_model
from modules.invitations.schemas import invitation_model, send_invitation_for_org
from modules.users.models import User
from modules.users.models import ACL, Membership, User
from modules.users.utils import get_current_active_user
from tortoise.expressions import Q
router = APIRouter(prefix="/api/v1/Users", tags=["User"])
router = APIRouter(prefix="/api/v1/invitations", tags=["invites"])
@router.get("/", response_model=List[invitation_model])
async def get_all_invitations(
user: Annotated[User, Depends(get_current_active_user)],
) -> List[Invite]:
invites: List[Invite] | None = await Invite.filter(
Q(receiver=user.username) | Q(receiver=user.email)
"""Returns all invitations for user requesting, except disabled invites.
Args:
user (Annotated[User, Depends(get_current_active_user)]): Returns user token.
Returns:
List[Invite]: A list of invitations.
"""
return await Invite.filter(
Q(receiver=user.username) | Q(receiver=user.email) & Q(disabled=False)
)
return invites
@router.get("/accept/{invitation_id}", response_model=invitation_model)
async def accept_invitation(
@router.delete("/{invitation_id}", response_model=invitation_model)
async def delete_invitation(
user: Annotated[User, Depends(get_current_active_user)], invitation_id: uuid.UUID
) -> Invite:
) -> None:
"""Removes an invitation you have sent
Args:
user (Annotated[User, Depends(get_current_active_user)]): Returns user token.
invitation_id (uuid.UUID): UUID for the invitation to be removed
Raises:
HTTPException: When invitation doesn't exist return 403.
Returns:
Invite: The Invitation model.
"""
invite: Invite | None = await Invite.get_or_none(
Q(id=invitation_id) & (Q(receiver=user.username) | Q(receiver=user.email))
Q(id=invitation_id) & Q(sender=user.id) & Q(disabled=False)
)
if not invite:
@@ -39,44 +57,29 @@ async def accept_invitation(
detail="The invitation doesn't exist or you don't have access to it.",
)
invite.accepted = True
return await invite.save()
await invite.delete()
return invite
@router.get("/reject/{invitation_id}", response_model=invitation_model)
async def reject_invitation(
user: Annotated[User, Depends(get_current_active_user)], invitation_id: uuid.UUID
) -> Invite:
invite: Invite | None = await Invite.get_or_none(
Q(id=invitation_id) & (Q(receiver=user.username) | Q(receiver=user.email))
)
if not invite:
raise HTTPException(
status_code=403,
detail="The invitation doesn't exist or you don't have access to it.",
)
invite.accepted = False
return await invite.save()
@router.get("/send")
async def send_invitation(
user: Annotated[User, Depends(get_current_active_user)],
):
# Check if user is Manager or Higher to send an invitation.
# Should send an E-Mail as notification.
pass
@router.get("/cancel/{invitation_id}", response_model=invitation_model)
@router.get("/accept/{invitation_id}", status_code=status.HTTP_204_NO_CONTENT)
async def accept_invitation(
user: Annotated[User, Depends(get_current_active_user)], invitation_id: uuid.UUID
) -> None:
"""Accepts the invitation sent by a different organization.
Args:
user (Annotated[User, Depends(get_current_active_user)]): Returns user token.
invitation_id (uuid.UUID): UUID for the organization that the users wants to add the person to.
Raises:
HTTPException: Raises exception when invite is not available or disabled.
"""
invite: Invite | None = await Invite.get_or_none(
Q(id=invitation_id) & Q(sender=user.id)
)
Q(id=invitation_id)
& (Q(receiver=user.username) | Q(receiver=user.email))
& Q(disabled=False)
).prefetch_related("acl")
if not invite:
raise HTTPException(
@@ -84,4 +87,128 @@ async def accept_invitation(
detail="The invitation doesn't exist or you don't have access to it.",
)
return await invite.delete()
if invite.disabled:
raise HTTPException(
status_code=403,
detail="You have already declined the invitation or the invitation was removed, you can't accept it.",
)
invite.accepted = True
await invite.save()
# Disable invite after accepting, prevent changing it.
await invite.delete()
await Membership.create(
user=user, organization=await Organization.get(id=invite.org_id), acl=invite.acl
)
@router.get("/decline/{invitation_id}", status_code=status.HTTP_204_NO_CONTENT)
async def reject_invitation(
user: Annotated[User, Depends(get_current_active_user)], invitation_id: uuid.UUID
) -> None:
"""Declines an invitation to join an organization
Args:
user (Annotated[User, Depends(get_current_active_user)]): Returns user token.
invitation_id (uuid.UUID): The UUID of the invitation
Raises:
HTTPException: Checks if the invite exists.
HTTPException: Checks if the invite has already accepted.
Returns:
Invite: The Invitation model.
"""
invite: Invite | None = await Invite.get_or_none(
Q(id=invitation_id)
& (Q(receiver=user.username) | Q(receiver=user.email))
& Q(disabled=False)
).prefetch_related("acl")
if not invite:
raise HTTPException(
status_code=403,
detail="The invitation doesn't exist or you don't have access to it.",
)
if invite.accepted:
raise HTTPException(
status_code=403,
detail="The invitation was already accepted, you can't remove it.",
)
await invite.delete()
@router.post("/send", response_model=invitation_model)
async def send_invitation(
user: Annotated[User, Depends(get_current_active_user)],
invite_details: send_invitation_for_org,
) -> Invite:
"""Sends an invitation to e-mail or username.
Args:
user (Annotated[User, Depends(get_current_active_user)]): Returns user token.
invite_details (send_invitation_for_org): The details for the invitation.
Raises:
HTTPException: Checks access to the organization posted.
HTTPException: Checks for Manager or Admin permissions and declines if you are not.
Returns:
Invite: The Invitation model.
"""
# Should send an E-Mail as notification.
membership = await Membership.get_or_none(
Q(user=user.id) & Q(organization=invite_details.org_id)
).prefetch_related("acl")
if not membership:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have access to this organization.",
)
if not membership.acl.MANAGE or not membership.acl.ADMIN:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You are not allowed to send invitations for this organization.",
)
# Check if user is already part of organization
invited_user: User | None = await User.get_or_none(
Q(username=invite_details.receiver) | Q(email=invite_details.receiver)
)
if invited_user:
user_is_part_of_org = await Membership.get_or_none(
Q(user=invited_user) & Q(organization=invite_details.org_id)
)
if user_is_part_of_org:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="The person you've invited is already part of the organization.",
)
acl = None
if invite_details.acl:
acl = await ACL.create(
READ=invite_details.acl.READ,
WRITE=invite_details.acl.WRITE,
REPORT=invite_details.acl.REPORT,
MANAGE=invite_details.acl.MANAGE,
ADMIN=invite_details.acl.ADMIN,
)
else:
acl = await ACL.create(
READ=True,
)
return await Invite.create(
receiver=invite_details.receiver,
sender=user.id,
org_id=invite_details.org_id,
message=invite_details.message,
acl=acl,
)
api/asset_manager/src/modules/invitations/schemas.py
+14 -1
View File
@@ -1,7 +1,20 @@
import uuid
from pydantic import BaseModel
from tortoise.contrib.pydantic import pydantic_model_creator
from modules.invitations.models import Invite
invitation_model = pydantic_model_creator(Invite)
class acl_model(BaseModel):
READ: bool
WRITE: bool
REPORT: bool
MANAGE: bool
ADMIN: bool
class send_invitation_for_org(BaseModel):
org_id: uuid.UUID
receiver: str
acl: acl_model | None
message: str | None
api/asset_manager/src/modules/users/models.py
+2 -2
View File
@@ -99,8 +99,8 @@ class Membership(Model):
"""
id: uuid.UUID = fields.UUIDField(primary_key=True)
organization: Organization = fields.ForeignKeyField("models.Organization")
user: User = fields.ForeignKeyField("models.User")
organization: Organization = fields.ForeignKeyField("models.Organization", null=True, on_delete=fields.SET_NULL)
user: User = fields.ForeignKeyField("models.User", null=True, on_delete=fields.SET_NULL)
acl: ACL = fields.ForeignKeyField("models.ACL")
disabled: bool = fields.BooleanField(default=False)
created_at = fields.DatetimeField(null=True, auto_now_add=True)
api/asset_manager/src/tests/test_authentication/test_authentication.py
-1
View File
@@ -1,6 +1,5 @@
from tests.base_test import Test
from modules.users.models import User
import pytest # type: ignore
from httpx import AsyncClient
from config import settings
from unittest.mock import ANY
api/asset_manager/src/tests/test_invitation_routes/test_invitation_routes.py
+457
View File
@@ -0,0 +1,457 @@
from unittest.mock import ANY
from httpx import AsyncClient
from modules.users.models import ACL, Membership
from tests.base_test import Test
class TestInvitationalRoutes(Test):
async def test_send_invitations(self, client: AsyncClient, create_user_with_org):
admin, org, _, admintokens = await create_user_with_org(
email="superadmin12@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
_, _, _, usertokens = await create_user_with_org(email="user1231@localhost.com")
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "user1231@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 200
assert invite.json() == {
"accepted": False,
"created_at": ANY,
"disabled": False,
"disabled_at": None,
"id": ANY,
"message": "Hi! We would like to invite you to our organization.",
"modified_at": ANY,
"org_id": str(org.id),
"receiver": "user1231@localhost.com",
"sender": str(admin.id),
}
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == [
{
"id": ANY,
"receiver": "user1231@localhost.com",
"sender": str(admin.id),
"org_id": str(org.id),
"message": "Hi! We would like to invite you to our organization.",
"accepted": False,
"disabled": False,
"created_at": ANY,
"modified_at": ANY,
"disabled_at": None,
}
]
async def test_cannot_see_others_invitations(
self, client: AsyncClient, create_user_with_org
):
admin, org, _, admintokens = await create_user_with_org(
email="superadmin99@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
_, _, _, usertokens = await create_user_with_org(email="user18@localhost.com")
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "user1231@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 200
assert invite.json() == {
"accepted": False,
"created_at": ANY,
"disabled": False,
"disabled_at": None,
"id": ANY,
"message": "Hi! We would like to invite you to our organization.",
"modified_at": ANY,
"org_id": str(org.id),
"receiver": "user1231@localhost.com",
"sender": str(admin.id),
}
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == []
async def test_accept_sent_invitations(
self, client: AsyncClient, create_user_with_org
):
admin, org, _, admintokens = await create_user_with_org(
email="superadmin191@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
_, _, _, usertokens = await create_user_with_org(email="user8@localhost.com")
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "user8@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 200
assert invite.json() == {
"accepted": False,
"created_at": ANY,
"disabled": False,
"disabled_at": None,
"id": ANY,
"message": "Hi! We would like to invite you to our organization.",
"modified_at": ANY,
"org_id": str(org.id),
"receiver": "user8@localhost.com",
"sender": str(admin.id),
}
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == [
{
"id": ANY,
"receiver": "user8@localhost.com",
"sender": str(admin.id),
"org_id": str(org.id),
"message": "Hi! We would like to invite you to our organization.",
"accepted": False,
"disabled": False,
"created_at": ANY,
"modified_at": ANY,
"disabled_at": None,
}
]
invite_id = user_invites.json()[0]["id"]
accept_invite = await client.get(
f"https://localhost/api/v1/invitations/accept/{invite_id}",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert accept_invite.status_code == 204
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
# When an invite has been accepted, it should be removed.
assert user_invites.json() == []
async def test_decline_sent_invitations(
self, client: AsyncClient, create_user_with_org
):
admin, org, _, admintokens = await create_user_with_org(
email="superadmin11@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
_, _, _, usertokens = await create_user_with_org(email="user98@localhost.com")
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "user98@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 200
assert invite.json() == {
"accepted": False,
"created_at": ANY,
"disabled": False,
"disabled_at": None,
"id": ANY,
"message": "Hi! We would like to invite you to our organization.",
"modified_at": ANY,
"org_id": str(org.id),
"receiver": "user98@localhost.com",
"sender": str(admin.id),
}
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == [
{
"id": ANY,
"receiver": "user98@localhost.com",
"sender": str(admin.id),
"org_id": str(org.id),
"message": "Hi! We would like to invite you to our organization.",
"accepted": False,
"disabled": False,
"created_at": ANY,
"modified_at": ANY,
"disabled_at": None,
}
]
invite_id = user_invites.json()[0]["id"]
accept_invite = await client.get(
f"https://localhost/api/v1/invitations/decline/{invite_id}",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert accept_invite.status_code == 204
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == []
async def test_prevent_accepting_when_declined_sent_invitations(
self, client: AsyncClient, create_user_with_org
):
admin, org, _, admintokens = await create_user_with_org(
email="superadmin9612@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
_, _, _, usertokens = await create_user_with_org(email="user11918@localhost.com")
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "user11918@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 200
assert invite.json() == {
"accepted": False,
"created_at": ANY,
"disabled": False,
"disabled_at": None,
"id": ANY,
"message": "Hi! We would like to invite you to our organization.",
"modified_at": ANY,
"org_id": str(org.id),
"receiver": "user11918@localhost.com",
"sender": str(admin.id),
}
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == [
{
"id": ANY,
"receiver": "user11918@localhost.com",
"sender": str(admin.id),
"org_id": str(org.id),
"message": "Hi! We would like to invite you to our organization.",
"accepted": False,
"disabled": False,
"created_at": ANY,
"modified_at": ANY,
"disabled_at": None,
}
]
invite_id = user_invites.json()[0]["id"]
accept_invite = await client.get(
f"https://localhost/api/v1/invitations/decline/{invite_id}",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert accept_invite.status_code == 204
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == []
accept_invite = await client.get(
f"https://localhost/api/v1/invitations/accept/{invite_id}",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert accept_invite.status_code == 403
assert accept_invite.json() == {
"detail": "The invitation doesn't exist or you don't have access to it."
}
async def test_prevent_declining_when_accepted_sent_invitations(
self, client: AsyncClient, create_user_with_org
):
admin, org, _, admintokens = await create_user_with_org(
email="superadmin9712@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
_, _, _, usertokens = await create_user_with_org(email="user14918@localhost.com")
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "user14918@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 200
assert invite.json() == {
"accepted": False,
"created_at": ANY,
"disabled": False,
"disabled_at": None,
"id": ANY,
"message": "Hi! We would like to invite you to our organization.",
"modified_at": ANY,
"org_id": str(org.id),
"receiver": "user14918@localhost.com",
"sender": str(admin.id),
}
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == [
{
"id": ANY,
"receiver": "user14918@localhost.com",
"sender": str(admin.id),
"org_id": str(org.id),
"message": "Hi! We would like to invite you to our organization.",
"accepted": False,
"disabled": False,
"created_at": ANY,
"modified_at": ANY,
"disabled_at": None,
}
]
invite_id = user_invites.json()[0]["id"]
accept_invite = await client.get(
f"https://localhost/api/v1/invitations/accept/{invite_id}",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert accept_invite.status_code == 204
user_invites = await client.get(
"https://localhost/api/v1/invitations/",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert user_invites.status_code == 200
assert user_invites.json() == []
decline_invite = await client.get(
f"https://localhost/api/v1/invitations/accept/{invite_id}",
headers={"Authorization": f"Bearer {usertokens.access_token}"},
)
assert decline_invite.status_code == 403
assert decline_invite.json() == {
"detail": "The invitation doesn't exist or you don't have access to it."
}
async def test_prevent_adding_user_whom_already_belongs_to_your_organization(
self, client: AsyncClient, create_user_with_org
):
_, org, _, admintokens = await create_user_with_org(
email="us3r123@localhost.com",
username="awesomeadmin",
password="awesomeadmin",
is_admin=True,
)
user, _, _, _ = await create_user_with_org(email="us3r1234@localhost.com")
await Membership.create(
user=user,
organization=org,
acl=await ACL.create(READ=True)
)
invite = await client.post(
"https://localhost/api/v1/invitations/send",
json={
"org_id": str(org.id),
"receiver": "us3r1234@localhost.com",
"acl": None,
"message": "Hi! We would like to invite you to our organization.",
},
headers={"Authorization": f"Bearer {admintokens.access_token}"},
)
assert invite.status_code == 403
assert invite.json() == {
"detail": "The person you've invited is already part of the organization."
}
api/asset_manager/src/tests/test_organizations_routes/test_organization_routes.py
+1 -6
View File
@@ -1,12 +1,9 @@
from httpx import AsyncClient
from modules.users.models import ACL, Membership
from modules.organizations.models import Organization
from config import settings
from unittest.mock import ANY
from tests.base_test import Test
crypt = settings.CRYPT
class TestOrganizationRoute(Test):
async def test_get_organizations_from_api(
@@ -153,9 +150,7 @@ class TestOrganizationRoute(Test):
READ=True, WRITE=True, REPORT=True, MANAGE=False, ADMIN=False
)
await Membership.create(
user=user, organization=organization, acl=acl
)
await Membership.create(user=user, organization=organization, acl=acl)
deleted_org = await client.delete(
f"https://localhost/api/v1/organizations/{organization.id}",