Contributor frontend 4

Pipfile

@@ -23,11 +23,11 @@ name = "pypi"
 # 5. Run `pipenv install --dev` in your terminal.
 
 [packages]
-codeforlife = {ref = "v0.18.4", git = "https://github.com/ocadotechnology/codeforlife-package-python.git"}
+codeforlife = {ref = "v0.18.8", git = "https://github.com/ocadotechnology/codeforlife-package-python.git"}
 # 🚫 Don't add [packages] below that are inherited from the CFL package.
 
 [dev-packages]
-codeforlife = {ref = "v0.18.4", git = "https://github.com/ocadotechnology/codeforlife-package-python.git", extras = ["dev"]}
+codeforlife = {ref = "v0.18.8", git = "https://github.com/ocadotechnology/codeforlife-package-python.git", extras = ["dev"]}
 # codeforlife = {file = "../codeforlife-package-python", editable = true, extras = ["dev"]}
 # 🚫 Don't add [dev-packages] below that are inherited from the CFL package.
 

api/auth/backends/github.py

@@ -6,6 +6,7 @@
 
 import requests
 from codeforlife.request import HttpRequest
+from codeforlife.types import JsonDict
 from django.conf import settings
 from django.contrib.auth.backends import BaseBackend
 
@@ -35,48 +36,17 @@ def authenticate(  # type: ignore[override]
             },
             timeout=5,
         )
-
         if not response.ok:
             return None
-        auth_data = response.json()
 
-        # Code expired
-        if "error" in auth_data:
+        access_token: JsonDict = response.json()
+        if "error" in access_token:
             return None
 
