fix: foundation of breaking circular dependencies (#17)

* create user filter set

* ci[setup]: sync dependencies [skip ci]

* default installed django apps

* rename filter to filterset

* fix: include user urls in service

* rename results to data

* read only fields

* update read and write fields

* id is read only

* add class viewset

* use access_code as lookup

* read only class fields

* fix: viewsets

* test retrieve user

* add doc string

* finish user tests

* support indy students

* add school tests

* test filtering user list

* indy student is forbidden

* user_urls_path

* general tests and relocate login methods

* fix urls

* feedback pt.1

* update filters

* fix indy tests

* fix user queryset

* sort importd

* sort imports

* final feedback

* whitespaces

Co-Authored-By: cfl-bot <[email protected]>

Pipfile

@@ -6,6 +6,7 @@ name = "pypi"
 [packages]
 django = "==3.2.20"
 djangorestframework = "==3.13.1"
+django-filter = "==23.2"
 django-countries = "==7.3.1"
 django-two-factor-auth = "==1.13.2"
 django-cors-headers = "==4.1.0"

codeforlife/pagination.py

@@ -0,0 +1,20 @@
+from rest_framework.pagination import (
+    LimitOffsetPagination as _LimitOffsetPagination,
+)
+from rest_framework.response import Response
+
+
+class LimitOffsetPagination(_LimitOffsetPagination):
+    default_limit = 50
+    max_limit = 150
+
+    def get_paginated_response(self, data):
+        return Response(
+            {
+                "count": self.count,
+                "offset": self.offset,
+                "limit": self.limit,
+                "max_limit": self.max_limit,
+                "data": data,
+            }
+        )

codeforlife/settings/django.py

@@ -122,3 +122,13 @@
         "NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
     },
 ]
+
+# Installed Apps
+# https://docs.djangoproject.com/en/3.2/ref/settings/#installed-apps
+
+INSTALLED_APPS = [
+    "codeforlife.user",
+    "corsheaders",
+    "rest_framework",
+    "django_filters",
+]

codeforlife/settings/third_party.py

@@ -10,3 +10,16 @@
 CORS_ALLOW_ALL_ORIGINS = DEBUG
 CORS_ALLOW_CREDENTIALS = True
 CORS_ALLOWED_ORIGINS = ["https://www.codeforlife.education"]
+
+# REST framework
+# https://www.django-rest-framework.org/api-guide/settings/#settings
+
+REST_FRAMEWORK = {
+    "DEFAULT_PERMISSION_CLASSES": [
+        "rest_framework.permissions.IsAuthenticated",
+    ],
+    "DEFAULT_FILTER_BACKENDS": [
+        "django_filters.rest_framework.DjangoFilterBackend"
+    ],
+    "DEFAULT_PAGINATION_CLASS": "codeforlife.pagination.LimitOffsetPagination",
+}

codeforlife/tests/__init__.py

@@ -1 +1,2 @@
+from .api import APITestCase, APIClient
 from .cron import CronTestCase, CronTestClient

codeforlife/tests/api.py

