✅[#46] add session profile tests

open_api_framework/admin.py

@@ -10,10 +10,10 @@
 class SessionProfileAdmin(admin.ModelAdmin):
     list_display = ["session_key", "user", "exists"]
 
-    def __init__(self, model, admin_site):
-        engine = import_module(settings.SESSION_ENGINE)
-        self.SessionStore = engine.SessionStore
-        super().__init__(model, admin_site)
+    @property
+    def SessionStore(self):
+
+        return import_module(settings.SESSION_ENGINE).SessionStore
 
     def get_queryset(self, request):
         qs = super().get_queryset(request)
@@ -29,14 +29,14 @@ def has_change_permission(self, request, obj=None):
     def exists(self, obj):
         return self.SessionStore().exists(obj.session_key)
 
-    def expiration_time(self, obj):
-
-        session = self.SessionStore(obj.session_key)
-
-        if session.exists(obj.session_key):
-            return session.get_expiry_date()
-        return None
-
     def delete_model(self, request, obj):
         self.SessionStore(obj.session_key).flush()
         super().delete_model(request, obj)
+
+    def delete_queryset(self, request, queryset):
+
+        session_keys = list(queryset.values_list("session_key", flat=True))
+        for session_key in session_keys:
+            self.SessionStore(session_key).flush()
+
+        super().delete_queryset(request, queryset)

open_api_framework/conf/base.py

@@ -243,7 +243,6 @@
     "django.middleware.csrf.CsrfViewMiddleware",
     "django.contrib.auth.middleware.AuthenticationMiddleware",
     "maykin_2fa.middleware.OTPMiddleware",
-    "mozilla_django_oidc_db.middleware.SessionRefresh",
     "django.contrib.messages.middleware.MessageMiddleware",
     "django.middleware.clickjacking.XFrameOptionsMiddleware",
     "axes.middleware.AxesMiddleware",

pyproject.toml

@@ -73,6 +73,7 @@ tests = [
     "isort",
     "black",
     "flake8",
+    "factory-boy",
 ]
 coverage = [
     "pytest-cov",

testapp/settings.py

@@ -80,3 +80,5 @@
 # These are excluded from generate_envvar_docs test by their group
 VARIABLE_TO_BE_EXCLUDED = config("VARIABLE_TO_BE_EXCLUDED1", "foo", group="Excluded")
 VARIABLE_TO_BE_EXCLUDED = config("VARIABLE_TO_BE_EXCLUDED2", "bar", group="Excluded")
+
+SESSION_ENGINE = "django.contrib.sessions.backends.db"

tests/conftest.py

@@ -1,6 +1,17 @@
-# import pytest
+from importlib import import_module
 
+from django.conf import settings as django_settings
 
-# @pytest.fixture
-# def some_fixture(request):
-#     return "foo"
+import pytest
+
+
+@pytest.fixture
+def cache_session_store(settings):
+    settings.SESSION_ENGINE = "django.contrib.sessions.backends.cache"
+    return import_module(django_settings.SESSION_ENGINE).SessionStore
+
+
+@pytest.fixture
+def db_session_store(settings):
+    settings.SESSION_ENGINE = "django.contrib.sessions.backends.db"
+    return import_module(django_settings.SESSION_ENGINE).SessionStore

tests/factories.py

@@ -0,0 +1,24 @@
+from importlib import import_module
+
+from django.conf import settings
+
+import factory
+import factory.fuzzy
+from sessionprofile.models import SessionProfile
+
+
+class SessionProfileFactory(factory.django.DjangoModelFactory):
+
+    session_key = factory.fuzzy.FuzzyText(length=40)
+
+    class Meta:
+        model = SessionProfile
+
+    @classmethod
+    def create(cls, **kwargs):
+        SessionStore = import_module(settings.SESSION_ENGINE).SessionStore
+
+        instance = super().create(**kwargs)
+        SessionStore(instance.session_key).save(True)
+
+        return instance

tests/test_admin.py

@@ -0,0 +1,162 @@
+from django.contrib.sessions.models import Session
+from django.urls import reverse
+
+import pytest
+from pytest_django.fixtures import admin_user
+from sessionprofile.models import SessionProfile
+
+from .factories import SessionProfileFactory
+
+
+@pytest.fixture
+def session_changelist_url():
+    return reverse("admin:sessionprofile_sessionprofile_changelist")
+
+
+def test_session_profile_sanity(client, admin_user, session_changelist_url):
+
+    client.force_login(admin_user)
+    response = client.get(session_changelist_url)
+    assert response.status_code == 200
+
+    assert SessionProfile.objects.count() == 1
+
+    session = SessionProfile.objects.get()
+    assert client.session.session_key == session.session_key
+
+
+admin_user2 = admin_user
+
+
+def test_only_session_profile_of_user_shown(
+    client, admin_user, django_user_model, session_changelist_url
+):
+
+    other_admin = django_user_model.objects.create_superuser("garry")
+
+    client.force_login(other_admin)
+    response = client.get(session_changelist_url)
+    assert response.status_code == 200
+
+    client.force_login(admin_user)
+    response = client.get(session_changelist_url)
+    assert response.status_code == 200
+
+    # two sessions, one for each user
+    assert SessionProfile.objects.count() == 2
+
+    # Session created after response, needs to be called again
+    response = client.get(session_changelist_url)
+
+    admin_user_session = SessionProfile.objects.get(user=admin_user)
+    assert admin_user_session.session_key in response.content.decode()
+
+    other_user_session = SessionProfile.objects.get(user=other_admin)
+    assert other_user_session.session_key not in response.content.decode()
+
+
+def test_delete_with_session_db_backend(
+    client, admin_user, session_changelist_url, db_session_store
+):
+    client.force_login(admin_user)
+
+    session = SessionProfileFactory(user=admin_user)
+
+    assert SessionProfile.objects.count() == 1
+    # sesison created by login
+    assert Session.objects.count() == 2
+    assert db_session_store().exists(session.session_key)
+
+    url = reverse("admin:sessionprofile_sessionprofile_delete", args=[session.pk])
+
+    response = client.post(url, {"post": "yes"})
+    assert response.status_code == 302
+
+    # new session saved upon request
+    assert SessionProfile.objects.count() == 1
+    assert SessionProfile.objects.count() != session
+    assert Session.objects.count() == 1
+    assert not db_session_store().exists(session.session_key)
+
+
+def test_delete_with_session_cache_backend(
+    client, admin_user, session_changelist_url, cache_session_store
+):
+    client.force_login(admin_user)
+
+    session = SessionProfileFactory(user=admin_user)
+
+    assert SessionProfile.objects.count() == 1
+    assert Session.objects.count() == 0
+    assert cache_session_store().exists(session.session_key)
+
+    url = reverse("admin:sessionprofile_sessionprofile_delete", args=[session.pk])
+
+    response = client.post(url, {"post": "yes"})
+    assert response.status_code == 302
+
+    # new session saved upon request
+    assert SessionProfile.objects.count() == 1
+    assert SessionProfile.objects.count() != session
+    assert Session.objects.count() == 0
+    assert not cache_session_store().exists(session.session_key)
+
+
+def test_delete_action_with_session_db_backend(
+    client, admin_user, session_changelist_url, db_session_store
+):
+    client.force_login(admin_user)
+    sessions = SessionProfileFactory.create_batch(5, user=admin_user)
+
+    # one created from user login
+    assert Session.objects.count() == 6
+    assert SessionProfile.objects.count() == 5
+
+    session_keys = [session.session_key for session in sessions]
+    for session_key in session_keys:
+        assert db_session_store().exists(session_key)
+
+    response = client.post(
+        session_changelist_url,
+        {"action": "delete_selected", "_selected_action": session_keys, "post": "yes"},
+    )
+    assert response.status_code == 302
+
+    # one is created as the post request is sent
+    assert SessionProfile.objects.count() == 1
+    assert Session.objects.count() == 1
+
+    for session_key in session_keys:
+        assert not db_session_store().exists(session_key)
+
+
+def test_delete_action_with_session_cache_backend(
+    client, admin_user, session_changelist_url, cache_session_store
+):
+
+    client.force_login(admin_user)
+    sessions = SessionProfileFactory.create_batch(5, user=admin_user)
+
+    # no db sessions are created
+    assert Session.objects.count() == 0
+    assert SessionProfile.objects.count() == 5
+
+    session_keys = [session.session_key for session in sessions]
+
+    # sessions are created
+    for session_key in session_keys:
+        assert cache_session_store().exists(session_key)
+
+    response = client.post(
+        session_changelist_url,
+        {"action": "delete_selected", "_selected_action": session_keys, "post": "yes"},
+    )
+    assert response.status_code == 302
+
+    # one is created as the post request is sent
+    assert SessionProfile.objects.count() == 1
+    assert Session.objects.count() == 0
+
+    # sessions should be deleted
+    for session_key in session_keys:
+        assert not cache_session_store().exists(session_key)