-        # Get user's information
-        response = requests.get(
-            url="https://api.github.com/user",
-            headers={
-                "Accept": "application/json",
-                "X-GitHub-Api-Version": "2022-11-28",
-                # pylint: disable-next=line-too-long
-                "Authorization": f"{auth_data['token_type']} {auth_data['access_token']}",
-            },
-            timeout=5,
+        return Contributor.sync_with_github(
+            auth=f"{access_token['token_type']} {access_token['access_token']}"
         )
 
-        if not response.ok:
-            return None
-
-        contributor_data = response.json()
-
-        try:
-            return Contributor.objects.get_or_create(
-                id=contributor_data["id"],
-                defaults={
-                    "id": contributor_data["id"],
-                    "email": contributor_data["email"],
-                    "name": contributor_data["name"],
-                    "location": contributor_data.get("location"),
-                    "html_url": contributor_data["html_url"],
-                    "avatar_url": contributor_data["avatar_url"],
-                },
-            )[0]
-        # pylint: disable-next=bare-except
-        except:
-            return None
-
     # pylint: disable-next=arguments-renamed
     def get_user(self, contributor_id: int):
         try:

api/auth/backends/github_test.py

@@ -80,13 +80,15 @@ def test_login__invalid_code(self):
         with patch.object(
             requests, "post", return_value=response
         ) as requests_post:
-            res = self.backend.authenticate(request=self.request, code=code)
+            contributor = self.backend.authenticate(
+                request=self.request, code=code
+            )
 
-            assert not res
+            assert not contributor
             self._assert_request_github_access_token(requests_post, code)
 
-    def test_login__code_expired(self):
-        """Provided code must not expired."""
+    def test_login__error(self):
+        """Login cannot return an error in the response."""
         code = "7f06468085765cdc1578"
 
         response = requests.Response()
@@ -98,50 +100,9 @@ def test_login__code_expired(self):
         with patch.object(
             requests, "post", return_value=response
         ) as requests_post:
-            res = self.backend.authenticate(request=self.request, code=code)
+            contributor = self.backend.authenticate(
+                request=self.request, code=code
+            )
 
-            assert not res
+            assert not contributor
             self._assert_request_github_access_token(requests_post, code)
-
-    def test_login__token_failure(self):
-        """Access token did not get accepted by github."""
-        code = "7f06468085765cdc1578"
-
-        response_get = requests.Response()
-        response_get.status_code = status.HTTP_401_UNAUTHORIZED
-
-        with patch.object(
-            requests, "post", return_value=self.gh_access_token_response
-        ) as requests_post:
-            with patch.object(
-                requests, "get", return_value=response_get
-            ) as requests_get:
-                res = self.backend.authenticate(request=self.request, code=code)
-
-                assert not res
-                self._assert_request_github_access_token(requests_post, code)
-                self._assert_request_github_user(requests_get, "Bearer 123254")
-
-    def test_login__invalid_contributor_data(self):
-        """
-        Github did not provide data needed to log the user in.
-        """
-        code = "7f06468085765cdc1578"
-
-        response_get = requests.Response()
-        response_get.status_code = status.HTTP_200_OK
-        response_get.encoding = "utf-8"
-        # pylint: disable-next=protected-access
-        response_get._content = json.dumps({}).encode("utf-8")
-
-        with patch.object(
-            requests, "post", return_value=self.gh_access_token_response
-        ) as requests_post:
-            with patch.object(
-                requests, "get", return_value=response_get
-            ) as requests_get:
-                res = self.backend.authenticate(request=self.request, code=code)
-
-                assert not res
-                self._assert_request_github_access_token(requests_post, code)
-                self._assert_request_github_user(requests_get, "Bearer 123254")

api/common/__init__.py

@@ -0,0 +1,10 @@
+"""
+© Ocado Group
+Created on 13/09/2024 at 12:00:14(+03:00).
+"""
+
+from .model_serializer import ModelSerializer
+from .model_serializer_test_case import ModelSerializerTestCase
+from .model_view_set import ModelViewSet
+from .model_view_set_test_case import ModelViewSetTestCase
+from .request import Request

api/common/api_request_factory.py

@@ -0,0 +1,212 @@
+"""
+© Ocado Group
+Created on 13/09/2024 at 12:00:25(+03:00).
+"""
+
+import typing as t
+
+from django.core.handlers.wsgi import WSGIRequest
+from rest_framework.parsers import (
+    FileUploadParser,
+    FormParser,
+    JSONParser,
+    MultiPartParser,
+)
+from rest_framework.test import APIRequestFactory as _APIRequestFactory
+
+from ..models import Contributor
+from .request import Request
+
+
+class APIRequestFactory(_APIRequestFactory):
+    """Custom API request factory that returns DRF's Request object."""
+
+    def request(self, user: t.Optional[Contributor] = None, **kwargs):
+        wsgi_request = t.cast(WSGIRequest, super().request(**kwargs))
+
+        request = Request(
+            wsgi_request,
+            parsers=[
+                JSONParser(),
+                FormParser(),
+                MultiPartParser(),
+                FileUploadParser(),
+            ],
+        )
+
+        if user:
+            # pylint: disable-next=attribute-defined-outside-init
+            request.user = user
+
+        return request
+
+    # pylint: disable-next=too-many-arguments
+    def generic(
+        self,
+        method: str,
+        path: t.Optional[str] = None,
+        data: t.Optional[str] = None,
+        content_type: t.Optional[str] = None,
+        secure: bool = True,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        return t.cast(
+            Request,
+            super().generic(
+                method,
+                path or "/",
+                data or "",
+                content_type or "application/json",
+                secure,
+                user=user,
+                **extra,
+            ),
+        )
+
+    def get(  # type: ignore[override]
+        self,
+        path: t.Optional[str] = None,
+        data: t.Any = None,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        return t.cast(
+            Request,
+            super().get(
+                path or "/",
+                data,
+                user=user,
+                **extra,
+            ),
+        )
+
+    # pylint: disable-next=too-many-arguments
+    def post(  # type: ignore[override]
+        self,
+        path: t.Optional[str] = None,
+        data: t.Any = None,
+        # pylint: disable-next=redefined-builtin
+        format: t.Optional[str] = None,
+        content_type: t.Optional[str] = None,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        if format is None and content_type is None:
+            format = "json"
+
+        return t.cast(
+            Request,
+            super().post(
+                path or "/",
+                data,
+                format,
+                content_type,
+                user=user,
+                **extra,
+            ),
+        )
+
+    # pylint: disable-next=too-many-arguments
+    def put(  # type: ignore[override]
+        self,
+        path: t.Optional[str] = None,
+        data: t.Any = None,
+        # pylint: disable-next=redefined-builtin
+        format: t.Optional[str] = None,
+        content_type: t.Optional[str] = None,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        if format is None and content_type is None:
+            format = "json"
+
+        return t.cast(
+            Request,
+            super().put(
+                path or "/",
+                data,
+                format,
+                content_type,
+                user=user,
+                **extra,
+            ),
+        )
+
+    # pylint: disable-next=too-many-arguments
+    def patch(  # type: ignore[override]
+        self,
+        path: t.Optional[str] = None,
+        data: t.Any = None,
+        # pylint: disable-next=redefined-builtin
+        format: t.Optional[str] = None,
+        content_type: t.Optional[str] = None,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        if format is None and content_type is None:
+            format = "json"
+
+        return t.cast(
+            Request,
+            super().patch(
+                path or "/",
+                data,
+                format,
+                content_type,
+                user=user,
+                **extra,
+            ),
+        )
+
+    # pylint: disable-next=too-many-arguments
+    def delete(  # type: ignore[override]
+        self,
+        path: t.Optional[str] = None,
+        data: t.Any = None,
+        # pylint: disable-next=redefined-builtin
+        format: t.Optional[str] = None,
+        content_type: t.Optional[str] = None,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        if format is None and content_type is None:
+            format = "json"
+
+        return t.cast(
+            Request,
+            super().delete(
+                path or "/",
+                data,
+                format,
+                content_type,
+                user=user,
+                **extra,
+            ),
+        )
+
+    # pylint: disable-next=too-many-arguments
+    def options(  # type: ignore[override]
+        self,
+        path: t.Optional[str] = None,
+        data: t.Optional[t.Union[t.Dict[str, str], str]] = None,
+        # pylint: disable-next=redefined-builtin
+        format: t.Optional[str] = None,
+        content_type: t.Optional[str] = None,
+        user: t.Optional[Contributor] = None,
+        **extra
+    ):
+        if format is None and content_type is None:
+            format = "json"
+
+        return t.cast(
+            Request,
+            super().options(
+                path or "/",
+                data or {},
+                format,
+                content_type,
+                user=user,
+                **extra,
+            ),
+        )