Add domain endpoints (#17)

Co-authored-by: Christopher Bartz <christopher.bartz@canonical.com>

httprequest_lego_provider/serializers.py

@@ -0,0 +1,37 @@
+# Copyright 2024 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""Serializers."""
+
+from rest_framework import serializers
+
+from .models import Domain, DomainUserPermission
+
+
+class DomainSerializer(serializers.ModelSerializer):
+    """Serializer for the Domain objects."""
+
+    class Meta:
+        """Serializer configuration.
+
+        Attributes:
+            model: the model to serialize.
+            fields: fields to serialize.
+        """
+
+        model = Domain
+        fields = "__all__"
+
+
+class DomainUserPermissionSerializer(serializers.ModelSerializer):
+    """Serializer for the DomainUserPermission objects."""
+
+    class Meta:
+        """Serializer configuration.
+
+        Attributes:
+            model: the model to serialize.
+            fields: fields to serialize.
+        """
+
+        model = DomainUserPermission
+        fields = "__all__"

httprequest_lego_provider/tests/conftest.py

@@ -37,6 +37,32 @@ def user_auth_token(username: str, user_password: str, user: User) -> str:
     return base64.b64encode(bytes(f"{username}:{user_password}", "utf-8")).decode("utf-8")
 
 
+@pytest.fixture(scope="module")
+def admin_username() -> str:
+    """Provide an admin username."""
+    return "test_admin_user"
+
+
+@pytest.fixture(scope="module")
+def admin_user_password() -> str:
+    """Provide an admin user password."""
+    return secrets.token_hex()
+
+
+@pytest.fixture(scope="function")
+def admin_user(admin_username: str, admin_user_password: str) -> User:
+    """Provide an admin user."""
+    return User.objects.create_user(admin_username, password=admin_user_password, is_staff=True)
+
+
+@pytest.fixture(scope="function")
+def admin_user_auth_token(admin_username: str, admin_user_password: str, admin_user: User) -> str:
+    """Provide the auth_token for the admin user."""
+    return base64.b64encode(bytes(f"{admin_username}:{admin_user_password}", "utf-8")).decode(
+        "utf-8"
+    )
+
+
 @pytest.fixture(scope="module")
 def fqdn():
     """Provide a valid FQDN."""

httprequest_lego_provider/tests/test_views.py

@@ -7,6 +7,7 @@
 from unittest.mock import patch
 
 import pytest
+from django.contrib.auth.models import User
 from django.test import Client
 
 from httprequest_lego_provider.forms import FQDN_PREFIX
@@ -53,7 +54,7 @@ def test_post_present_when_auth_header_invalid(client: Client):
 @pytest.mark.django_db
 def test_post_present_when_logged_in_and_no_fqdn(client: Client, user_auth_token: str, fqdn: str):
     """
-    arrange: log in a user.
+    arrange: log in a non-admin user.
     act: submit a POST request for the present URL.
     assert: a 403 is returned.
     """
@@ -72,7 +73,7 @@ def test_post_present_when_logged_in_and_no_permission(
     client: Client, user_auth_token: str, domain: Domain
 ):
     """
-    arrange: log in a user and insert a domain in the database.
+    arrange: log in a non-admin user and insert a domain in the database.
     act: submit a POST request for the present URL.
     assert: a 403 is returned.
     """
@@ -128,7 +129,7 @@ def test_post_present_when_logged_in_and_fqdn_invalid(client: Client, user_auth_
 @pytest.mark.django_db
 def test_get_present_when_logged_in(client: Client, user_auth_token: str):
     """
-    arrange: log in a user.
+    arrange: log in a non-admin user.
     act: submit a GET request for the present URL.
     assert: a 405 is returned.
     """
@@ -154,7 +155,7 @@ def test_post_cleanup_when_not_logged_in(client: Client):
 @pytest.mark.django_db
 def test_post_cleanup_when_logged_in_and_no_fqdn(client: Client, user_auth_token: str):
     """
-    arrange: log in a user.
+    arrange: log in a non-admin user.
     act: submit a POST request for the cleanup URL.
     assert: a 403 is returned.
     """
@@ -173,7 +174,7 @@ def test_post_cleanup_when_logged_in_and_no_permission(
     client: Client, user_auth_token: str, domain: Domain
 ):
     """
-    arrange: log in a user.
+    arrange: log in a non-admin user.
     act: submit a POST request for the cleanup URL.
     assert: a 403 is returned.
     """
@@ -229,7 +230,7 @@ def test_post_cleanup_when_logged_in_and_fqdn_invalid(client: Client, user_auth_
 @pytest.mark.django_db
 def test_get_cleanup_when_logged_in(client: Client, user_auth_token: str):
     """
-    arrange: log in a user.
+    arrange: log in a non-admin user.
     act: submit a GET request for the cleanup URL.
     assert: a 405 is returned.
     """
@@ -264,3 +265,134 @@ def test_test_jwt_token_login(
         )
 
         assert response.status_code == 204
+
+
+@pytest.mark.django_db
+def test_post_domain_when_logged_in_as_non_admin_user(client: Client, user_auth_token: str):
+    """
+    arrange: log in a non-admin user.
+    act: submit a POST request for the domain URL.
+    assert: a 403 is returned and the domain is not inserted in the database.
+    """
+    response = client.post(
+        "/api/v1/domains/",
+        data={"fqdn": "example.com"},
+        headers={"AUTHORIZATION": f"Basic {user_auth_token}"},
+    )
+
+    with pytest.raises(Domain.DoesNotExist):
+        Domain.objects.get(fqdn="example.com")
+    assert response.status_code == 403
+
+
+@pytest.mark.django_db
+def test_post_domain_when_logged_in_as_admin_user(client: Client, admin_user_auth_token: str):
+    """
+    arrange: log in an admin user.
+    act: submit a POST request for the domain URL.
+    assert: a 201 is returned and the domain is inserted in the database.
+    """
+    response = client.post(
+        "/api/v1/domains/",
+        data={"fqdn": "example.com"},
+        headers={"AUTHORIZATION": f"Basic {admin_user_auth_token}"},
+    )
+
+    assert Domain.objects.get(fqdn="example.com") is not None
+    assert response.status_code == 201
+
+
+@pytest.mark.django_db
+def test_post_domain_when_logged_in_as_admin_user_and_domain_invalid(
+    client: Client, admin_user_auth_token: str
+):
+    """
+    arrange: log in a admin user.
+    act: submit a POST request with an invalid value for the domain URL.
+    assert: a 400 is returned.
+    """
+    response = client.post(
+        "/api/v1/domains/",
+        data={"fqdn": "invalid-value"},
+        headers={"AUTHORIZATION": f"Basic {admin_user_auth_token}"},
+    )
+
+    with pytest.raises(Domain.DoesNotExist):
+        Domain.objects.get(fqdn="invalid-value")
+    assert response.status_code == 400
+
+
+@pytest.mark.django_db
+def test_post_domain_user_permission_when_logged_in_as_non_admin_user(
+    client: Client, user_auth_token: str, domain: Domain, user: User
+):
+    """
+    arrange: log in a non-admin user.
+    act: submit a POST request for the domain user permission URL.
+    assert: a 403 is returned and the domain is not inserted in the database.
+    """
+    response = client.post(
+        "/api/v1/domain-user-permissions/",
+        data={"domain": domain.id, "user": user.id, "text": "whatever"},
+        headers={"AUTHORIZATION": f"Basic {user_auth_token}"},
+    )
+
+    assert not DomainUserPermission.objects.filter(user=user, domain=domain)
+    assert response.status_code == 403
+
+
+@pytest.mark.django_db
+def test_post_domain_user_permission_with_invalid_domain_when_logged_in_as_admin_user(
+    client: Client, admin_user_auth_token: str, user: User
+):
+    """
+    arrange: log in an admin user.
+    act: submit a POST request for the domain user permission URL for a non existing domain.
+    assert: a 400 is returned and the domain is not inserted in the database.
+    """
+    response = client.post(
+        "/api/v1/domain-user-permissions/",
+        data={"domain": 1, "user": user.id, "text": "whatever"},
+        headers={"AUTHORIZATION": f"Basic {admin_user_auth_token}"},
+    )
+
+    assert not DomainUserPermission.objects.filter(user=user, domain=1)
+    assert response.status_code == 400
+
+
+@pytest.mark.django_db
+def test_post_domain_user_permission_with_invalid_user_when_logged_in_as_admin_user(
+    client: Client, admin_user_auth_token: str, domain: Domain
+):
+    """
+    arrange: log in an admin user.
+    act: submit a POST request for the domain user permission URL for a non existing user.
+    assert: a 400 is returned and the domain is not inserted in the database.
+    """
+    response = client.post(
+        "/api/v1/domain-user-permissions/",
+        data={"domain": domain.id, "user": 99, "text": "whatever"},
+        headers={"AUTHORIZATION": f"Basic {admin_user_auth_token}"},
+    )
+
+    assert not DomainUserPermission.objects.filter(user=99, domain=domain)
+    assert response.status_code == 400
+
+
+@pytest.mark.django_db
+def test_post_domain_user_permission_when_logged_in_as_admin_user(
+    client: Client, admin_user_auth_token: str, user: User, domain: Domain
+):
+    """
+    arrange: log in an admin user.
+    act: submit a POST request for the domain user permission URL for a existing domain.
+    assert: a 201 is returned and the domain user permission is inserted in the database.
+    """
+    response = client.post(
+        "/api/v1/domain-user-permissions/",
+        data={"domain": domain.id, "user": user.id, "text": "whatever"},
+        headers={"AUTHORIZATION": f"Basic {admin_user_auth_token}"},
+    )
+
+    assert DomainUserPermission.objects.filter(user=99, domain=domain) is not None
+    assert response.status_code == 201

httprequest_lego_provider/urls.py

@@ -3,11 +3,17 @@
 """Urls."""
 
 from django.urls import include, path
+from rest_framework.routers import DefaultRouter
 
 from . import views
 
+router = DefaultRouter()
+router.register("domains", views.DomainViewSet)
+router.register("domain-user-permissions", views.DomainUserPermissionViewSet)
+
 urlpatterns = [
     path("api/v1/cleanup/", views.handle_cleanup, name="cleanup"),
     path("api/v1/present/", views.handle_present, name="present"),
     path("api/v1/accounts/", include("django.contrib.auth.urls")),
+    path("api/v1/", include(router.urls)),
 ]

httprequest_lego_provider/views.py

@@ -2,15 +2,21 @@
 # See LICENSE file for licensing details.
 """Views."""
 
+# Disable too-many-ancestors rule since we can't control inheritance for the ViewSets.
+# pylint:disable=too-many-ancestors
+
 from typing import Optional
 
 from django.core.exceptions import PermissionDenied
 from django.http import HttpRequest, HttpResponse
+from rest_framework import viewsets
 from rest_framework.decorators import api_view
+from rest_framework.permissions import IsAdminUser
 
 from .dns import remove_dns_record, write_dns_record
 from .forms import CleanupForm, PresentForm
 from .models import Domain, DomainUserPermission
+from .serializers import DomainSerializer, DomainUserPermissionSerializer
 
 
 @api_view(["POST"])
@@ -66,3 +72,31 @@ def handle_cleanup(request: HttpRequest) -> Optional[HttpResponse]:
     except Domain.DoesNotExist:
         pass
     raise PermissionDenied
+
+
+class DomainViewSet(viewsets.ModelViewSet):
+    """Views for the Domain.
+
+    Attributes:
+        queryset: query for the objects in the model.
+        serializer_class: class used for serialization.
+        permission_classes: list of classes to match permissions.
+    """
+
+    queryset = Domain.objects.all()
+    serializer_class = DomainSerializer
+    permission_classes = [IsAdminUser]
+
+
+class DomainUserPermissionViewSet(viewsets.ModelViewSet):
+    """Views for the DomainUserPermission.
+
+    Attributes:
+        queryset: query for the objects in the model.
+        serializer_class: class used for serialization.
+        permission_classes: list of classes to match permissions.
+    """
+
+    queryset = DomainUserPermission.objects.all()
+    serializer_class = DomainUserPermissionSerializer
+    permission_classes = [IsAdminUser]