studio/people
Archived
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✅(backend) drop JWT authentication in API tests

Browse Source 
Force login to bypass authorization checks when necessary.

Note: Generating a session cookie through OIDC flow
is not supported while testing our API.
This commit is contained in:
Lebaud Antoine
2024-02-14 23:14:30 +01:00
committed by aleb_the_flash
parent 927d0e5a22
commit ec28c28d47
10 changed files with 368 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/backend/core/tests/teams/test_core_api_teams_create.py
View File

@@ -11,7 +11,6 @@ from rest_framework.test import APIClient
from core.factories import IdentityFactory, TeamFactory
from core.models import Team
from core.tests.utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -36,15 +35,16 @@ def test_api_teams_create_authenticated():
"""
identity = IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
response = APIClient().post(
client = APIClient()
client.force_login(identity.user)
response = client.post(
"/api/v1.0/teams/",
{
"name": "my team",
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_201_CREATED
@@ -58,12 +58,12 @@ def test_api_teams_create_authenticated_slugify_name():
Creating teams should automatically generate a slug.
"""
identity = IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
client = APIClient()
client.force_login(identity.user)
response = APIClient().post(
response = client.post(
"/api/v1.0/teams/",
{"name": "my team"},
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_201_CREATED
@@ -87,14 +87,15 @@ def test_api_teams_create_authenticated_expected_slug(param):
Creating teams should automatically create unaccented, no unicode, lower-case slug.
"""
identity = IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
response = APIClient().post(
client = APIClient()
client.force_login(identity.user)
response = client.post(
"/api/v1.0/teams/",
{
"name": param[0],
},
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_201_CREATED
@@ -109,14 +110,15 @@ def test_api_teams_create_authenticated_unique_slugs():
"""
TeamFactory(name="existing team")
identity = IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
response = APIClient().post(
client = APIClient()
client.force_login(identity.user)
response = client.post(
"/api/v1.0/teams/",
{
"name": "èxisting team",
},
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_400_BAD_REQUEST

34
src/backend/core/tests/teams/test_core_api_teams_delete.py
View File

@@ -11,7 +11,6 @@ from rest_framework.status import (
from rest_framework.test import APIClient
from core import factories, models
from core.tests.utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -34,13 +33,14 @@ def test_api_teams_delete_authenticated_unrelated():
related.
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(identity.user)
team = factories.TeamFactory()
response = APIClient().delete(
f"/api/v1.0/teams/{team.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/",
)
assert response.status_code == HTTP_404_NOT_FOUND
@@ -55,12 +55,14 @@ def test_api_teams_delete_authenticated_member():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
response = APIClient().delete(
f"/api/v1.0/teams/{team.id}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.delete(
f"/api/v1.0/teams/{team.id}/",
)
assert response.status_code == HTTP_403_FORBIDDEN
@@ -77,12 +79,14 @@ def test_api_teams_delete_authenticated_administrator():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
response = APIClient().delete(
f"/api/v1.0/teams/{team.id}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.delete(
f"/api/v1.0/teams/{team.id}/",
)
assert response.status_code == HTTP_403_FORBIDDEN
@@ -99,12 +103,14 @@ def test_api_teams_delete_authenticated_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
response = APIClient().delete(
f"/api/v1.0/teams/{team.id}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.delete(
f"/api/v1.0/teams/{team.id}/",
)
assert response.status_code == HTTP_204_NO_CONTENT

47
src/backend/core/tests/teams/test_core_api_teams_list.py
View File

@@ -10,8 +10,6 @@ from rest_framework.test import APIClient
from core import factories
from ..utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -28,10 +26,13 @@ def test_api_teams_list_anonymous():
def test_api_teams_list_authenticated():
"""Authenticated users should be able to list teams they are an owner/administrator/member of."""
"""Authenticated users should be able to list teams
they arean owner/administrator/member of."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
expected_ids = {
str(access.team.id)
@@ -39,8 +40,8 @@ def test_api_teams_list_authenticated():
}
factories.TeamFactory.create_batch(2) # Other teams
response = APIClient().get(
"/api/v1.0/teams/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/teams/",
)
assert response.status_code == HTTP_200_OK
@@ -57,7 +58,9 @@ def test_api_teams_list_pagination(
"""Pagination should work as expected."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team_ids = [
str(access.team.id)
@@ -65,8 +68,8 @@ def test_api_teams_list_pagination(
]
# Get page 1
response = APIClient().get(
"/api/v1.0/teams/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/teams/",
)
assert response.status_code == HTTP_200_OK
@@ -81,8 +84,8 @@ def test_api_teams_list_pagination(
team_ids.remove(item["id"])
# Get page 2
response = APIClient().get(
"/api/v1.0/teams/?page=2", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/teams/?page=2",
)
assert response.status_code == HTTP_200_OK
@@ -101,14 +104,16 @@ def test_api_teams_list_authenticated_distinct():
"""A team with several related users should only be listed once."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
other_user = factories.UserFactory()
team = factories.TeamFactory(users=[user, other_user])
response = APIClient().get(
"/api/v1.0/teams/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/teams/",
)
assert response.status_code == HTTP_200_OK
@@ -123,15 +128,16 @@ def test_api_teams_order():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team_ids = [
str(team.id) for team in factories.TeamFactory.create_batch(5, users=[user])
]
response = APIClient().get(
response = client.get(
"/api/v1.0/teams/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
@@ -152,15 +158,16 @@ def test_api_teams_order_param():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team_ids = [
str(team.id) for team in factories.TeamFactory.create_batch(5, users=[user])
]
response = APIClient().get(
response = client.get(
"/api/v1.0/teams/?ordering=created_at",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200

18
src/backend/core/tests/teams/test_core_api_teams_retrieve.py
View File

@@ -6,7 +6,6 @@ from rest_framework.status import HTTP_200_OK, HTTP_401_UNAUTHORIZED, HTTP_404_N
from rest_framework.test import APIClient
from core import factories
from core.tests.utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -28,13 +27,14 @@ def test_api_teams_retrieve_authenticated_unrelated():
not related.
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(identity.user)
team = factories.TeamFactory()
response = APIClient().get(
f"/api/v1.0/teams/{team.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
f"/api/v1.0/teams/{team.id!s}/",
)
assert response.status_code == HTTP_404_NOT_FOUND
assert response.json() == {"detail": "Not found."}
@@ -47,14 +47,16 @@ def test_api_teams_retrieve_authenticated_related():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access1 = factories.TeamAccessFactory(team=team, user=user)
access2 = factories.TeamAccessFactory(team=team)
response = APIClient().get(
f"/api/v1.0/teams/{team.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
f"/api/v1.0/teams/{team.id!s}/",
)
assert response.status_code == HTTP_200_OK
content = response.json()

44
src/backend/core/tests/teams/test_core_api_teams_update.py
View File

@@ -15,7 +15,6 @@ from rest_framework.test import APIClient
from core import factories
from core.api import serializers
from core.tests.utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -46,18 +45,18 @@ def test_api_teams_update_authenticated_unrelated():
Authenticated users should not be allowed to update a team to which they are not related.
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(identity.user)
team = factories.TeamFactory()
old_team_values = serializers.TeamSerializer(instance=team).data
new_team_values = serializers.TeamSerializer(instance=factories.TeamFactory()).data
response = APIClient().put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/",
new_team_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_404_NOT_FOUND
@@ -75,17 +74,18 @@ def test_api_teams_update_authenticated_members():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
old_team_values = serializers.TeamSerializer(instance=team).data
new_team_values = serializers.TeamSerializer(instance=factories.TeamFactory()).data
response = APIClient().put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/",
new_team_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_403_FORBIDDEN
@@ -102,18 +102,19 @@ def test_api_teams_update_authenticated_administrators():
"""Administrators of a team should be allowed to update it."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
initial_values = serializers.TeamSerializer(instance=team).data
# generate new random values
new_values = serializers.TeamSerializer(instance=factories.TeamFactory.build()).data
response = APIClient().put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/",
new_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -132,7 +133,9 @@ def test_api_teams_update_authenticated_owners():
apart from read-only fields."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
old_team_values = serializers.TeamSerializer(instance=team).data
@@ -140,11 +143,10 @@ def test_api_teams_update_authenticated_owners():
new_team_values = serializers.TeamSerializer(
instance=factories.TeamFactory.build()
).data
response = APIClient().put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/",
new_team_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -165,18 +167,19 @@ def test_api_teams_update_administrator_or_owner_of_another():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
factories.TeamFactory(users=[(user, random.choice(["administrator", "owner"]))])
team = factories.TeamFactory(name="Old name")
old_team_values = serializers.TeamSerializer(instance=team).data
new_team_values = serializers.TeamSerializer(instance=factories.TeamFactory()).data
response = APIClient().put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/",
new_team_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_404_NOT_FOUND
@@ -194,7 +197,9 @@ def test_api_teams_update_existing_slug_should_return_error():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
factories.TeamFactory(name="Existing team", users=[(user, "administrator")])
my_team = factories.TeamFactory(name="New team", users=[(user, "administrator")])
@@ -202,11 +207,10 @@ def test_api_teams_update_existing_slug_should_return_error():
updated_values = serializers.TeamSerializer(instance=my_team).data
# Update my team's name for existing team. Creates a duplicate slug
updated_values["name"] = "existing team"
response = APIClient().put(
response = client.put(
f"/api/v1.0/teams/{my_team.id!s}/",
updated_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_400_BAD_REQUEST
assert response.json()["slug"] == ["Team with this Slug already exists."]

187
src/backend/core/tests/test_api_contacts.py
View File

@@ -9,8 +9,6 @@ from rest_framework.test import APIClient
from core import factories, models
from core.api import serializers
from .utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -75,7 +73,6 @@ def test_api_contacts_list_authenticated_no_query():
contact = factories.ContactFactory(owner=user)
user.profile_contact = contact
user.save()
jwt_token = OIDCToken.for_user(user)
# Let's have 5 contacts in database:
assert user.profile_contact is not None # Excluded because profile contact
@@ -87,9 +84,10 @@ def test_api_contacts_list_authenticated_no_query():
base=base_contact, owner=user, full_name="Bernard"
) # Included
response = APIClient().get(
"/api/v1.0/contacts/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/contacts/")
assert response.status_code == 200
assert response.json() == [
@@ -111,7 +109,6 @@ def test_api_contacts_list_authenticated_by_full_name():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
dave = factories.BaseContactFactory(full_name="David Bowman")
nicole = factories.BaseContactFactory(full_name="Nicole Foole")
@@ -119,36 +116,31 @@ def test_api_contacts_list_authenticated_by_full_name():
factories.BaseContactFactory(full_name="Heywood Floyd")
# Full query should work
response = APIClient().get(
"/api/v1.0/contacts/?q=David%20Bowman", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/contacts/?q=David%20Bowman")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Partial query should work
response = APIClient().get(
"/api/v1.0/contacts/?q=ank", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=ank")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(frank.id)]
# Result that matches a trigram twice ranks better than result that matches once
response = APIClient().get(
"/api/v1.0/contacts/?q=ole", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=ole")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
# "Nicole Foole" matches twice on "ole"
assert contact_ids == [str(nicole.id), str(frank.id)]
response = APIClient().get(
"/api/v1.0/contacts/?q=ool", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=ool")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
@@ -159,23 +151,21 @@ def test_api_contacts_list_authenticated_uppercase_content():
"""Upper case content should be found by lower case query."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
dave = factories.BaseContactFactory(full_name="EEE", short_name="AAA")
# Unaccented full name
response = APIClient().get(
"/api/v1.0/contacts/?q=eee", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(user)
response = client.get("/api/v1.0/contacts/?q=eee")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = APIClient().get(
"/api/v1.0/contacts/?q=aaa", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=aaa")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
@@ -186,23 +176,21 @@ def test_api_contacts_list_authenticated_capital_query():
"""Upper case query should find lower case content."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
dave = factories.BaseContactFactory(full_name="eee", short_name="aaa")
client = APIClient()
client.force_login(user)
# Unaccented full name
response = APIClient().get(
"/api/v1.0/contacts/?q=EEE", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=EEE")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = APIClient().get(
"/api/v1.0/contacts/?q=AAA", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=AAA")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
@@ -213,23 +201,21 @@ def test_api_contacts_list_authenticated_accented_content():
"""Accented content should be found by unaccented query."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
dave = factories.BaseContactFactory(full_name="ééé", short_name="ààà")
client = APIClient()
client.force_login(user)
# Unaccented full name
response = APIClient().get(
"/api/v1.0/contacts/?q=eee", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=eee")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = APIClient().get(
"/api/v1.0/contacts/?q=aaa", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=aaa")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
@@ -240,23 +226,21 @@ def test_api_contacts_list_authenticated_accented_query():
"""Accented query should find unaccented content."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
dave = factories.BaseContactFactory(full_name="eee", short_name="aaa")
client = APIClient()
client.force_login(user)
# Unaccented full name
response = APIClient().get(
"/api/v1.0/contacts/?q=ééé", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=ééé")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
assert contact_ids == [str(dave.id)]
# Unaccented short name
response = APIClient().get(
"/api/v1.0/contacts/?q=ààà", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/contacts/?q=ààà")
assert response.status_code == 200
contact_ids = [contact["id"] for contact in response.json()]
@@ -281,13 +265,13 @@ def test_api_contacts_retrieve_authenticated_owned():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
contact = factories.ContactFactory(owner=user)
response = APIClient().get(
f"/api/v1.0/contacts/{contact.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(user)
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 200
assert response.json() == {
@@ -305,13 +289,12 @@ def test_api_contacts_retrieve_authenticated_public():
Authenticated users should be able to retrieve public contacts.
"""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
contact = factories.BaseContactFactory()
response = APIClient().get(
f"/api/v1.0/contacts/{contact.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(identity.user)
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 200
assert response.json() == {
"id": str(contact.id),
@@ -328,13 +311,12 @@ def test_api_contacts_retrieve_authenticated_other():
Authenticated users should not be allowed to retrieve another user's contacts.
"""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
contact = factories.ContactFactory()
response = APIClient().get(
f"/api/v1.0/contacts/{contact.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(identity.user)
response = client.get(f"/api/v1.0/contacts/{contact.id!s}/")
assert response.status_code == 403
assert response.json() == {
"detail": "You do not have permission to perform this action."
@@ -357,17 +339,17 @@ def test_api_contacts_create_anonymous_forbidden():
def test_api_contacts_create_authenticated_missing_base():
"""Anonymous users should be able to create users."""
identity = factories.IdentityFactory(user__profile_contact=None)
user = identity.user
jwt_token = OIDCToken.for_user(user)
response = APIClient().post(
client = APIClient()
client.force_login(identity.user)
response = client.post(
"/api/v1.0/contacts/",
{
"full_name": "David Bowman",
"short_name": "Dave",
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 400
assert models.Contact.objects.exists() is False
@@ -379,14 +361,15 @@ def test_api_contacts_create_authenticated_successful():
"""Authenticated users should be able to create contacts."""
identity = factories.IdentityFactory(user__profile_contact=None)
user = identity.user
jwt_token = OIDCToken.for_user(user)
base_contact = factories.BaseContactFactory()
client = APIClient()
client.force_login(user)
# Existing override for another user should not interfere
factories.ContactFactory(base=base_contact)
response = APIClient().post(
response = client.post(
"/api/v1.0/contacts/",
{
"base": str(base_contact.id),
@@ -395,7 +378,6 @@ def test_api_contacts_create_authenticated_successful():
"data": CONTACT_DATA,
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 201
@@ -426,12 +408,14 @@ def test_api_contacts_create_authenticated_existing_override():
"""
identity = factories.IdentityFactory(user__profile_contact=None)
user = identity.user
jwt_token = OIDCToken.for_user(user)
base_contact = factories.BaseContactFactory()
factories.ContactFactory(base=base_contact, owner=user)
response = APIClient().post(
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/contacts/",
{
"base": str(base_contact.id),
@@ -440,7 +424,6 @@ def test_api_contacts_create_authenticated_existing_override():
"data": CONTACT_DATA,
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 400
@@ -481,7 +464,9 @@ def test_api_contacts_update_authenticated_owned():
"""
identity = factories.IdentityFactory(user__profile_contact=None)
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
contact = factories.ContactFactory(owner=user) # Owned by the logged-in user
old_contact_values = serializers.ContactSerializer(instance=contact).data
@@ -491,11 +476,10 @@ def test_api_contacts_update_authenticated_owned():
).data
new_contact_values["base"] = str(factories.ContactFactory().id)
response = APIClient().put(
response = client.put(
f"/api/v1.0/contacts/{contact.id!s}/",
new_contact_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
@@ -515,7 +499,9 @@ def test_api_contacts_update_authenticated_profile():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
contact = factories.ContactFactory(owner=user)
user.profile_contact = contact
@@ -527,11 +513,10 @@ def test_api_contacts_update_authenticated_profile():
).data
new_contact_values["base"] = str(factories.ContactFactory().id)
response = APIClient().put(
response = client.put(
f"/api/v1.0/contacts/{contact.id!s}/",
new_contact_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
@@ -550,7 +535,9 @@ def test_api_contacts_update_authenticated_other():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
contact = factories.ContactFactory() # owned by another user
old_contact_values = serializers.ContactSerializer(instance=contact).data
@@ -560,11 +547,10 @@ def test_api_contacts_update_authenticated_other():
).data
new_contact_values["base"] = str(factories.ContactFactory().id)
response = APIClient().put(
response = client.put(
f"/api/v1.0/contacts/{contact.id!s}/",
new_contact_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -588,13 +574,13 @@ def test_api_contacts_delete_list_authenticated():
"""Authenticated users should not be allowed to delete a list of contacts."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
factories.ContactFactory.create_batch(2)
response = APIClient().delete(
"/api/v1.0/contacts/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.delete("/api/v1.0/contacts/")
assert response.status_code == 405
assert models.Contact.objects.count() == 4
@@ -617,13 +603,14 @@ def test_api_contacts_delete_authenticated_public():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
contact = factories.BaseContactFactory()
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -636,13 +623,13 @@ def test_api_contacts_delete_authenticated_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
contact = factories.ContactFactory(owner=user)
response = APIClient().delete(
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 204
@@ -656,15 +643,15 @@ def test_api_contacts_delete_authenticated_profile():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
contact = factories.ContactFactory(owner=user, base=None)
user.profile_contact = contact
user.save()
response = APIClient().delete(
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 204
@@ -677,13 +664,13 @@ def test_api_contacts_delete_authenticated_other():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
contact = factories.ContactFactory()
response = APIClient().delete(
client = APIClient()
client.force_login(user)
response = client.delete(
f"/api/v1.0/contacts/{contact.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403

179
src/backend/core/tests/test_api_team_accesses.py
View File

@@ -10,8 +10,6 @@ from rest_framework.test import APIClient
from core import factories, models
from core.api import serializers
from .utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -34,18 +32,18 @@ def test_api_team_accesses_list_authenticated_unrelated():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
team = factories.TeamFactory()
factories.TeamAccessFactory.create_batch(3, team=team)
client = APIClient()
client.force_login(user)
# Accesses for other teams to which the user is related should not be listed either
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
response = APIClient().get(
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
assert response.json() == {
@@ -63,7 +61,9 @@ def test_api_team_accesses_list_authenticated_related():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
user_access = models.TeamAccess.objects.create(team=team, user=user) # random role
@@ -73,9 +73,8 @@ def test_api_team_accesses_list_authenticated_related():
other_access = factories.TeamAccessFactory(user=user)
factories.TeamAccessFactory(team=other_access.team)
response = APIClient().get(
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
@@ -129,15 +128,14 @@ def test_api_team_accesses_retrieve_authenticated_unrelated():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team)
response = APIClient().get(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
response = client.get(f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/")
assert response.status_code == 403
assert response.json() == {
"detail": "You do not have permission to perform this action."
@@ -148,9 +146,8 @@ def test_api_team_accesses_retrieve_authenticated_unrelated():
factories.TeamAccessFactory(),
factories.TeamAccessFactory(user=user),
]:
response = APIClient().get(
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 404
@@ -164,14 +161,15 @@ def test_api_team_accesses_retrieve_authenticated_related():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[user])
access = factories.TeamAccessFactory(team=team)
response = APIClient().get(
response = client.get(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
@@ -211,18 +209,19 @@ def test_api_team_accesses_create_authenticated_unrelated():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
other_user = factories.UserFactory()
team = factories.TeamFactory()
response = APIClient().post(
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -236,21 +235,21 @@ def test_api_team_accesses_create_authenticated_member():
"""Members of a team should not be allowed to create team accesses."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
other_user = factories.UserFactory()
api_client = APIClient()
for role in [role[0] for role in models.RoleChoices.choices]:
response = api_client.post(
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -267,22 +266,21 @@ def test_api_team_accesses_create_authenticated_administrator():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
api_client = APIClient()
# It should not be allowed to create an owner access
response = api_client.post(
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": "owner",
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -295,14 +293,13 @@ def test_api_team_accesses_create_authenticated_administrator():
[role[0] for role in models.RoleChoices.choices if role[0] != "owner"]
)
response = api_client.post(
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 201
@@ -322,21 +319,22 @@ def test_api_team_accesses_create_authenticated_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
other_user = factories.UserFactory()
role = random.choice([role[0] for role in models.RoleChoices.choices])
response = APIClient().post(
response = client.post(
f"/api/v1.0/teams/{team.id!s}/accesses/",
{
"user": str(other_user.id),
"role": role,
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 201
@@ -382,7 +380,9 @@ def test_api_team_accesses_update_authenticated_unrelated():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
access = factories.TeamAccessFactory()
old_values = serializers.TeamAccessSerializer(instance=access).data
@@ -393,13 +393,11 @@ def test_api_team_accesses_update_authenticated_unrelated():
"role": random.choice(models.RoleChoices.choices)[0],
}
api_client = APIClient()
for field, value in new_values.items():
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -412,7 +410,9 @@ def test_api_team_accesses_update_authenticated_member():
"""Members of a team should not be allowed to update its accesses."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
access = factories.TeamAccessFactory(team=team)
@@ -424,13 +424,11 @@ def test_api_team_accesses_update_authenticated_member():
"role": random.choice(models.RoleChoices.choices)[0],
}
api_client = APIClient()
for field, value in new_values.items():
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
{**old_values, field: value},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -446,7 +444,9 @@ def test_api_team_accesses_update_administrator_except_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
access = factories.TeamAccessFactory(
@@ -461,14 +461,12 @@ def test_api_team_accesses_update_administrator_except_owner():
"role": random.choice(["administrator", "member"]),
}
api_client = APIClient()
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
if (
@@ -493,7 +491,9 @@ def test_api_team_accesses_update_administrator_from_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
@@ -506,13 +506,11 @@ def test_api_team_accesses_update_administrator_from_owner():
"role": random.choice(models.RoleChoices.choices)[0],
}
api_client = APIClient()
for field, value in new_values.items():
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, field: value},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
access.refresh_from_db()
@@ -527,7 +525,9 @@ def test_api_team_accesses_update_administrator_to_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
other_user = factories.UserFactory()
@@ -544,14 +544,12 @@ def test_api_team_accesses_update_administrator_to_owner():
"role": "owner",
}
api_client = APIClient()
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
# We are not allowed or not really updating the role
if field == "role" or new_data["role"] == old_values["role"]:
@@ -571,7 +569,9 @@ def test_api_team_accesses_update_owner_except_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
factories.UserFactory()
@@ -587,14 +587,12 @@ def test_api_team_accesses_update_owner_except_owner():
"role": random.choice(models.RoleChoices.choices)[0],
}
api_client = APIClient()
for field, value in new_values.items():
new_data = {**old_values, field: value}
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data=new_data,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
if (
@@ -620,7 +618,9 @@ def test_api_team_accesses_update_owner_for_owners():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(team=team, role="owner")
@@ -632,13 +632,11 @@ def test_api_team_accesses_update_owner_for_owners():
"role": random.choice(models.RoleChoices.choices)[0],
}
api_client = APIClient()
for field, value in new_values.items():
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, field: value},
content_type="application/json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
access.refresh_from_db()
@@ -653,19 +651,19 @@ def test_api_team_accesses_update_owner_self():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user, role="owner")
old_values = serializers.TeamAccessSerializer(instance=access).data
new_role = random.choice(["administrator", "member"])
api_client = APIClient()
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, "role": new_role},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -675,11 +673,10 @@ def test_api_team_accesses_update_owner_self():
# Add another owner and it should now work
factories.TeamAccessFactory(team=team, role="owner")
response = api_client.put(
response = client.put(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
data={**old_values, "role": new_role},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 200
@@ -709,13 +706,14 @@ def test_api_team_accesses_delete_authenticated():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
access = factories.TeamAccessFactory()
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/teams/{access.team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -729,7 +727,9 @@ def test_api_team_accesses_delete_member():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "member")])
access = factories.TeamAccessFactory(team=team)
@@ -737,9 +737,8 @@ def test_api_team_accesses_delete_member():
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -753,7 +752,9 @@ def test_api_team_accesses_delete_administrators():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "administrator")])
access = factories.TeamAccessFactory(
@@ -763,9 +764,8 @@ def test_api_team_accesses_delete_administrators():
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 204
@@ -779,7 +779,9 @@ def test_api_team_accesses_delete_owners_except_owners():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(
@@ -789,9 +791,8 @@ def test_api_team_accesses_delete_owners_except_owners():
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 204
@@ -805,7 +806,9 @@ def test_api_team_accesses_delete_owners_for_owners():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory(users=[(user, "owner")])
access = factories.TeamAccessFactory(team=team, role="owner")
@@ -813,9 +816,8 @@ def test_api_team_accesses_delete_owners_for_owners():
assert models.TeamAccess.objects.count() == 2
assert models.TeamAccess.objects.filter(user=access.user).exists()
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403
@@ -828,15 +830,16 @@ def test_api_team_accesses_delete_owners_last_owner():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
team = factories.TeamFactory()
access = factories.TeamAccessFactory(team=team, user=user, role="owner")
assert models.TeamAccess.objects.count() == 1
response = APIClient().delete(
response = client.delete(
f"/api/v1.0/teams/{team.id!s}/accesses/{access.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == 403

167
src/backend/core/tests/test_api_users.py
View File

@@ -16,8 +16,6 @@ from core import factories, models
from core.api import serializers
from core.api.viewsets import Pagination
from .utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -37,11 +35,13 @@ def test_api_users_list_authenticated():
Authenticated users should be able to list all users.
"""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
client = APIClient()
client.force_login(identity.user)
factories.UserFactory.create_batch(2)
response = APIClient().get(
"/api/v1.0/users/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/",
)
assert response.status_code == HTTP_200_OK
assert len(response.json()["results"]) == 3
@@ -54,7 +54,9 @@ def test_api_users_authenticated_list_by_email():
"""
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(email="david.bowman@work.com")
nicole = factories.IdentityFactory(email="nicole_foole@work.com")
@@ -62,9 +64,8 @@ def test_api_users_authenticated_list_by_email():
factories.IdentityFactory(email="heywood_floyd@work.com")
# Full query should work
response = APIClient().get(
response = client.get(
"/api/v1.0/users/?q=david.bowman@work.com",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -72,18 +73,14 @@ def test_api_users_authenticated_list_by_email():
assert user_ids[0] == str(dave.user.id)
# Partial query should work
response = APIClient().get(
"/api/v1.0/users/?q=fran", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/users/?q=fran")
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
assert user_ids[0] == str(frank.user.id)
# Result that matches a trigram twice ranks better than result that matches once
response = APIClient().get(
"/api/v1.0/users/?q=ole", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/users/?q=ole")
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
@@ -91,9 +88,7 @@ def test_api_users_authenticated_list_by_email():
assert user_ids == [str(nicole.user.id), str(frank.user.id)]
# Even with a low similarity threshold, query should match expected emails
response = APIClient().get(
"/api/v1.0/users/?q=ool", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
response = client.get("/api/v1.0/users/?q=ool")
assert response.status_code == HTTP_200_OK
user_ids = [user["id"] for user in response.json()["results"]]
@@ -106,16 +101,17 @@ def test_api_users_authenticated_list_multiple_identities_single_user():
"""
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
dave = factories.UserFactory()
factories.IdentityFactory(user=dave, email="david.bowman@work.com")
factories.IdentityFactory(user=dave, email="david.bowman@fun.fr")
# Full query should work
response = APIClient().get(
response = client.get(
"/api/v1.0/users/?q=david.bowman@work.com",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -131,7 +127,9 @@ def test_api_users_authenticated_list_multiple_identities_multiple_users():
"""
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
dave = factories.UserFactory()
davina = factories.UserFactory()
@@ -142,9 +140,8 @@ def test_api_users_authenticated_list_multiple_identities_multiple_users():
factories.IdentityFactory(user=prudence, email="prudence.crandall@work.com")
# Full query should work
response = APIClient().get(
response = client.get(
"/api/v1.0/users/?q=david.bowman@work.com",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -157,14 +154,15 @@ def test_api_users_authenticated_list_uppercase_content():
"""Upper case content should be found by lower case query."""
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(email="DAVID.BOWMAN@INTENSEWORK.COM")
# Unaccented full address
response = APIClient().get(
response = client.get(
"/api/v1.0/users/?q=david.bowman@intensework.com",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -172,8 +170,8 @@ def test_api_users_authenticated_list_uppercase_content():
assert user_ids == [str(dave.user.id)]
# Partial query
response = APIClient().get(
"/api/v1.0/users/?q=david", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/?q=david",
)
assert response.status_code == HTTP_200_OK
@@ -185,14 +183,15 @@ def test_api_users_list_authenticated_capital_query():
"""Upper case query should find lower case content."""
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
dave = factories.IdentityFactory(email="david.bowman@work.com")
# Full uppercase query
response = APIClient().get(
response = client.get(
"/api/v1.0/users/?q=DAVID.BOWMAN@WORK.COM",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -200,8 +199,8 @@ def test_api_users_list_authenticated_capital_query():
assert user_ids == [str(dave.user.id)]
# Partial uppercase email
response = APIClient().get(
"/api/v1.0/users/?q=DAVID", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/?q=DAVID",
)
assert response.status_code == HTTP_200_OK
@@ -213,14 +212,15 @@ def test_api_contacts_list_authenticated_accented_query():
"""Accented content should be found by unaccented query."""
user = factories.UserFactory(email="tester@ministry.fr")
factories.IdentityFactory(user=user, email=user.email)
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
helene = factories.IdentityFactory(email="helene.bowman@work.com")
# Accented full query
response = APIClient().get(
response = client.get(
"/api/v1.0/users/?q=hélène.bowman@work.com",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -228,8 +228,8 @@ def test_api_contacts_list_authenticated_accented_query():
assert user_ids == [str(helene.user.id)]
# Unaccented partial email
response = APIClient().get(
"/api/v1.0/users/?q=hélène", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/?q=hélène",
)
assert response.status_code == HTTP_200_OK
@@ -244,13 +244,15 @@ def test_api_users_list_pagination(
"""Pagination should work as expected."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
factories.UserFactory.create_batch(4)
# Get page 1
response = APIClient().get(
"/api/v1.0/users/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/",
)
assert response.status_code == HTTP_200_OK
@@ -262,8 +264,8 @@ def test_api_users_list_pagination(
assert content["previous"] is None
# Get page 2
response = APIClient().get(
"/api/v1.0/users/?page=2", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/?page=2",
)
assert response.status_code == HTTP_200_OK
@@ -291,7 +293,9 @@ def test_api_users_retrieve_me_authenticated():
"""Authenticated users should be able to retrieve their own user via the "/users/me" path."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
# Define profile contact
contact = factories.ContactFactory(owner=user)
@@ -299,8 +303,8 @@ def test_api_users_retrieve_me_authenticated():
user.save()
factories.UserFactory.create_batch(2)
response = APIClient().get(
"/api/v1.0/users/me/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
"/api/v1.0/users/me/",
)
assert response.status_code == HTTP_200_OK
@@ -333,10 +337,12 @@ def test_api_users_retrieve_authenticated_self():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
response = APIClient().get(
f"/api/v1.0/users/{user.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
client = APIClient()
client.force_login(user)
response = client.get(
f"/api/v1.0/users/{user.id!s}/",
)
assert response.status_code == HTTP_405_METHOD_NOT_ALLOWED
assert response.json() == {"detail": 'Method "GET" not allowed.'}
@@ -348,12 +354,14 @@ def test_api_users_retrieve_authenticated_other():
limited information.
"""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
client = APIClient()
client.force_login(identity.user)
other_user = factories.UserFactory()
response = APIClient().get(
f"/api/v1.0/users/{other_user.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
response = client.get(
f"/api/v1.0/users/{other_user.id!s}/",
)
assert response.status_code == HTTP_405_METHOD_NOT_ALLOWED
assert response.json() == {"detail": 'Method "GET" not allowed.'}
@@ -379,16 +387,17 @@ def test_api_users_create_authenticated():
"""Authenticated users should not be able to create users via the API."""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
response = APIClient().post(
client = APIClient()
client.force_login(user)
response = client.post(
"/api/v1.0/users/",
{
"language": "fr-fr",
"password": "mypassword",
},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_405_METHOD_NOT_ALLOWED
assert response.json() == {"detail": 'Method "POST" not allowed.'}
@@ -426,18 +435,19 @@ def test_api_users_update_authenticated_self():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
old_user_values = dict(serializers.UserSerializer(instance=user).data)
new_user_values = dict(
serializers.UserSerializer(instance=factories.UserFactory()).data
)
response = APIClient().put(
response = client.put(
f"/api/v1.0/users/{user.id!s}/",
new_user_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -453,17 +463,18 @@ def test_api_users_update_authenticated_self():
def test_api_users_update_authenticated_other():
"""Authenticated users should not be allowed to update other users."""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
client = APIClient()
client.force_login(identity.user)
user = factories.UserFactory()
old_user_values = dict(serializers.UserSerializer(instance=user).data)
new_user_values = serializers.UserSerializer(instance=factories.UserFactory()).data
response = APIClient().put(
response = client.put(
f"/api/v1.0/users/{user.id!s}/",
new_user_values,
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_403_FORBIDDEN
@@ -506,7 +517,9 @@ def test_api_users_patch_authenticated_self():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
old_user_values = dict(serializers.UserSerializer(instance=user).data)
new_user_values = dict(
@@ -514,11 +527,10 @@ def test_api_users_patch_authenticated_self():
)
for key, new_value in new_user_values.items():
response = APIClient().patch(
response = client.patch(
f"/api/v1.0/users/{user.id!s}/",
{key: new_value},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_200_OK
@@ -534,7 +546,9 @@ def test_api_users_patch_authenticated_self():
def test_api_users_patch_authenticated_other():
"""Authenticated users should not be allowed to patch other users."""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
client = APIClient()
client.force_login(identity.user)
user = factories.UserFactory()
old_user_values = dict(serializers.UserSerializer(instance=user).data)
@@ -543,11 +557,10 @@ def test_api_users_patch_authenticated_other():
)
for key, new_value in new_user_values.items():
response = APIClient().put(
response = client.put(
f"/api/v1.0/users/{user.id!s}/",
{key: new_value},
format="json",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_403_FORBIDDEN
@@ -572,11 +585,12 @@ def test_api_users_delete_list_authenticated():
"""Authenticated users should not be allowed to delete a list of users."""
factories.UserFactory.create_batch(2)
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
client = APIClient()
client.force_login(identity.user)
response = client.delete(
"/api/v1.0/users/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
"/api/v1.0/users/",
)
assert response.status_code == HTTP_405_METHOD_NOT_ALLOWED
@@ -598,12 +612,12 @@ def test_api_users_delete_authenticated():
Authenticated users should not be allowed to delete a user other than themselves.
"""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
other_user = factories.UserFactory()
response = APIClient().delete(
f"/api/v1.0/users/{other_user.id!s}/", HTTP_AUTHORIZATION=f"Bearer {jwt_token}"
)
client = APIClient()
client.force_login(identity.user)
response = client.delete(f"/api/v1.0/users/{other_user.id!s}/")
assert response.status_code == HTTP_405_METHOD_NOT_ALLOWED
assert models.User.objects.count() == 2
@@ -612,11 +626,12 @@ def test_api_users_delete_authenticated():
def test_api_users_delete_self():
"""Authenticated users should not be able to delete their own user."""
identity = factories.IdentityFactory()
jwt_token = OIDCToken.for_user(identity.user)
response = APIClient().delete(
client = APIClient()
client.force_login(identity.user)
response = client.delete(
f"/api/v1.0/users/{identity.user.id!s}/",
HTTP_AUTHORIZATION=f"Bearer {jwt_token}",
)
assert response.status_code == HTTP_405_METHOD_NOT_ALLOWED

8
src/backend/core/tests/test_throttle.py
View File

@@ -8,8 +8,6 @@ from core import factories
from people.settings import REST_FRAMEWORK # pylint: disable=E0611
from .utils import OIDCToken
pytestmark = pytest.mark.django_db
@@ -19,9 +17,9 @@ def test_throttle():
"""
identity = factories.IdentityFactory()
user = identity.user
jwt_token = OIDCToken.for_user(user)
client = APIClient()
client.force_login(user)
endpoint = "/api/v1.0/users/"
# loop to activate throttle protection
@@ -29,8 +27,8 @@ def test_throttle():
REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"]["burst"].replace("/minute", "")
)
for _ in range(0, throttle_limit):
client.get(endpoint, HTTP_AUTHORIZATION=f"Bearer {jwt_token}")
client.get(endpoint)
# this call should err
response = client.get(endpoint, HTTP_AUTHORIZATION=f"Bearer {jwt_token}")
response = client.get(endpoint)
assert response.status_code == 429 # too many requests

25
src/backend/core/tests/utils.py
View File

@@ -1,25 +0,0 @@
"""Utils for tests in the People core application"""
from rest_framework_simplejwt.tokens import AccessToken
class OIDCToken(AccessToken):
"""Set payload on token from user/contact/email"""
@classmethod
def for_user(cls, user):
"""Returns an authorization token for the given user for testing."""
identity = user.identities.filter(is_main=True).first()
token = cls()
token["first_name"] = (
user.profile_contact.short_name if user.profile_contact else "David"
)
token["last_name"] = (
" ".join(user.profile_contact.full_name.split()[1:])
if user.profile_contact
else "Bowman"
)
token["sub"] = str(identity.sub)
token["email"] = user.email
return token