import base64
import json
from unittest import mock
from allauth.mfa.models import Authenticator
from allauth.mfa.totp.internal import auth as totp_auth
from django.contrib.auth.models import Group
from django.contrib.auth.models import Permission
from django.contrib.auth.models import User
from guardian.shortcuts import assign_perm
from guardian.shortcuts import get_perms
from guardian.shortcuts import get_users_with_perms
from rest_framework import status
from rest_framework.test import APITestCase
from documents.models import Correspondent
from documents.models import Document
from documents.models import DocumentType
from documents.models import MatchingModel
from documents.models import StoragePath
from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
class TestApiAuth(DirectoriesMixin, APITestCase):
def test_auth_required(self):
d = Document.objects.create(title="Test")
self.assertEqual(
self.client.get("/api/documents/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get(f"/api/documents/{d.id}/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get(f"/api/documents/{d.id}/download/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get(f"/api/documents/{d.id}/preview/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get(f"/api/documents/{d.id}/thumb/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/tags/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/correspondents/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/document_types/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/logs/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/saved_views/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/search/autocomplete/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/documents/bulk_edit/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/documents/bulk_download/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
self.assertEqual(
self.client.get("/api/documents/selection_data/").status_code,
status.HTTP_401_UNAUTHORIZED,
)
def test_api_version_no_auth(self):
response = self.client.get("/api/documents/")
self.assertNotIn("X-Api-Version", response)
self.assertNotIn("X-Version", response)
def test_api_version_with_auth(self):
user = User.objects.create_superuser(username="test")
self.client.force_authenticate(user)
response = self.client.get("/api/documents/")
self.assertIn("X-Api-Version", response)
self.assertIn("X-Version", response)
def test_api_insufficient_permissions(self):
user = User.objects.create_user(username="test")
self.client.force_authenticate(user)
Document.objects.create(title="Test")
self.assertEqual(
self.client.get("/api/documents/").status_code,
status.HTTP_403_FORBIDDEN,
)
self.assertEqual(
self.client.get("/api/tags/").status_code,
status.HTTP_403_FORBIDDEN,
)
self.assertEqual(
self.client.get("/api/correspondents/").status_code,
status.HTTP_403_FORBIDDEN,
)
self.assertEqual(
self.client.get("/api/document_types/").status_code,
status.HTTP_403_FORBIDDEN,
)
self.assertEqual(
self.client.get("/api/logs/").status_code,
status.HTTP_403_FORBIDDEN,
)
self.assertEqual(
self.client.get("/api/saved_views/").status_code,
status.HTTP_403_FORBIDDEN,
)
def test_api_sufficient_permissions(self):
user = User.objects.create_user(username="test")
user.user_permissions.add(*Permission.objects.all())
user.is_staff = True
self.client.force_authenticate(user)
Document.objects.create(title="Test")
self.assertEqual(
self.client.get("/api/documents/").status_code,
status.HTTP_200_OK,
)
self.assertEqual(self.client.get("/api/tags/").status_code, status.HTTP_200_OK)
self.assertEqual(
self.client.get("/api/correspondents/").status_code,
status.HTTP_200_OK,
)
self.assertEqual(
self.client.get("/api/document_types/").status_code,
status.HTTP_200_OK,
)
self.assertEqual(self.client.get("/api/logs/").status_code, status.HTTP_200_OK)
self.assertEqual(
self.client.get("/api/saved_views/").status_code,
status.HTTP_200_OK,
)
def test_api_get_object_permissions(self):
user1 = User.objects.create_user(username="test1")
user2 = User.objects.create_user(username="test2")
user1.user_permissions.add(*Permission.objects.filter(codename="view_document"))
self.client.force_authenticate(user1)
self.assertEqual(
self.client.get("/api/documents/").status_code,
status.HTTP_200_OK,
)
d = Document.objects.create(title="Test", content="the content 1", checksum="1")
# no owner
self.assertEqual(
self.client.get(f"/api/documents/{d.id}/").status_code,
status.HTTP_200_OK,
)
d2 = Document.objects.create(
title="Test 2",
content="the content 2",
checksum="2",
owner=user2,
)
self.assertEqual(
self.client.get(f"/api/documents/{d2.id}/").status_code,
status.HTTP_404_NOT_FOUND,
)
def test_api_default_owner(self):
"""
GIVEN:
- API request to create an object (Tag)
WHEN:
- owner is not set at all
THEN:
- Object created with current user as owner
"""
user1 = User.objects.create_superuser(username="user1")
self.client.force_authenticate(user1)
response = self.client.post(
"/api/tags/",
json.dumps(
{
"name": "test1",
"matching_algorithm": MatchingModel.MATCH_AUTO,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
tag1 = Tag.objects.filter(name="test1").first()
self.assertEqual(tag1.owner, user1)
def test_api_set_no_owner(self):
"""
GIVEN:
- API request to create an object (Tag)
WHEN:
- owner is passed as None
THEN:
- Object created with no owner
"""
user1 = User.objects.create_superuser(username="user1")
self.client.force_authenticate(user1)
response = self.client.post(
"/api/tags/",
json.dumps(
{
"name": "test1",
"matching_algorithm": MatchingModel.MATCH_AUTO,
"owner": None,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
tag1 = Tag.objects.filter(name="test1").first()
self.assertEqual(tag1.owner, None)
def test_api_set_owner_w_permissions(self):
"""
GIVEN:
- API request to create an object (Tag) that supplies set_permissions object
WHEN:
- owner is passed as user id
- view > users is set & view > groups is set
THEN:
- Object permissions are set appropriately
"""
user1 = User.objects.create_superuser(username="user1")
user2 = User.objects.create(username="user2")
group1 = Group.objects.create(name="group1")
self.client.force_authenticate(user1)
response = self.client.post(
"/api/tags/",
json.dumps(
{
"name": "test1",
"matching_algorithm": MatchingModel.MATCH_AUTO,
"owner": user1.id,
"set_permissions": {
"view": {
"users": [user2.id],
"groups": [group1.id],
},
"change": {
"users": None,
"groups": None,
},
},
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
tag1 = Tag.objects.filter(name="test1").first()
from guardian.core import ObjectPermissionChecker
checker = ObjectPermissionChecker(user2)
self.assertEqual(checker.has_perm("view_tag", tag1), True)
self.assertIn("view_tag", get_perms(group1, tag1))
def test_api_set_other_owner_w_permissions(self):
"""
GIVEN:
- API request to create an object (Tag)
WHEN:
- a different owner than is logged in is set
- view > groups is set
THEN:
- Object permissions are set appropriately
"""
user1 = User.objects.create_superuser(username="user1")
user2 = User.objects.create(username="user2")
group1 = Group.objects.create(name="group1")
self.client.force_authenticate(user1)
response = self.client.post(
"/api/tags/",
json.dumps(
{
"name": "test1",
"matching_algorithm": MatchingModel.MATCH_AUTO,
"owner": user2.id,
"set_permissions": {
"view": {
"users": None,
"groups": [group1.id],
},
"change": {
"users": None,
"groups": None,
},
},
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
tag1 = Tag.objects.filter(name="test1").first()
self.assertEqual(tag1.owner, user2)
self.assertIn("view_tag", get_perms(group1, tag1))
def test_api_set_doc_permissions(self):
"""
GIVEN:
- API request to update doc permissions and owner
WHEN:
- owner is set
- view > users is set & view > groups is set
THEN:
- Object permissions are set appropriately
"""
doc = Document.objects.create(
title="test",
mime_type="application/pdf",
content="this is a document",
)
user1 = User.objects.create_superuser(username="user1")
user2 = User.objects.create(username="user2")
group1 = Group.objects.create(name="group1")
self.client.force_authenticate(user1)
response = self.client.patch(
f"/api/documents/{doc.id}/",
json.dumps(
{
"owner": user1.id,
"set_permissions": {
"view": {
"users": [user2.id],
"groups": [group1.id],
},
"change": {
"users": None,
"groups": None,
},
},
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
doc = Document.objects.get(pk=doc.id)
self.assertEqual(doc.owner, user1)
from guardian.core import ObjectPermissionChecker
checker = ObjectPermissionChecker(user2)
self.assertTrue(checker.has_perm("view_document", doc))
self.assertIn("view_document", get_perms(group1, doc))
def test_patch_doesnt_remove_permissions(self):
"""
GIVEN:
- existing document with permissions set
WHEN:
- PATCH API request to update doc that is not json
THEN:
- Object permissions are not removed
"""
doc = Document.objects.create(
title="test",
mime_type="application/pdf",
content="this is a document",
)
user1 = User.objects.create_superuser(username="user1")
user2 = User.objects.create(username="user2")
group1 = Group.objects.create(name="group1")
doc.owner = user1
doc.save()
assign_perm("view_document", user2, doc)
assign_perm("change_document", user2, doc)
assign_perm("view_document", group1, doc)
assign_perm("change_document", group1, doc)
self.client.force_authenticate(user1)
response = self.client.patch(
f"/api/documents/{doc.id}/",
{
"archive_serial_number": "123",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
doc = Document.objects.get(pk=doc.id)
self.assertEqual(doc.owner, user1)
from guardian.core import ObjectPermissionChecker
checker = ObjectPermissionChecker(user2)
self.assertTrue(checker.has_perm("view_document", doc))
self.assertIn("view_document", get_perms(group1, doc))
self.assertTrue(checker.has_perm("change_document", doc))
self.assertIn("change_document", get_perms(group1, doc))
def test_dynamic_permissions_fields(self):
user1 = User.objects.create_user(username="user1")
user1.user_permissions.add(*Permission.objects.filter(codename="view_document"))
user2 = User.objects.create_user(username="user2")
Document.objects.create(title="Test", content="content 1", checksum="1")
doc2 = Document.objects.create(
title="Test2",
content="content 2",
checksum="2",
owner=user2,
)
doc3 = Document.objects.create(
title="Test3",
content="content 3",
checksum="3",
owner=user2,
)
doc4 = Document.objects.create(
title="Test4",
content="content 4",
checksum="4",
owner=user1,
)
assign_perm("view_document", user1, doc2)
assign_perm("view_document", user1, doc3)
assign_perm("change_document", user1, doc3)
assign_perm("view_document", user2, doc4)
self.client.force_authenticate(user1)
response = self.client.get(
"/api/documents/",
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
resp_data = response.json()
# The response will contain the documents in reversed order of creation
# due to #6982, but previously this code relied on implicit ordering
# so let's ensure the order is as expected:
results = resp_data["results"][::-1]
self.assertNotIn("permissions", results[0])
self.assertIn("user_can_change", results[0])
self.assertTrue(results[0]["user_can_change"]) # doc1
self.assertFalse(results[0]["is_shared_by_requester"]) # doc1
self.assertFalse(results[1]["user_can_change"]) # doc2
self.assertTrue(results[2]["user_can_change"]) # doc3
self.assertTrue(results[3]["is_shared_by_requester"]) # doc4
response = self.client.get(
"/api/documents/?full_perms=true",
format="json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
resp_data = response.json()
# See above about response ordering
results = resp_data["results"][::-1]
self.assertIn("permissions", results[0])
self.assertNotIn("user_can_change", results[0])
self.assertNotIn("is_shared_by_requester", results[0])
@mock.patch("allauth.mfa.adapter.DefaultMFAAdapter.is_mfa_enabled")
def test_basic_auth_mfa_enabled(self, mock_is_mfa_enabled):
"""
GIVEN:
- User with MFA enabled
WHEN:
- API request is made with basic auth
THEN:
- MFA required error is returned
"""
user1 = User.objects.create_user(username="user1")
user1.set_password("password")
user1.save()
mock_is_mfa_enabled.return_value = True
response = self.client.get(
"/api/documents/",
HTTP_AUTHORIZATION="Basic " + base64.b64encode(b"user1:password").decode(),
)
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
self.assertEqual(response.data["detail"], "MFA required")
@mock.patch("allauth.mfa.totp.internal.auth.TOTP.validate_code")
def test_get_token_mfa_enabled(self, mock_validate_code):
"""
GIVEN:
- User with MFA enabled
WHEN:
- API request is made to obtain an auth token
THEN:
- MFA code is required
"""
user1 = User.objects.create_user(username="user1")
user1.set_password("password")
user1.save()
response = self.client.post(
"/api/token/",
data={
"username": "user1",
"password": "password",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
secret = totp_auth.generate_totp_secret()
totp_auth.TOTP.activate(
user1,
secret,
)
# no code
response = self.client.post(
"/api/token/",
data={
"username": "user1",
"password": "password",
},
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["non_field_errors"][0], "MFA code is required")
# invalid code
mock_validate_code.return_value = False
response = self.client.post(
"/api/token/",
data={
"username": "user1",
"password": "password",
"code": "123456",
},
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
self.assertEqual(response.data["non_field_errors"][0], "Invalid MFA code")
# valid code
mock_validate_code.return_value = True
response = self.client.post(
"/api/token/",
data={
"username": "user1",
"password": "password",
"code": "123456",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
class TestApiUser(DirectoriesMixin, APITestCase):
ENDPOINT = "/api/users/"
def setUp(self):
super().setUp()
self.user = User.objects.create_superuser(username="temp_admin")
self.client.force_authenticate(user=self.user)
def test_get_users(self):
"""
GIVEN:
- Configured users
WHEN:
- API call is made to get users
THEN:
- Configured users are provided
"""
user1 = User.objects.create(
username="testuser",
password="test",
first_name="Test",
last_name="User",
)
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["count"], 2)
returned_user2 = response.data["results"][1]
self.assertEqual(returned_user2["username"], user1.username)
self.assertEqual(returned_user2["password"], "**********")
self.assertEqual(returned_user2["first_name"], user1.first_name)
self.assertEqual(returned_user2["last_name"], user1.last_name)
def test_create_user(self):
"""
WHEN:
- API request is made to add a user account
THEN:
- A new user account is created
"""
user1 = {
"username": "testuser",
"password": "test",
"first_name": "Test",
"last_name": "User",
}
response = self.client.post(
self.ENDPOINT,
data=user1,
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
returned_user1 = User.objects.get(username="testuser")
self.assertEqual(returned_user1.username, user1["username"])
self.assertEqual(returned_user1.first_name, user1["first_name"])
self.assertEqual(returned_user1.last_name, user1["last_name"])
def test_delete_user(self):
"""
GIVEN:
- Existing user account
WHEN:
- API request is made to delete a user account
THEN:
- Account is deleted
"""
user1 = User.objects.create(
username="testuser",
password="test",
first_name="Test",
last_name="User",
)
nUsers = User.objects.count()
response = self.client.delete(
f"{self.ENDPOINT}{user1.pk}/",
)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(User.objects.count(), nUsers - 1)
def test_update_user(self):
"""
GIVEN:
- Existing user accounts
WHEN:
- API request is made to update user account
THEN:
- The user account is updated, password only updated if not '****'
"""
user1 = User.objects.create(
username="testuser",
password="test",
first_name="Test",
last_name="User",
)
initial_password = user1.password
response = self.client.patch(
f"{self.ENDPOINT}{user1.pk}/",
data={
"first_name": "Updated Name 1",
"password": "******",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
returned_user1 = User.objects.get(pk=user1.pk)
self.assertEqual(returned_user1.first_name, "Updated Name 1")
self.assertEqual(returned_user1.password, initial_password)
response = self.client.patch(
f"{self.ENDPOINT}{user1.pk}/",
data={
"first_name": "Updated Name 2",
"password": "123xyz",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
returned_user2 = User.objects.get(pk=user1.pk)
self.assertEqual(returned_user2.first_name, "Updated Name 2")
self.assertNotEqual(returned_user2.password, initial_password)
def test_deactivate_totp(self):
"""
GIVEN:
- Existing user account with TOTP enabled
WHEN:
- API request by a superuser is made to deactivate TOTP
- API request by a regular user is made to deactivate TOTP
THEN:
- TOTP is deactivated, if exists
- Regular user is forbidden from deactivating TOTP
"""
user1 = User.objects.create(
username="testuser",
password="test",
first_name="Test",
last_name="User",
)
Authenticator.objects.create(
user=user1,
type=Authenticator.Type.TOTP,
data={},
)
response = self.client.post(
f"{self.ENDPOINT}{user1.pk}/deactivate_totp/",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(Authenticator.objects.filter(user=user1).count(), 0)
# fail if already deactivated
response = self.client.post(
f"{self.ENDPOINT}{user1.pk}/deactivate_totp/",
)
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
regular_user = User.objects.create_user(username="regular_user")
regular_user.user_permissions.add(
*Permission.objects.all(),
)
self.client.force_authenticate(regular_user)
Authenticator.objects.create(
user=user1,
type=Authenticator.Type.TOTP,
data={},
)
response = self.client.post(
f"{self.ENDPOINT}{user1.pk}/deactivate_totp/",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
def test_only_superusers_can_create_or_alter_superuser_status(self):
"""
GIVEN:
- Existing user account
WHEN:
- API request is made to add a user account with superuser status
- API request is made to change superuser status
THEN:
- Only superusers can change superuser status
"""
user1 = User.objects.create_user(username="user1")
user1.user_permissions.add(*Permission.objects.all())
user2 = User.objects.create_superuser(username="user2")
self.client.force_authenticate(user1)
response = self.client.patch(
f"{self.ENDPOINT}{user1.pk}/",
json.dumps(
{
"is_superuser": True,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
response = self.client.post(
f"{self.ENDPOINT}",
json.dumps(
{
"username": "user3",
"is_superuser": True,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.client.force_authenticate(user2)
response = self.client.patch(
f"{self.ENDPOINT}{user1.pk}/",
json.dumps(
{
"is_superuser": True,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
returned_user1 = User.objects.get(pk=user1.pk)
self.assertEqual(returned_user1.is_superuser, True)
response = self.client.patch(
f"{self.ENDPOINT}{user1.pk}/",
json.dumps(
{
"is_superuser": False,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
returned_user1 = User.objects.get(pk=user1.pk)
self.assertEqual(returned_user1.is_superuser, False)
class TestApiGroup(DirectoriesMixin, APITestCase):
ENDPOINT = "/api/groups/"
def setUp(self):
super().setUp()
self.user = User.objects.create_superuser(username="temp_admin")
self.client.force_authenticate(user=self.user)
def test_get_groups(self):
"""
GIVEN:
- Configured groups
WHEN:
- API call is made to get groups
THEN:
- Configured groups are provided
"""
group1 = Group.objects.create(
name="Test Group",
)
response = self.client.get(self.ENDPOINT)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data["count"], 1)
returned_group1 = response.data["results"][0]
self.assertEqual(returned_group1["name"], group1.name)
def test_create_group(self):
"""
WHEN:
- API request is made to add a group
THEN:
- A new group is created
"""
group1 = {
"name": "Test Group",
}
response = self.client.post(
self.ENDPOINT,
data=group1,
)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
returned_group1 = Group.objects.get(name="Test Group")
self.assertEqual(returned_group1.name, group1["name"])
def test_delete_group(self):
"""
GIVEN:
- Existing group
WHEN:
- API request is made to delete a group
THEN:
- Group is deleted
"""
group1 = Group.objects.create(
name="Test Group",
)
response = self.client.delete(
f"{self.ENDPOINT}{group1.pk}/",
)
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertEqual(len(Group.objects.all()), 0)
def test_update_group(self):
"""
GIVEN:
- Existing groups
WHEN:
- API request is made to update group
THEN:
- The group is updated
"""
group1 = Group.objects.create(
name="Test Group",
)
response = self.client.patch(
f"{self.ENDPOINT}{group1.pk}/",
data={
"name": "Updated Name 1",
},
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
returned_group1 = Group.objects.get(pk=group1.pk)
self.assertEqual(returned_group1.name, "Updated Name 1")
class TestBulkEditObjectPermissions(APITestCase):
def setUp(self):
super().setUp()
self.temp_admin = User.objects.create_superuser(username="temp_admin")
self.client.force_authenticate(user=self.temp_admin)
self.t1 = Tag.objects.create(name="t1")
self.t2 = Tag.objects.create(name="t2")
self.c1 = Correspondent.objects.create(name="c1")
self.dt1 = DocumentType.objects.create(name="dt1")
self.sp1 = StoragePath.objects.create(name="sp1")
self.user1 = User.objects.create(username="user1")
self.user2 = User.objects.create(username="user2")
self.user3 = User.objects.create(username="user3")
def test_bulk_object_set_permissions(self):
"""
GIVEN:
- Existing objects
WHEN:
- bulk_edit_objects API endpoint is called with set_permissions operation
THEN:
- Permissions and / or owner are changed
"""
permissions = {
"view": {
"users": [self.user1.id, self.user2.id],
"groups": [],
},
"change": {
"users": [self.user1.id],
"groups": [],
},
}
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.t1.id, self.t2.id],
"object_type": "tags",
"operation": "set_permissions",
"permissions": permissions,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn(self.user1, get_users_with_perms(self.t1))
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.c1.id],
"object_type": "correspondents",
"operation": "set_permissions",
"permissions": permissions,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn(self.user1, get_users_with_perms(self.c1))
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.dt1.id],
"object_type": "document_types",
"operation": "set_permissions",
"permissions": permissions,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn(self.user1, get_users_with_perms(self.dt1))
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.sp1.id],
"object_type": "storage_paths",
"operation": "set_permissions",
"permissions": permissions,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn(self.user1, get_users_with_perms(self.sp1))
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.t1.id, self.t2.id],
"object_type": "tags",
"operation": "set_permissions",
"owner": self.user3.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(Tag.objects.get(pk=self.t2.id).owner, self.user3)
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.sp1.id],
"object_type": "storage_paths",
"operation": "set_permissions",
"owner": self.user3.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(StoragePath.objects.get(pk=self.sp1.id).owner, self.user3)
def test_bulk_object_set_permissions_merge(self):
"""
GIVEN:
- Existing objects
WHEN:
- bulk_edit_objects API endpoint is called with set_permissions operation with merge=True or merge=False (default)
THEN:
- Permissions and / or owner are replaced or merged, depending on the merge flag
"""
permissions = {
"view": {
"users": [self.user1.id, self.user2.id],
"groups": [],
},
"change": {
"users": [self.user1.id],
"groups": [],
},
}
assign_perm("view_tag", self.user3, self.t1)
self.t1.owner = self.user3
self.t1.save()
# merge=True
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.t1.id, self.t2.id],
"object_type": "tags",
"owner": self.user1.id,
"permissions": permissions,
"operation": "set_permissions",
"merge": True,
},
),
content_type="application/json",
)
self.t1.refresh_from_db()
self.t2.refresh_from_db()
self.assertEqual(response.status_code, status.HTTP_200_OK)
# user3 should still be owner of t1 since was set prior
self.assertEqual(self.t1.owner, self.user3)
# user1 should now be owner of t2 since it didn't have an owner
self.assertEqual(self.t2.owner, self.user1)
# user1 should be added
self.assertIn(self.user1, get_users_with_perms(self.t1))
# user3 should be preserved
self.assertIn(self.user3, get_users_with_perms(self.t1))
# merge=False (default)
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.t1.id, self.t2.id],
"object_type": "tags",
"permissions": permissions,
"operation": "set_permissions",
"merge": False,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
# user1 should be added
self.assertIn(self.user1, get_users_with_perms(self.t1))
# user3 should be removed
self.assertNotIn(self.user3, get_users_with_perms(self.t1))
def test_bulk_edit_object_permissions_insufficient_perms(self):
"""
GIVEN:
- Objects owned by user other than logged in user
WHEN:
- bulk_edit_objects API endpoint is called with set_permissions operation
THEN:
- User is not able to change permissions
"""
self.t1.owner = User.objects.get(username="temp_admin")
self.t1.save()
self.client.force_authenticate(user=self.user1)
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.t1.id, self.t2.id],
"object_type": "tags",
"operation": "set_permissions",
"owner": self.user1.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
self.assertEqual(response.content, b"Insufficient permissions")
def test_bulk_edit_object_permissions_validation(self):
"""
GIVEN:
- Existing objects
WHEN:
- bulk_edit_objects API endpoint is called with set_permissions operation with invalid params
THEN:
- Validation fails
"""
# not a list
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": self.t1.id,
"object_type": "tags",
"operation": "set_permissions",
"owner": self.user1.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# not a list of ints
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": ["one"],
"object_type": "tags",
"owner": self.user1.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# duplicates
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [self.t1.id, self.t2.id, self.t1.id],
"object_type": "tags",
"operation": "set_permissions",
"owner": self.user1.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
# not a valid object type
response = self.client.post(
"/api/bulk_edit_objects/",
json.dumps(
{
"objects": [1],
"object_type": "madeup",
"operation": "set_permissions",
"owner": self.user1.id,
},
),
content_type="application/json",
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)