41 |
41 |
from authentic2.utils.misc import good_next_url
|
42 |
42 |
from django_rbac.models import SEARCH_OP
|
43 |
43 |
|
44 |
|
from .utils import assert_event, basic_authorization_header, get_link_from_mail, login
|
|
44 |
from .utils import (
|
|
45 |
assert_event,
|
|
46 |
basic_authorization_header,
|
|
47 |
basic_authorization_oidc_client,
|
|
48 |
get_link_from_mail,
|
|
49 |
login,
|
|
50 |
)
|
45 |
51 |
|
46 |
52 |
pytestmark = pytest.mark.django_db
|
47 |
53 |
|
... | ... | |
1150 |
1156 |
assert set(response.json['unknown_uuids']) == set(unknown_uuids)
|
1151 |
1157 |
|
1152 |
1158 |
|
|
1159 |
def test_user_synchronization_modification_profile(app):
|
|
1160 |
from authentic2_idp_oidc.models import OIDCAuthorization, OIDCClient
|
|
1161 |
from authentic2_idp_oidc.utils import make_sub
|
|
1162 |
|
|
1163 |
dayafter = datetime.datetime.now() + datetime.timedelta(days=2)
|
|
1164 |
uuids = []
|
|
1165 |
users = []
|
|
1166 |
precreate_dt = datetime.datetime.now()
|
|
1167 |
profile_type = ProfileType.objects.create(name='Referee', slug='referee')
|
|
1168 |
oidc_client = OIDCClient.objects.create(
|
|
1169 |
name='Synchronized client',
|
|
1170 |
slug='synchronized-client',
|
|
1171 |
sector_identifier_uri='https://sync-client.example.org/',
|
|
1172 |
identifier_policy=OIDCClient.POLICY_PAIRWISE_REVERSIBLE,
|
|
1173 |
has_api_access=True,
|
|
1174 |
authorization_mode=OIDCClient.AUTHORIZATION_MODE_BY_SERVICE,
|
|
1175 |
)
|
|
1176 |
headers = basic_authorization_oidc_client(oidc_client)
|
|
1177 |
|
|
1178 |
for i in range(100):
|
|
1179 |
user = User.objects.create(first_name='john', last_name='doe', email='john.doe.%s@ad.dre.ss' % i)
|
|
1180 |
uuids.append(user.uuid)
|
|
1181 |
profile = None
|
|
1182 |
if i % 2:
|
|
1183 |
profile = Profile.objects.create(
|
|
1184 |
profile_type=profile_type,
|
|
1185 |
user=user,
|
|
1186 |
email=f'referee-{i}@ad.dre.ss',
|
|
1187 |
identifier=f'referee-{i}',
|
|
1188 |
data={'foo': i},
|
|
1189 |
)
|
|
1190 |
users.append(
|
|
1191 |
(
|
|
1192 |
user,
|
|
1193 |
profile,
|
|
1194 |
)
|
|
1195 |
)
|
|
1196 |
|
|
1197 |
url = reverse('a2-api-users-synchronization')
|
|
1198 |
|
|
1199 |
# first attempt with no profile information
|
|
1200 |
uuids = [make_sub(oidc_client, user) for user, _ in users]
|
|
1201 |
content = {
|
|
1202 |
'known_uuids': uuids,
|
|
1203 |
}
|
|
1204 |
response = app.post_json(url, params=content, headers=headers)
|
|
1205 |
assert response
|
|
1206 |
assert not response.json['unknown_uuids']
|
|
1207 |
|
|
1208 |
# this time subs with profile info
|
|
1209 |
uuids = [make_sub(oidc_client, user, profile) for user, profile in users]
|
|
1210 |
content = {
|
|
1211 |
'known_uuids': uuids,
|
|
1212 |
}
|
|
1213 |
response = app.post_json(url, params=content, headers=headers)
|
|
1214 |
assert response
|
|
1215 |
assert not response.json['unknown_uuids']
|
|
1216 |
|
|
1217 |
response = app.get(
|
|
1218 |
'/api/users/?modified__gt=%s' % precreate_dt.strftime('%Y-%m-%dT%H:%M:%S'), headers=headers
|
|
1219 |
)
|
|
1220 |
assert len(response.json['results']) == 100
|
|
1221 |
|
|
1222 |
|
1153 |
1223 |
def test_user_synchronization_full(app, admin):
|
1154 |
1224 |
headers = basic_authorization_header(admin)
|
1155 |
1225 |
uuids = []
|
1156 |
|
-
|