18 |
18 |
import random
|
19 |
19 |
import uuid
|
20 |
20 |
|
|
21 |
import pytest
|
21 |
22 |
from django.contrib.contenttypes.models import ContentType
|
22 |
23 |
from django.urls import reverse
|
23 |
24 |
|
24 |
25 |
from authentic2.a2_rbac.models import Role
|
25 |
|
from authentic2.a2_rbac.utils import get_default_ou
|
26 |
26 |
from authentic2.apps.journal.models import Event, EventType
|
27 |
27 |
from authentic2.custom_user.models import User
|
28 |
28 |
from django_rbac.models import SEARCH_OP
|
29 |
29 |
|
30 |
|
from ..utils import basic_authorization_header
|
|
30 |
URL = '/api/users/synchronization/'
|
31 |
31 |
|
32 |
32 |
|
33 |
|
def test_basic(app, simple_user):
|
34 |
|
headers = basic_authorization_header(simple_user)
|
35 |
|
uuids = []
|
36 |
|
for _ in range(100):
|
37 |
|
user = User.objects.create(first_name='ben', last_name='dauve')
|
38 |
|
uuids.append(user.uuid)
|
39 |
|
unknown_uuids = [uuid.uuid4().hex for i in range(100)]
|
40 |
|
url = reverse('a2-api-users-synchronization')
|
|
33 |
@pytest.fixture
|
|
34 |
def user(simple_user):
|
|
35 |
role = Role.objects.get_admin_role(
|
|
36 |
ContentType.objects.get_for_model(User), name='role', slug='role', operation=SEARCH_OP
|
|
37 |
)
|
|
38 |
role.members.add(simple_user)
|
|
39 |
return simple_user
|
|
40 |
|
|
41 |
|
|
42 |
@pytest.fixture
|
|
43 |
def app(app, user):
|
|
44 |
app.set_authorization(('Basic', (user.username, user.username)))
|
|
45 |
return app
|
|
46 |
|
|
47 |
|
|
48 |
@pytest.fixture
|
|
49 |
def users(db):
|
|
50 |
return [User.objects.create(first_name='john', last_name='doe') for _ in range(10)]
|
|
51 |
|
|
52 |
|
|
53 |
@pytest.fixture
|
|
54 |
def uuids(users):
|
|
55 |
return [user.uuid for user in users]
|
|
56 |
|
|
57 |
|
|
58 |
def test_url(app, simple_user):
|
|
59 |
# URL is publikc api, check it
|
|
60 |
assert URL == reverse('a2-api-users-synchronization')
|
|
61 |
|
|
62 |
|
|
63 |
def test_authentication_required(app):
|
|
64 |
app.set_authorization(None)
|
|
65 |
app.post_json(URL, status=401)
|
|
66 |
|
|
67 |
|
|
68 |
def test_permission_required(app, user):
|
|
69 |
user.roles.clear()
|
|
70 |
app.post_json(URL, status=403)
|
|
71 |
|
|
72 |
|
|
73 |
@pytest.fixture(scope='session')
|
|
74 |
def unknown_uuids():
|
|
75 |
return [uuid.uuid4().hex for i in range(10)]
|
|
76 |
|
|
77 |
|
|
78 |
@pytest.fixture
|
|
79 |
def payload(uuids, unknown_uuids):
|
41 |
80 |
content = {
|
42 |
81 |
'known_uuids': uuids + unknown_uuids,
|
43 |
82 |
}
|
44 |
83 |
random.shuffle(content['known_uuids'])
|
|
84 |
return content
|
45 |
85 |
|
46 |
|
# test permission check
|
47 |
|
response = app.post_json(url, params=content, headers=headers, status=403)
|
48 |
|
r = Role.objects.get_admin_role(
|
49 |
|
ContentType.objects.get_for_model(User), name='role', slug='role', operation=SEARCH_OP
|
50 |
|
)
|
51 |
|
r.members.add(simple_user)
|
52 |
|
response = app.post_json(url, params=content, headers=headers)
|
|
86 |
|
|
87 |
def test_basic(app, payload, unknown_uuids):
|
|
88 |
response = app.post_json(URL, params=payload)
|
53 |
89 |
assert response.json['result'] == 1
|
54 |
90 |
assert set(response.json['unknown_uuids']) == set(unknown_uuids)
|
55 |
91 |
|
56 |
92 |
|
57 |
|
def test_full_known_users(app, admin):
|
58 |
|
headers = basic_authorization_header(admin)
|
59 |
|
uuids = []
|
60 |
|
for _ in range(100):
|
61 |
|
user = User.objects.create(first_name='jim', last_name='jam')
|
62 |
|
uuids.append(user.uuid)
|
63 |
|
unknown_uuids = [uuid.uuid4().hex for i in range(100)]
|
64 |
|
url = reverse('a2-api-users-synchronization')
|
65 |
|
content = {
|
66 |
|
'known_uuids': uuids + unknown_uuids,
|
67 |
|
'full_known_users': 1,
|
68 |
|
}
|
69 |
|
random.shuffle(content['known_uuids'])
|
70 |
|
response = app.post_json(url, params=content, headers=headers)
|
|
93 |
def test_full_known_users(app, payload):
|
|
94 |
payload['full_known_users'] = 1
|
|
95 |
response = app.post_json(URL, params=payload)
|
71 |
96 |
assert response.json['result'] == 1
|
72 |
97 |
|
73 |
98 |
# known users returned as part of api's full mode:
|
74 |
|
assert len(response.json['known_users']) == 100
|
|
99 |
assert len(response.json['known_users']) == 10
|
75 |
100 |
for user_dict in response.json['known_users']:
|
76 |
|
assert user_dict['first_name'] == 'jim'
|
77 |
|
assert user_dict['last_name'] == 'jam'
|
|
101 |
assert user_dict['first_name'] == 'john'
|
|
102 |
assert user_dict['last_name'] == 'doe'
|
78 |
103 |
assert {
|
79 |
104 |
'uuid',
|
80 |
105 |
'email',
|
... | ... | |
88 |
113 |
}.issubset(set(user_dict.keys()))
|
89 |
114 |
|
90 |
115 |
|
91 |
|
def test_timestamp(app, admin):
|
92 |
|
headers = basic_authorization_header(admin)
|
93 |
|
url = reverse('a2-api-users-synchronization')
|
|
116 |
def test_timestamp(app, users):
|
94 |
117 |
now = datetime.datetime.now()
|
95 |
|
|
96 |
|
ou = get_default_ou()
|
97 |
|
users = []
|
98 |
|
|
99 |
|
for i in range(6):
|
100 |
|
users.append(
|
101 |
|
User.objects.create(
|
102 |
|
first_name='john%s' % i,
|
103 |
|
last_name='doe',
|
104 |
|
username='user%s' % i,
|
105 |
|
email='user%s' % i,
|
106 |
|
ou=ou,
|
107 |
|
)
|
108 |
|
)
|
|
118 |
users = users[:6]
|
109 |
119 |
|
110 |
120 |
for i, event_name in enumerate(
|
111 |
121 |
[
|
... | ... | |
129 |
139 |
'timestamp': (now - datetime.timedelta(days=3)).isoformat(),
|
130 |
140 |
}
|
131 |
141 |
|
132 |
|
response = app.post(url, params=content, headers=headers)
|
|
142 |
response = app.post(URL, params=content)
|
133 |
143 |
|
134 |
144 |
for user in users[:3]:
|
135 |
145 |
assert user.uuid in response.json['modified_users_uuids']
|
... | ... | |
143 |
153 |
|
144 |
154 |
content['timestamp'] = (now - datetime.timedelta(days=7)).isoformat()
|
145 |
155 |
|
146 |
|
response = app.post(url, params=content, headers=headers)
|
|
156 |
response = app.post(URL, params=content)
|
147 |
157 |
|
148 |
158 |
for user in users[:3]:
|
149 |
159 |
assert user.uuid not in response.json['modified_users_uuids']
|
... | ... | |
155 |
165 |
for user in users[3:]:
|
156 |
166 |
user.delete()
|
157 |
167 |
|
158 |
|
response = app.post(url, params=content, headers=headers)
|
|
168 |
response = app.post(URL, params=content)
|
159 |
169 |
|
160 |
170 |
assert not response.json['modified_users_uuids']
|
161 |
171 |
for user in users:
|
... | ... | |
164 |
174 |
for user in users[:3]:
|
165 |
175 |
content['known_uuids'].remove(user.uuid)
|
166 |
176 |
|
167 |
|
response = app.post(url, params=content, headers=headers)
|
|
177 |
response = app.post(URL, params=content)
|
168 |
178 |
|
169 |
179 |
assert not response.json['modified_users_uuids']
|
170 |
180 |
assert len(response.json['unknown_uuids']) == 3
|
171 |
|
-
|