|
1
|
|
|
from rest_framework import status |
|
2
|
|
|
from rest_framework.test import APITestCase, force_authenticate |
|
3
|
|
|
|
|
4
|
|
|
from sigma_core.models.group import Group |
|
5
|
|
|
from sigma_core.models.cluster import Cluster |
|
6
|
|
|
from sigma_core.serializers.group import GroupSerializer |
|
7
|
|
|
from sigma_core.serializers.cluster import ClusterSerializer |
|
8
|
|
|
from sigma_core.tests.factories import UserFactory, GroupFactory, GroupMemberFactory, ClusterFactory |
|
9
|
|
|
|
|
10
|
|
|
|
|
11
|
|
|
def reload(obj): |
|
12
|
|
|
return obj.__class__.objects.get(pk=obj.pk) |
|
13
|
|
|
|
|
14
|
|
|
|
|
15
|
|
|
class ClusterTests(APITestCase): |
|
16
|
|
|
@classmethod |
|
17
|
|
|
def setUpTestData(self): |
|
18
|
|
|
super().setUpTestData() |
|
19
|
|
|
|
|
20
|
|
|
# Clusters |
|
21
|
|
|
self.clusters = ClusterFactory.create_batch(2) |
|
22
|
|
|
|
|
23
|
|
|
# Users |
|
24
|
|
|
self.users = UserFactory.create_batch(4) |
|
25
|
|
|
self.users[2].is_staff = True # Sigma admin |
|
26
|
|
|
self.users[2].save() |
|
27
|
|
|
|
|
28
|
|
|
# Memberships |
|
29
|
|
|
self.member1 = GroupMemberFactory(user=self.users[0], group=self.clusters[0], perm_rank=Group.ADMINISTRATOR_RANK) |
|
30
|
|
|
self.member2 = GroupMemberFactory(user=self.users[1], group=self.clusters[0], perm_rank=1) |
|
31
|
|
|
|
|
32
|
|
|
serializer = ClusterSerializer(self.clusters[0]) |
|
33
|
|
|
self.cluster_data = serializer.data |
|
34
|
|
|
self.clusters_url = "/cluster/" |
|
35
|
|
|
self.cluster_url = self.clusters_url + "%d/" |
|
36
|
|
|
|
|
37
|
|
|
self.new_cluster_data = {"name": "Ecole polytechnique", "design": "default"} |
|
38
|
|
|
# self.invite_data = {"user": self.users[0].id} |
|
39
|
|
|
|
|
40
|
|
|
#### List requests |
|
41
|
|
|
def test_get_clusters_list_unauthed(self): |
|
42
|
|
|
# Client not authenticated but can see clusters list |
|
43
|
|
|
response = self.client.get(self.clusters_url) |
|
44
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
45
|
|
|
self.assertEqual(len(response.data), len(self.clusters)) |
|
46
|
|
|
|
|
47
|
|
|
def test_get_clusters_list_ok(self): |
|
48
|
|
|
self.client.force_authenticate(user=self.users[0]) |
|
49
|
|
|
response = self.client.get(self.clusters_url) |
|
50
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
51
|
|
|
self.assertEqual(len(response.data), len(self.clusters)) |
|
52
|
|
|
|
|
53
|
|
|
#### Get requests |
|
54
|
|
|
def test_get_cluster_unauthed(self): |
|
55
|
|
|
# Client is not authenticated, can see cluster data but cannot see cluster details (especially members) |
|
56
|
|
|
response = self.client.get(self.cluster_url % self.clusters[0].id) |
|
57
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
58
|
|
|
self.assertEqual(len(response.data['users']), 0) |
|
59
|
|
|
|
|
60
|
|
|
def test_get_cluster_forbidden(self): |
|
61
|
|
|
# Client wants to see a cluster whose he is not member of |
|
62
|
|
|
self.client.force_authenticate(user=self.users[0]) |
|
63
|
|
|
response = self.client.get(self.cluster_url % self.clusters[1].id) |
|
64
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
65
|
|
|
self.assertEqual(len(response.data['users']), 0) |
|
66
|
|
|
|
|
67
|
|
|
def test_get_cluster_ok(self): |
|
68
|
|
|
# Client wants to see a cluster to which he belongs |
|
69
|
|
|
self.client.force_authenticate(user=self.users[1]) |
|
70
|
|
|
response = self.client.get(self.cluster_url % self.clusters[0].id) |
|
71
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
72
|
|
|
self.assertEqual(response.data, self.cluster_data) |
|
73
|
|
|
|
|
74
|
|
|
#### Create requests |
|
75
|
|
|
def test_create_cluster_unauthed(self): |
|
76
|
|
|
response = self.client.post(self.clusters_url, self.new_cluster_data) |
|
77
|
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
|
78
|
|
|
|
|
79
|
|
|
def test_create_cluster_forbidden(self): |
|
80
|
|
|
self.client.force_authenticate(user=self.users[0]) |
|
81
|
|
|
response = self.client.post(self.clusters_url, self.new_cluster_data) |
|
82
|
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) |
|
83
|
|
|
|
|
84
|
|
|
def test_create_cluster_wrong_data(self): |
|
85
|
|
|
self.client.force_authenticate(user=self.users[2]) |
|
86
|
|
|
response = self.client.post(self.clusters_url, {"name": ""}) |
|
87
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) |
|
88
|
|
|
|
|
89
|
|
|
def test_create_cluster_ok(self): |
|
90
|
|
|
self.client.force_authenticate(user=self.users[2]) |
|
91
|
|
|
response = self.client.post(self.clusters_url, self.new_cluster_data) |
|
92
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED) |
|
93
|
|
|
self.assertEqual(response.data['name'], "Ecole polytechnique") |
|
94
|
|
|
self.assertEqual(response.data['private'], False) |
|
95
|
|
|
self.assertEqual(response.data['default_member_rank'], -1) |
|
96
|
|
|
self.assertEqual(response.data['req_rank_invite'], Group.ADMINISTRATOR_RANK) |
|
97
|
|
|
|
|
98
|
|
|
#### Modification requests |
|
99
|
|
|
def test_update_cluster_unauthed(self): |
|
100
|
|
|
self.cluster_data['name'] = "Ecole polytechnique" |
|
101
|
|
|
response = self.client.put(self.cluster_url % self.cluster_data['id'], self.cluster_data) |
|
102
|
|
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED) |
|
103
|
|
|
|
|
104
|
|
|
def test_update_cluster_forbidden_1(self): |
|
105
|
|
|
self.client.force_authenticate(user=self.users[3]) |
|
106
|
|
|
self.cluster_data['name'] = "Ecole polytechnique" |
|
107
|
|
|
response = self.client.put(self.cluster_url % self.cluster_data['id'], self.cluster_data) |
|
108
|
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) |
|
109
|
|
|
|
|
110
|
|
|
def test_update_cluster_forbidden_2(self): |
|
111
|
|
|
self.client.force_authenticate(user=self.users[1]) |
|
112
|
|
|
self.cluster_data['name'] = "Ecole polytechnique" |
|
113
|
|
|
response = self.client.put(self.cluster_url % self.cluster_data['id'], self.cluster_data) |
|
114
|
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) |
|
115
|
|
|
|
|
116
|
|
|
def test_update_cluster_wrong_data(self): |
|
117
|
|
|
self.client.force_authenticate(user=self.users[2]) |
|
118
|
|
|
self.cluster_data['name'] = "" |
|
119
|
|
|
response = self.client.put(self.cluster_url % self.cluster_data['id'], self.cluster_data) |
|
120
|
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) |
|
121
|
|
|
|
|
122
|
|
|
def test_update_cluster_ok_staff(self): |
|
|
|
|
|
|
123
|
|
|
self.client.force_authenticate(user=self.users[2]) |
|
124
|
|
|
self.cluster_data['name'] = "Ecole polytechnique" |
|
125
|
|
|
response = self.client.put(self.cluster_url % self.cluster_data['id'], self.cluster_data) |
|
126
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
127
|
|
|
self.assertEqual(response.data['name'], "Ecole polytechnique") |
|
128
|
|
|
|
|
129
|
|
|
def test_update_cluster_ok_cluster_admin(self): |
|
|
|
|
|
|
130
|
|
|
self.client.force_authenticate(user=self.users[0]) |
|
131
|
|
|
self.cluster_data['name'] = "Ecole polytechnique" |
|
132
|
|
|
response = self.client.put(self.cluster_url % self.cluster_data['id'], self.cluster_data) |
|
133
|
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK) |
|
134
|
|
|
self.assertEqual(response.data['name'], "Ecole polytechnique") |
|
135
|
|
|
|
|
136
|
|
|
#### Invitation process |
|
137
|
|
|
|
|
138
|
|
|
#### Deletion requests |
|
139
|
|
|
|
Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation.
You can also find more detailed suggestions in the “Code” section of your repository.