0001-api-add-a-full_known_users-option-to-synchronization.patch

Paul Marillonnet, 25 octobre 2021 15:01

Voir les différences: en ligne côte à côte

Subject: [PATCH] api: add a full_known_users option to /synchronization/
 endpoint (#57567)

 src/authentic2/api_views.py | 34 ++++++++++++--
 src/authentic2/settings.py  |  2 +
 tests/test_api.py           | 93 +++++++++++++++++++++++++++++++++++++
 3 files changed, 126 insertions(+), 3 deletions(-)

         class SynchronizationSerializer(serializers.Serializer):
             known_uuids = serializers.ListField(child=serializers.CharField())
             full_known_users = serializers.BooleanField(required=False)
             page_size = serializers.IntegerField(
                 min_value=settings.REST_FRAMEWORK.get('MIN_PAGE_SIZE', 0) or 10,
                 max_value=settings.REST_FRAMEWORK.get('MAX_PAGE_SIZE', 0) or 1000,
                 required=False,
+            )
             page = serializers.IntegerField(min_value=1, required=False)
         def check_uuids(self, uuids):
             User = get_user_model()
             known_uuids = User.objects.filter(uuid__in=uuids).values_list('uuid', flat=True)
             return set(uuids) - set(known_uuids)
             return (known_uuids, set(uuids) - set(known_uuids))
         def get_users_from_uuids(self, known_uuids):
             User = get_user_model()
             known_users = User.objects.filter(uuid__in=known_uuids)
             return [BaseUserSerializer(user).data for user in known_users]
         @action(detail=False, methods=['post'], permission_classes=(DjangoPermission('custom_user.search_user'),))
         def synchronization(self, request):
-...
                 response = {'result': 0, 'errors': serializer.errors}
                 return Response(response, status.HTTP_400_BAD_REQUEST)
             hooks.call_hooks('api_modify_serializer_after_validation', self, serializer)
             unknown_uuids = self.check_uuids(serializer.validated_data.get('known_uuids', []))
             known_uuids, unknown_uuids = self.check_uuids(serializer.validated_data.get('known_uuids', []))
             full_known_users = serializer.validated_data.get('full_known_users', None)
             data = {
                 'result': 1,
                 'unknown_uuids': unknown_uuids,
+            }
             if full_known_users:
                 page_size = serializer.validated_data.get('page_size', None)
                 page = serializer.validated_data.get('page', None)
                 if page:
                     page_size = page_size or settings.REST_FRAMEWORK.get('PAGE_SIZE', 0) or 100
                     if len(known_uuids) <= (page - 1) * page_size:
                         known_uuids = []
                     else:
                         upper_bound = min(len(known_uuids), page * page_size)
                         known_uuids = known_uuids[(page - 1) * page_size : upper_bound]
                 if known_uuids:
                     data['known_users'] = self.get_users_from_uuids(known_uuids)
                 else:
                     data['known_users'] = []
             else:
                 data['unknown_uuids'] = unknown_uuids
             hooks.call_hooks('api_modify_response', self, 'synchronization', data)
             return Response(data)

         ),
         'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.LimitOffsetPagination',
         'PAGE_SIZE': 100,
         'MAX_PAGE_SIZE': 1000,
         'MIN_PAGE_SIZE': 10,
+    }

         assert set(response.json['unknown_uuids']) == set(unknown_uuids)
     def test_user_synchronization_full(app, admin):
         headers = basic_authorization_header(admin)
         uuids = []
         for _ in range(100):
             user = User.objects.create(first_name='jim', last_name='jam')
             uuids.append(user.uuid)
         unknown_uuids = [uuid.uuid4().hex for i in range(100)]
         url = reverse('a2-api-users-synchronization')
         content = {
             'known_uuids': uuids + unknown_uuids,
             'full_known_users': 1,
+        }
         random.shuffle(content['known_uuids'])
         response = app.post_json(url, params=content, headers=headers)
         assert response.json['result'] == 1
         # known users returned as part of api's full mode:
         assert len(response.json['known_users']) == 100
         for user_dict in response.json['known_users']:
             assert user_dict['first_name'] == 'jim'
             assert user_dict['last_name'] == 'jam'
             assert {
                 'uuid',
                 'email',
                 'is_staff',
                 'is_superuser',
                 'email_verified',
                 'ou',
                 'is_active',
                 'deactivation',
                 'modified',
             }.issubset(set(user_dict.keys()))
     def test_user_synchronization_full_paginated(app, admin):
         headers = basic_authorization_header(admin)
         uuids = []
         for _ in range(100):
             user = User.objects.create(first_name='jim', last_name='jam')
             uuids.append(user.uuid)
         unknown_uuids = [uuid.uuid4().hex for i in range(100)]
         url = reverse('a2-api-users-synchronization')
         # only page-size parameter declared, no effect
         content = {
             'known_uuids': uuids + unknown_uuids,
             'full_known_users': 1,
             'page_size': 30,
+        }
         random.shuffle(content['known_uuids'])
         response = app.post_json(url, params=content, headers=headers)
         assert len(response.json['known_users']) == 100
         # page and page-size parameters
         content = {
             'known_uuids': uuids + unknown_uuids,
             'full_known_users': 1,
             'page_size': 30,
+        }
         for i in range(3):
             content['page'] = i + 1
             random.shuffle(content['known_uuids'])
             response = app.post_json(url, params=content, headers=headers)
             assert len(response.json['known_users']) == 30
         # last page
         content['page'] = 4
         random.shuffle(content['known_uuids'])
         response = app.post_json(url, params=content, headers=headers)
         assert len(response.json['known_users']) == 10
         for i in range(5, 8):
             content['page'] = i
             random.shuffle(content['known_uuids'])
             response = app.post_json(url, params=content, headers=headers)
             assert len(response.json['known_users']) == 0
         # page size larger than result size:
         content = {
             'known_uuids': uuids + unknown_uuids,
             'full_known_users': 1,
             'page_size': 120,
             'page': 1,
+        }
         random.shuffle(content['known_uuids'])
         response = app.post_json(url, params=content, headers=headers)
         assert len(response.json['known_users']) == 100
         content['page'] = 2
         response = app.post_json(url, params=content, headers=headers)
         assert len(response.json['known_users']) == 0
     def test_api_drf_authentication_class(app, admin, user_ou1, oidc_client):
         url = '/api/users/%s/' % user_ou1.uuid
         # test invalid client
+    -

Projet

Général

Profil

Produits Entr'ouvert » Authentic 2

0001-api-add-a-full_known_users-option-to-synchronization.patch