@@ -0,0 +1,281 @@
+import typing as t
+from unittest.mock import patch
+
+from django.db.models import Model
+from django.db.models.query import QuerySet
+from django.urls import reverse
+from django.utils import timezone
+from django.utils.http import urlencode
+from pyotp import TOTP
+from rest_framework.response import Response
+from rest_framework.serializers import ModelSerializer
+from rest_framework.test import APIClient as _APIClient
+from rest_framework.test import APITestCase as _APITestCase
+from rest_framework.viewsets import ModelViewSet
+
+from ..user.models import AuthFactor, User
+
+AnyModelViewSet = t.TypeVar("AnyModelViewSet", bound=ModelViewSet)
+AnyModelSerializer = t.TypeVar("AnyModelSerializer", bound=ModelSerializer)
+AnyModel = t.TypeVar("AnyModel", bound=Model)
+
+
+class APIClient(_APIClient):
+    StatusCodeAssertion = t.Optional[t.Union[int, t.Callable[[int], bool]]]
+    ListFilters = t.Optional[t.Dict[str, str]]
+
+    @staticmethod
+    def status_code_is_ok(status_code: int):
+        return 200 <= status_code < 300
+
+    def generic(
+        self,
+        method,
+        path,
+        data="",
+        content_type="application/octet-stream",
+        secure=False,
+        status_code_assertion: StatusCodeAssertion = None,
+        **extra,
+    ):
+        wsgi_response = super().generic(
+            method, path, data, content_type, secure, **extra
+        )
+
+        # Use a custom kwarg to handle the common case of checking the
+        # response's status code.
+        if status_code_assertion is None:
+            status_code_assertion = self.status_code_is_ok
+        elif isinstance(status_code_assertion, int):
+            expected_status_code = status_code_assertion
+            status_code_assertion = (
+                lambda status_code: status_code == expected_status_code
+            )
+        assert status_code_assertion(
+            wsgi_response.status_code
+        ), f"Unexpected status code: {wsgi_response.status_code}."
+
+        return wsgi_response
+
+    def login(self, **credentials):
+        assert super().login(
+            **credentials
+        ), f"Failed to login with credentials: {credentials}."
+
+        user = User.objects.get(session=self.session.session_key)
+
+        if user.session.session_auth_factors.filter(
+            auth_factor__type=AuthFactor.Type.OTP
+        ).exists():
+            now = timezone.now()
+            otp = TOTP(user.otp_secret).at(now)
+            with patch.object(timezone, "now", return_value=now):
+                assert super().login(
+                    otp=otp
+                ), f'Failed to login with OTP "{otp}" at {now}.'
+
+        assert user.is_authenticated, "Failed to authenticate user."
+
+        return user
+
+    def login_teacher(self, is_admin: bool, **credentials):
+        user = self.login(**credentials)
+        assert user.teacher
+        assert user.teacher.school
+        assert is_admin == user.teacher.is_admin
+        return user
+
+    def login_student(self, **credentials):
+        user = self.login(**credentials)
+        assert user.student
+        assert user.student.class_field.teacher.school
+        return user
+
+    def login_indy_student(self, **credentials):
+        user = self.login(**credentials)
+        assert user.student
+        assert not user.student.class_field
+        return user
+
+    @staticmethod
+    def assert_data_equals_model(
+        data: t.Dict[str, t.Any],
+        model: AnyModel,
+        model_serializer_class: t.Type[AnyModelSerializer],
+    ):
+        assert (
+            data == model_serializer_class(model).data
+        ), "Data does not equal serialized model."
+
+    def retrieve(
+        self,
+        basename: str,
+        model: AnyModel,
+        model_serializer_class: t.Type[AnyModelSerializer],
+        status_code_assertion: StatusCodeAssertion = None,
+        model_view_set_class: t.Type[AnyModelViewSet] = None,
+        **kwargs,
+    ):
+        lookup_field = (
+            "pk"
+            if model_view_set_class is None
+            else model_view_set_class.lookup_field
+        )
+
+        response: Response = self.get(
+            reverse(
+                f"{basename}-detail",
+                kwargs={lookup_field: getattr(model, lookup_field)},
+            ),
+            status_code_assertion=status_code_assertion,
+            **kwargs,
+        )
+
+        if self.status_code_is_ok(response.status_code):
+            self.assert_data_equals_model(
+                response.json(),
+                model,
+                model_serializer_class,
+            )
+
+        return response
+
+    def list(
+        self,
+        basename: str,
+        models: t.Iterable[AnyModel],
+        model_serializer_class: t.Type[AnyModelSerializer],
+        status_code_assertion: StatusCodeAssertion = None,
+        filters: ListFilters = None,
+        **kwargs,
+    ):
+        model_class: t.Type[AnyModel] = model_serializer_class.Meta.model
+        assert model_class.objects.difference(
+            model_class.objects.filter(pk__in=[model.pk for model in models])
+        ).exists(), "List must exclude some models for a valid test."
+
+        response: Response = self.get(
+            f"{reverse(f'{basename}-list')}?{urlencode(filters or {})}",
+            status_code_assertion=status_code_assertion,
+            **kwargs,
+        )
+
+        if self.status_code_is_ok(response.status_code):
+            for data, model in zip(response.json()["data"], models):
+                self.assert_data_equals_model(
+                    data,
+                    model,
+                    model_serializer_class,
+                )
+
+        return response
+
+
+class APITestCase(_APITestCase):
+    client: APIClient
+    client_class = APIClient
+
+    def get_other_user(
+        self,
+        user: User,
+        other_users: QuerySet[User],
+        is_teacher: bool,
+    ):
+        """
+        Get a different user.
+        """
+
+        other_user = other_users.first()
+        assert other_user
+        assert user != other_user
+        assert other_user.is_teacher if is_teacher else other_user.is_student
+        return other_user
+
+    def get_other_school_user(
+        self,
+        user: User,
+        other_users: QuerySet[User],
+        is_teacher: bool,
+    ):
+        """
+        Get a different user that is in a school.
+        - the provided user does not have to be in a school.
+        - the other user has to be in a school.
+        """
+
+        other_user = self.get_other_user(user, other_users, is_teacher)
+        assert (
+            other_user.teacher.school
+            if is_teacher
+            else other_user.student.class_field.teacher.school
+        )
+        return other_user
+
+    def get_another_school_user(
+        self,
+        user: User,
+        other_users: QuerySet[User],
+        is_teacher: bool,
+        same_school: bool,
+        same_class: bool = None,
+    ):
+        """
+        Get a different user that is also in a school.
+         - the provided user has to be in a school.
+         - the other user has to be in a school.
+        """
+
+        other_user = self.get_other_school_user(user, other_users, is_teacher)
+
+        school = (
+            user.teacher.school
+            if user.teacher
+            else user.student.class_field.teacher.school
+        )
+        assert school
+
+        other_school = (
+            other_user.teacher.school
+            if is_teacher
+            else other_user.student.class_field.teacher.school
+        )
+        assert other_school
+
+        if same_school:
+            assert school == other_school
+
+            # Cannot assert that 2 teachers are in the same class since a class
+            # can only have 1 teacher.
+            if not (user.is_teacher and other_user.is_teacher):
+                # At this point, same_class needs to be set.
+                assert same_class is not None, "same_class must be set."
+
+                # If one of the users is a teacher.
+                if user.is_teacher or is_teacher:
+                    # Get the teacher.
+                    teacher = other_user if is_teacher else user
+
+                    # Get the student's class' teacher.
+                    class_teacher = (
+                        user if is_teacher else other_user
+                    ).student.class_field.teacher.new_user
+
+                    # Assert the teacher is the class' teacher.
+                    assert (
+                        teacher == class_teacher
+                        if same_class
+                        else teacher != class_teacher
+                    )
+                # Else, both users are students.
+                else:
+                    assert (
+                        user.student.class_field
+                        == other_user.student.class_field
+                        if same_class
+                        else user.student.class_field
+                        != other_user.student.class_field
+                    )
+        else:
+            assert school != other_school
+
+        return other_user

codeforlife/tests/cron.py

@@ -1,28 +1,11 @@
-from rest_framework.test import APIClient, APITestCase
+from .api import APIClient, APITestCase
 
 
 class CronTestClient(APIClient):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs, HTTP_X_APPENGINE_CRON="true")
 
-    def generic(
-        self,
-        method,
-        path,
-        data="",
-        content_type="application/octet-stream",
-        secure=False,
-        **extra,
-    ):
-        wsgi_response = super().generic(
-            method, path, data, content_type, secure, **extra
-        )
-        assert (
-            200 <= wsgi_response.status_code < 300
-        ), f"Response has error status code: {wsgi_response.status_code}"
-
-        return wsgi_response
-
 
 class CronTestCase(APITestCase):
+    client: CronTestClient
     client_class = CronTestClient