| 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 |
|
-
|