Tests and schema added

Author: MaxFabre

core/api.py

@@ -2,10 +2,10 @@
 
 from annotated_types import Ge, Le, MinLen
 from django.conf import settings
-from django.db.models import F, Q
-from ninja import Schema
+from django.contrib.auth import login
+from django.db.models import F
 from django.http import HttpResponse
-from ninja import File, Query
+from ninja import File, Query, Status
 from ninja.security import SessionAuth
 from ninja_extra import ControllerBase, api_controller, paginate, route
 from ninja_extra.exceptions import PermissionDenied
@@ -19,6 +19,7 @@
 from core.schemas import (
     FamilyGodfatherSchema,
     GroupSchema,
+    LoginSchema,
     MarkdownSchema,
     SithFileSchema,
     UploadedFileSchema,
@@ -29,8 +30,8 @@
     UserSchema,
 )
 from core.templatetags.renderer import markdown
+from core.views.forms import LoginForm
 from counter.utils import is_logged_in_counter
-from counter.models import Customer
 
 
 @api_controller("/markdown")
@@ -170,25 +171,28 @@ def get_family_graph(
         }
 
 
-class LoginSchema(Schema):
-    identifier: str
-    password: str
-
-
 @api_controller("/auth")
 class AuthController(ControllerBase):
-    @route.post("/login", auth=None)
+    @route.post(
+        "/login",
+        auth=None,
+        response={200: dict[Literal["id"], int], 401: dict[str, list[str]]},
+    )
     def login(self, body: LoginSchema):
         """Authenticate a user by username, email or AE account id.
 
         Returns the user's id on success, 401 on failure.
         """
-        ident = body.identifier.strip()
-        user = User.objects.filter(Q(username=ident) | Q(email=ident)).first()
-        if user is None:
-            cust = Customer.objects.select_related("user").filter(account_id=ident).first()
-            if cust:
-                user = cust.user
-        if user is None or not user.check_password(body.password):
-            return HttpResponse(status=401)
+        if self.context.request.user.is_authenticated:
+            raise PermissionDenied
+
+        login_form = LoginForm(
+            self.context.request,
+            data={"username": body.identifier, "password": body.password},
+        )
+        if not login_form.is_valid():
+            return Status(401, login_form.errors)
+
+        user = login_form.get_user()
+        login(self.context.request, user)
         return {"id": user.id}

core/schemas.py

@@ -155,6 +155,11 @@ class MarkdownSchema(Schema):
     text: str
 
 
+class LoginSchema(Schema):
+    identifier: str
+    password: str
+
+
 class FamilyGodfatherSchema(Schema):
     godfather: int
     godchild: int

core/tests/test_auth_api.py

@@ -0,0 +1,58 @@
+import json
+
+import pytest
+from django.contrib.auth.hashers import make_password
+from django.test import Client
+from django.urls import reverse
+from model_bakery import baker
+
+from core.models import User
+from counter.models import Customer
+
+
+def post_login(client: Client, identifier: str, password: str):
+    return client.post(
+        reverse("api:login"),
+        data=json.dumps({"identifier": identifier, "password": password}),
+        content_type="application/json",
+    )
+
+@pytest.fixture()
+def user(db) -> User:
+    return baker.make(User, password=make_password("plop"))
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "identifier_getter",
+    [
+        lambda user: user.username,
+        lambda user: user.email,
+        lambda user: Customer.get_or_create(user)[0].account_id,
+    ],
+)
+def test_api_login_success(client: Client, user: User, identifier_getter):
+    response = post_login(client, identifier_getter(user), "plop")
+
+    assert response.status_code == 200
+    assert response.json() == {"id": user.id}
+    assert int(client.session["_auth_user_id"]) == user.id
+
+
+@pytest.mark.django_db
+def test_api_login_fail_invalid_credentials(client: Client, user: User):
+    response = post_login(client, user.username, "wrong-password")
+
+    assert response.status_code == 401
+    assert "_auth_user_id" not in client.session
+
+
+@pytest.mark.django_db
+def test_api_login_fail_if_already_authenticated(client: Client, user: User):
+    already_logged_user = baker.make(User)
+    client.force_login(already_logged_user)
+
+    response = post_login(client, user.username, "plop")
+
+    assert response.status_code == 403
+    assert int(client.session["_auth_user_id"]) == already_logged_user.id