Merge pull request #6 from revsys/version-app

version app, test and black fixes

ak/tests/test_default_pages.py

@@ -3,7 +3,7 @@ import random
 
 
 def test_homepage(db, tp):
-    """ Ensure we can hit the homepage """
+    """Ensure we can hit the homepage"""
     # Use any page that is named 'home' otherwise use /
     url = tp.reverse("home")
     if not url:
@@ -14,21 +14,21 @@ def test_homepage(db, tp):
 
 
 def test_200_page(db, tp):
-    """ Test a 200 OK page """
+    """Test a 200 OK page"""
 
     response = tp.get("ok")
     tp.response_200(response)
 
 
 def test_403_page(db, tp):
-    """ Test a 403 error page """
+    """Test a 403 error page"""
 
     response = tp.get("forbidden")
     tp.response_403(response)
 
 
 def test_404_page(db, tp):
-    """ Test a 404 error page """
+    """Test a 404 error page"""
 
     rando = random.randint(1000, 20000)
     url = f"/this/should/not/exist/{rando}/"
@@ -40,7 +40,7 @@ def test_404_page(db, tp):
 
 
 def test_500_page(db, tp):
-    """ Test our 500 error page """
+    """Test our 500 error page"""
 
     url = tp.reverse("internal_server_error")
 

config/settings.py

@@ -77,7 +77,7 @@ INSTALLED_APPS += [
 ]
 
 # Our Apps
-INSTALLED_APPS += ["ak", "users"]
+INSTALLED_APPS += ["ak", "users", "versions"]
 
 AUTH_USER_MODEL = "users.User"
 
@@ -106,8 +106,10 @@ ROOT_URLCONF = "config.urls"
 TEMPLATES = [
     {
         "BACKEND": "django.template.backends.django.DjangoTemplates",
-        "DIRS": [str(BASE_DIR.joinpath("templates")), MACHINA_MAIN_TEMPLATE_DIR,],
-        "APP_DIRS": True,
+        "DIRS": [
+            str(BASE_DIR.joinpath("templates")),
+            MACHINA_MAIN_TEMPLATE_DIR,
+        ],
         "OPTIONS": {
             "context_processors": [
                 "django.template.context_processors.debug",
@@ -243,7 +245,9 @@ CACHES = {
 }
 
 HAYSTACK_CONNECTIONS = {
-    "default": {"ENGINE": "haystack.backends.simple_backend.SimpleEngine",},
+    "default": {
+        "ENGINE": "haystack.backends.simple_backend.SimpleEngine",
+    },
 }
 
 DEFAULT_AUTO_FIELD = "django.db.models.AutoField"

config/urls.py

@@ -10,10 +10,12 @@ from ak.views import (
     NotFoundView,
     OKView,
 )
+from versions.views import *
 
 router = routers.SimpleRouter()
 
 router.register(r"users", UserViewSet, basename="users")
+router.register(r"versions", VersionViewSet, basename="versions")
 
 urlpatterns = [
     path("", HomepageView.as_view(), name="home"),

requirements.in

@@ -30,6 +30,8 @@ pytest
 pytest-cov
 pytest-django
 pytest-xdist
+Faker
+factory_boy
 
 # Packaging
 pip-tools

requirements.txt

@@ -104,6 +104,12 @@ environs[django]==9.3.2
     # via -r ./requirements.in
 execnet==1.8.0
     # via pytest-xdist
+factory-boy==3.2.1
+    # via -r ./requirements.in
+faker==9.8.2
+    # via
+    #   -r ./requirements.in
+    #   factory-boy
 fs==2.4.13
     # via django-bakery
 gevent==21.1.2
@@ -169,7 +175,9 @@ pytest==6.2.4
     #   pytest-forked
     #   pytest-xdist
 python-dateutil==2.8.1
-    # via botocore
+    # via
+    #   botocore
+    #   faker
 python-dotenv==0.17.1
     # via environs
 python-json-logger==2.0.1
@@ -200,6 +208,8 @@ structlog==21.1.0
     # via -r ./requirements.in
 tabulate==0.8.9
     # via interrogate
+text-unidecode==1.3
+    # via faker
 toml==0.10.2
     # via
     #   black

users/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
 from django.contrib.auth.admin import UserAdmin
-from django.utils.translation import ugettext_lazy as _
+from django.utils.translation import gettext_lazy as _
 
 from .models import User
 

users/factories.py

@@ -1,10 +1,18 @@
 import factory
 
+import django.contrib.auth.models as auth_models
 from django.utils import timezone
 
 from .models import User
 
 
+class VersionGroupFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = auth_models.Group
+
+    name = "version_manager"
+
+
 class UserFactory(factory.django.DjangoModelFactory):
     email = factory.Sequence(lambda n: "user%s@example.com" % n)
     first_name = factory.Sequence(lambda n: "User%s Bob" % n)
@@ -18,6 +26,15 @@ class UserFactory(factory.django.DjangoModelFactory):
         model = User
         django_get_or_create = ("email",)
 
+    @factory.post_generation
+    def groups(self, create, extracted, **kwargs):
+        if not create:
+            return
+
+        if extracted:
+            for group in extracted:
+                self.groups.add(group)
+
 
 class StaffUserFactory(UserFactory):
     is_staff = True

users/management/commands/create_groups.py

@@ -0,0 +1,20 @@
+from django.contrib.auth.models import Group, Permission
+from django.core.management import BaseCommand
+from django.contrib.contenttypes.models import ContentType
+from versions.models import Version
+
+
+class Command(BaseCommand):
+    def __init__(self, *args, **kwargs):
+        super(Command, self).__init__(*args, **kwargs)
+
+    help = "Create default groups"
+
+    def handle(self, *args, **options):
+        Group.objects.get(name="version_manager").delete()
+        # Code to add permission to group ???
+        # ct = ContentType.objects.get_for_model(Version)
+
+        # Now what - Say I want to add 'Can add version' permission to new_group?
+        Permission.objects.get(codename="can_add_version").delete()
+        # new_group.permissions.add(permission)

users/migrations/0001_initial.py

@@ -113,7 +113,9 @@ class Migration(migrations.Migration):
                 "verbose_name": "user",
                 "verbose_name_plural": "users",
             },
-            managers=[("objects", users.models.UserManager()),],
+            managers=[
+                ("objects", users.models.UserManager()),
+            ],
         ),
         migrations.CreateModel(
             name="LastSeen",

users/migrations/0005_auto_20211121_0908.py

@@ -0,0 +1,32 @@
+# Generated by Django 3.2.2 on 2021-11-21 09:08
+
+from django.db import migrations
+from django.contrib.auth.models import Group, Permission
+from django.core.management import BaseCommand
+from django.contrib.contenttypes.models import ContentType
+from versions.models import Version
+
+
+def gen_version_manager_group(apps, schema_editor):
+    new_group, created = Group.objects.get_or_create(name="version_manager")
+    # Code to add permission to group ???
+    ct = ContentType.objects.get_for_model(Version)
+
+    # Now what - Say I want to add 'Can add version' permission to new_group?
+    permission, created = Permission.objects.get_or_create(
+        codename="can_add_version", name="Can add version", content_type=ct
+    )
+    new_group.permissions.add(permission)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("users", "0004_auto_20211105_0915"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            gen_version_manager_group, reverse_code=migrations.RunPython.noop
+        ),
+    ]

users/models.py

@@ -9,7 +9,7 @@ from django.contrib.auth.models import (
     BaseUserManager,
 )
 from django.core.mail import send_mail
-from django.utils.translation import ugettext_lazy as _
+from django.utils.translation import gettext_lazy as _
 from django.utils import timezone
 
 
@@ -124,7 +124,7 @@ class BaseUser(AbstractBaseUser, PermissionsMixin):
         send_mail(subject, message, from_email, [self.email], **kwargs)
 
     def save(self, *args, **kwargs):
-        """ Ensure email is always lower case """
+        """Ensure email is always lower case"""
         self.email = self.email.lower()
 
         return super().save(*args, **kwargs)
@@ -150,7 +150,9 @@ class LastSeen(models.Model):
     """
 
     user = models.OneToOneField(
-        settings.AUTH_USER_MODEL, related_name="last_seen", on_delete=models.CASCADE,
+        settings.AUTH_USER_MODEL,
+        related_name="last_seen",
+        on_delete=models.CASCADE,
     )
     at = models.DateTimeField(default=timezone.now)
 
@@ -165,7 +167,7 @@ class LastSeen(models.Model):
 
 @receiver(post_save, sender=User)
 def create_last_seen_for_user(sender, instance, created, raw, **kwargs):
-    """ Create LastSeen row when a User is created """
+    """Create LastSeen row when a User is created"""
     if raw:
         return
 

users/views.py

@@ -17,7 +17,7 @@ class UserViewSet(viewsets.ModelViewSet):
     permission_classes = [CustomUserPermissions]
 
     def get_serializer_class(self):
-        """ Pick the right serializer based on the user """
+        """Pick the right serializer based on the user"""
         if self.request.user.is_staff or self.request.user.is_superuser:
             return FullUserSerializer
         else:

versions/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

versions/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class VersionsConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "versions"

versions/factories.py

@@ -0,0 +1,12 @@
+import factory
+
+from .models import Version
+
+
+class VersionFactory(factory.django.DjangoModelFactory):
+    name = factory.Sequence(lambda n: "version%s" % n)
+    file = factory.Faker("file_name")
+    release_date = factory.Faker("date_object")
+
+    class Meta:
+        model = Version

versions/migrations/0001_initial.py

@@ -0,0 +1,34 @@
+# Generated by Django 3.2.2 on 2021-11-21 08:28
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="Version",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("name", models.CharField(help_text="Version name", max_length=256)),
+                (
+                    "checksum",
+                    models.CharField(default=None, max_length=64, unique=True),
+                ),
+                ("file", models.FileField(upload_to="uploads/")),
+                ("release_date", models.DateField()),
+            ],
+        ),
+    ]

versions/models.py

@@ -0,0 +1,18 @@
+import hashlib
+from django.db import models
+
+# Create your models here.
+
+
+class Version(models.Model):
+    name = models.CharField(
+        max_length=256, null=False, blank=False, help_text="Version name"
+    )
+    checksum = models.CharField(max_length=64, unique=True, default=None)
+    file = models.FileField(upload_to="uploads/")
+    release_date = models.DateField(auto_now=False, auto_now_add=False)
+
+    def save(self, *args, **kwargs):
+        if self.checksum is None:
+            self.checksum = hashlib.sha256(self.name.encode("utf-8")).hexdigest()
+        super().save(*args, **kwargs)

versions/permissions.py

@@ -0,0 +1,17 @@
+from rest_framework import permissions
+
+
+def is_version_manager(user):
+    return user.groups.filter(name="version_manager").exists()
+
+
+class SuperUserOrVersionManager(permissions.BasePermission):
+    def has_permission(self, request, view):
+        if request.user.is_superuser:
+            return True
+
+        if is_version_manager(request.user):
+            return True
+
+        if request.method in permissions.SAFE_METHODS:
+            return True

versions/serializers.py

@@ -0,0 +1,8 @@
+from rest_framework import serializers
+from versions.models import Version
+
+
+class VersionSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Version
+        fields = ["id", "name", "checksum", "file", "release_date"]

versions/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

versions/tests/test_api.py

@@ -0,0 +1,157 @@
+from django.db import transaction
+from django.db import IntegrityError
+import tempfile
+from django.urls import reverse
+from rest_framework.test import APIClient
+from test_plus.test import TestCase
+
+from users.factories import UserFactory, SuperUserFactory, VersionGroupFactory
+from versions.factories import VersionFactory
+
+from PIL import Image
+
+
+class VersionViewTests(TestCase):
+    client_class = APIClient
+
+    def setUp(self):
+        self.user = UserFactory()
+        self.group_user = UserFactory.create(groups=(VersionGroupFactory.create(),))
+        self.super_user = SuperUserFactory()
+        self.version_manager = UserFactory()
+        self.version = VersionFactory()
+
+    def test_list_version(self):
+        """
+        Tests with a regular user
+        """
+        # Does API work without auth?
+        response = self.get("versions-list")
+        self.response_403(response)
+
+        # Does API work with auth?
+        with self.login(self.user):
+            response = self.get("versions-list")
+            self.response_200(response)
+            self.assertEqual(len(response.data), 1)
+            # Are non-staff shown/hidden the right fields?
+            self.assertIn("name", response.data[0])
+            self.assertIn("checksum", response.data[0])
+
+    def test_create(self):
+        image = Image.new("RGB", (100, 100))
+
+        tmp_file = tempfile.NamedTemporaryFile(suffix=".jpg")
+        image.save(tmp_file)
+
+        tmp_file.seek(0)
+
+        payload = {
+            "file": tmp_file,
+            "name": "First Version",
+            "release_date": "2021-01-01",
+        }
+
+        # Does API work without auth?
+        response = self.client.post(
+            reverse("versions-list"), files=payload, format="multipart"
+        )
+        self.response_403(response)
+
+        tmp_file.seek(0)
+        # Does API work with normal user?
+        with self.login(self.user):
+            response = self.client.post(
+                reverse("versions-list"), data=payload, format="multipart"
+            )
+            self.response_403(response)
+
+        tmp_file.seek(0)
+        # Does API work with super user?
+        with self.login(self.super_user):
+            try:
+                with transaction.atomic():
+                    response = self.client.post(
+                        reverse("versions-list"), data=payload, format="multipart"
+                    )
+                    self.response_201(response)
+            except IntegrityError:
+                pass
+
+        tmp_file.seek(0)
+        # Does API work with version_manager user?
+        with self.login(self.group_user):
+            try:
+                with transaction.atomic():
+                    response = self.client.post(
+                        reverse("versions-list"), data=payload, format="multipart"
+                    )
+                    self.response_201(response)
+            except IntegrityError:
+                pass
+
+    def test_delete(self):
+        url = reverse("versions-detail", kwargs={"pk": self.version.pk})
+
+        # Does this API work without auth?
+        response = self.client.delete(url, format="json")
+        self.response_403(response)
+
+        # Does this API wotk with non-staff user?
+        with self.login(self.user):
+            response = self.client.delete(url, format="json")
+            self.response_403(response)
+
+        # Does this API work with super user?
+        with self.login(self.super_user):
+            response = self.client.delete(url, format="json")
+            self.assertEqual(response.status_code, 204)
+
+            # Confirm object is gone
+            response = self.get(url)
+            self.response_404(response)
+
+    def test_update(self):
+        old_name = self.version.name
+        url = reverse("versions-detail", kwargs={"pk": self.version.pk})
+
+        image = Image.new("RGB", (100, 100))
+
+        tmp_file = tempfile.NamedTemporaryFile(suffix=".jpg")
+        image.save(tmp_file)
+
+        tmp_file.seek(0)
+
+        payload = {"file": tmp_file, "name": "Second Version"}
+
+        # Does API work without auth?
+        response = self.client.post(url, files=payload, format="multipart")
+        self.response_403(response)
+
+        tmp_file.seek(0)
+        # Does API work with normal user?
+        with self.login(self.user):
+            response = self.client.patch(url, data=payload, format="multipart")
+            self.response_403(response)
+
+        tmp_file.seek(0)
+        # Does API work with super user?
+        with self.login(self.super_user):
+            try:
+                with transaction.atomic():
+                    response = self.client.patch(url, data=payload, format="multipart")
+                    self.response_200(response)
+                    self.assertNotEqual(old_name, response.data["name"])
+            except IntegrityError:
+                pass
+
+        tmp_file.seek(0)
+        # Does API work with version_manager user?
+        with self.login(self.group_user):
+            try:
+                with transaction.atomic():
+                    response = self.client.patch(url, data=payload, format="multipart")
+                    self.response_200(response)
+                    self.assertNotEqual(old_name, response.data["name"])
+            except IntegrityError:
+                pass

versions/views.py

@@ -0,0 +1,14 @@
+from rest_framework import viewsets
+from rest_framework.response import Response
+from rest_framework import permissions
+from versions.permissions import SuperUserOrVersionManager
+
+from versions.models import Version
+from versions.serializers import VersionSerializer
+
+
+class VersionViewSet(viewsets.ModelViewSet):
+    model = Version
+    queryset = Version.objects.all()
+    serializer_class = VersionSerializer
+    permission_classes = [permissions.IsAuthenticated, SuperUserOrVersionManager]