Testers for session lifetime

src/flask_session/base.py

@@ -225,7 +225,6 @@ def should_set_storage(self, app: Flask, session: ServerSideSession) -> bool:
         the ``SESSION_REFRESH_EACH_REQUEST`` config is true, the session is
         always set to storage. In the second case, this means refreshing the
         storage expiry even if the session has not been modified.
-
         .. versionadded:: 0.7.0
         """
 

src/flask_session/dynamodb/dynamodb.py

@@ -144,10 +144,9 @@ def _create_table(self):
     def _retrieve_session_data(self, store_id: str) -> Optional[dict]:
         # Get the saved session (document) from the database
         document = self.store.get_item(Key={"id": store_id}).get("Item")
-        session_is_not_expired = Decimal(datetime.utcnow().timestamp()) <= document.get(
+        if document and Decimal(datetime.utcnow().timestamp()) <= document.get(
             "expiration"
-        )
-        if document and session_is_not_expired:
+        ):
             serialized_session_data = want_bytes(document.get("val").value)
             return self.serializer.loads(serialized_session_data)
         return None

tests/abs_test.py

@@ -0,0 +1,68 @@
+from abc import ABC, abstractmethod
+from contextlib import contextmanager
+from datetime import datetime, timezone
+
+import time
+from tests.utils import session_permanent, session_refresh_each_request
+
+
+class ABSTestSession(ABC):
+
+    @contextmanager
+    def setup_filesystem(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    def retrieve_stored_session(self, key, app):
+        raise NotImplementedError
+
+    @session_permanent
+    @session_refresh_each_request
+    def test_default(self, app_utils,_session_permanent,
+                                     _session_refresh_each_request):
+        raise NotImplementedError
+
+    def _default_test(self, app_utils, app):
+        app_utils.test_session(app)
+        app_utils.test_regenerate_session(app)
+
+        # Check if the session is stored in the filesystem
+        cookie = app_utils.test_session_with_cookie(app)
+        session_id = cookie.split(";")[0].split("=")[1]
+        stored_session = self.retrieve_stored_session(f"session:{session_id}", app)
+        assert stored_session.get("value") == "44"
+
+    @session_permanent
+    @session_refresh_each_request
+    def test_lifetime(self, app_utils,
+                                 _session_permanent,
+                                 _session_refresh_each_request):
+        raise NotImplementedError
+
+    def _test_lifetime(self, app, _session_permanent):
+        client = app.test_client()
+
+        response = client.post("/set", data={"value": "44"})
+
+        if _session_permanent is False:
+            assert "Expires" not in response.headers.getlist("Set-Cookie")[0]
+            return
+
+        datetime_expires_by_cookies = datetime.strptime(
+            response.headers.getlist("Set-Cookie")[0].split(";")[1].split("=")[
+                1],
+            "%a, %d %b %Y %H:%M:%S GMT")
+        assert datetime_expires_by_cookies.replace(
+            tzinfo=timezone.utc) > datetime.utcnow().replace(
+            tzinfo=timezone.utc)
+        session_id = client.get("/get-session-id").data
+        assert self.retrieve_stored_session(
+            f"session:{session_id.decode('utf-8')}", app).get("value") == "44"
+        time.sleep(5)
+        assert not self.retrieve_stored_session(
+            f"session:{session_id.decode('utf-8')}", app)
+        assert client.get("/get-session-id").data != session_id
+        assert datetime_expires_by_cookies.replace(
+            tzinfo=timezone.utc) < datetime.utcnow().replace(
+            tzinfo=timezone.utc)
+

tests/conftest.py

@@ -14,7 +14,7 @@ def create_app(self, config_dict=None):
             if config_dict:
                 app.config.update(config_dict)
             app.config["SESSION_SERIALIZATION_FORMAT"] = "json"
-            app.config["SESSION_PERMANENT"] = False
+            # app.config["SESSION_PERMANENT"] = False
 
             @app.route("/", methods=["GET"])
             def app_hello():
@@ -35,10 +35,19 @@ def app_del():
                 del flask.session["value"]
                 return "value deleted"
 
+            @app.route("/regenerate", methods=["POST"])
+            def app_regenerate():
+                app.session_interface.regenerate(flask.session)
+                return "value regenerated"
+
             @app.route("/get")
             def app_get():
                 return flask.session.get("value", "no value set")
 
+            @app.route("/get-session-id")
+            def app_get_session():
+                return flask.session.sid
+
             flask_session.Session(app)
             return app
 
@@ -51,31 +60,51 @@ def test_session(self, app):
             # Check if the Vary header is not set
             rv = client.get("/")
             assert "Vary" not in rv.headers
+            assert client.get("/get-session-id").data != client.get("/get-session-id").data
 
             # Set a value and check it
+
             assert client.post("/set", data={"value": "42"}).data == b"value set"
             assert client.get("/get").data == b"42"
 
             # Check if the Vary header is set
             rv = client.get("/get")
             assert rv.headers["Vary"] == "Cookie"
+            assert client.get("/get-session-id").data == client.get(
+                "/get-session-id").data
 
             # Modify and delete the value
             assert client.post("/modify", data={"value": "43"}).data == b"value set"
             assert client.get("/get").data == b"43"
             assert client.post("/delete").data == b"value deleted"
             assert client.get("/get").data == b"no value set"
 
+        def test_regenerate_session(self, app):
+            client = app.test_client()
+            assert client.get("/get-session-id").data != client.get(
+                "/get-session-id").data
+            assert client.post("/set", data={"value": "42"}).data == b"value set"
+            assert client.get("/get").data == b"42"
+            assert client.get("/get-session-id").data == client.get(
+                "/get-session-id").data
+            original = client.get("/get-session-id").data
+            rg = client.post("/regenerate").data
+
+            assert client.get("/get-session-id").data != original
+
         def test_session_with_cookie(self, app):
             client = app.test_client()
 
             # Access cookies from the response to cross check with the stored session
             response = client.post("/set", data={"value": "44"})
             session_cookie = None
             for cookie in response.headers.getlist("Set-Cookie"):
-                if "session=" in cookie:
+                if app.config['SESSION_COOKIE_NAME'] + '=' in cookie:
                     session_cookie = cookie
                     break
+            assert session_cookie.split('; ')[0].replace(
+                app.config['SESSION_COOKIE_NAME'] + '=', '') == client.get(
+                "/get-session-id").data.decode('utf-8')
             assert session_cookie is not None, "Session cookie was not set."
             return session_cookie
 

tests/test_cachelib.py

@@ -1,13 +1,16 @@
 import os
 import shutil
 from contextlib import contextmanager
+from datetime import timedelta
 
 import flask
+from tests.utils import session_permanent, session_refresh_each_request
 from cachelib.file import FileSystemCache
 from flask_session.cachelib import CacheLibSession
+from tests.abs_test import ABSTestSession
 
 
-class TestCachelibSession:
+class TestCachelibSession(ABSTestSession):
     session_dir = "testing_session_storage"
 
     @contextmanager
@@ -22,11 +25,16 @@ def setup_filesystem(self):
     def retrieve_stored_session(self, key, app):
         return app.session_interface.cache.get(key)
 
-    def test_filesystem_default(self, app_utils):
+    @session_permanent
+    @session_refresh_each_request
+    def test_default(self, app_utils,_session_permanent,
+                                     _session_refresh_each_request):
         app = app_utils.create_app(
             {
                 "SESSION_TYPE": "cachelib",
                 "SESSION_SERIALIZATION_FORMAT": "json",
+                "SESSION_PERMANENT": _session_permanent,
+                "SESSION_REFRESH_EACH_REQUEST": _session_refresh_each_request,
                 "SESSION_CACHELIB": FileSystemCache(
                     threshold=500, cache_dir=self.session_dir
                 ),
@@ -39,10 +47,30 @@ def test_filesystem_default(self, app_utils):
                 flask.session,
                 CacheLibSession,
             )
-            app_utils.test_session(app)
+            self._default_test(app_utils, app)
 
-            # Check if the session is stored in the filesystem
-            cookie = app_utils.test_session_with_cookie(app)
-            session_id = cookie.split(";")[0].split("=")[1]
-            stored_session = self.retrieve_stored_session(f"session:{session_id}", app)
-            assert stored_session.get("value") == "44"
+    @session_permanent
+    @session_refresh_each_request
+    def test_lifetime(self, app_utils,
+                                 _session_permanent,
+                                 _session_refresh_each_request):
+        app = app_utils.create_app(
+            {
+                "SESSION_TYPE": "cachelib",
+                "SESSION_SERIALIZATION_FORMAT": "json",
+                "SESSION_PERMANENT": _session_permanent,
+                "SESSION_REFRESH_EACH_REQUEST": _session_refresh_each_request,
+                "SESSION_CACHELIB": FileSystemCache(
+                    threshold=500, cache_dir=self.session_dir
+                ),
+                "PERMANENT_SESSION_LIFETIME": timedelta(seconds=4),
+            }
+        )
+
+        # Should be using FileSystem
+        with self.setup_filesystem(), app.test_request_context():
+            assert isinstance(
+                flask.session,
+                CacheLibSession,
+            )
+            self._test_lifetime(app, _session_permanent)

tests/test_dynamodb.py

@@ -1,13 +1,21 @@
+import json
+from decimal import Decimal
 from contextlib import contextmanager
+from datetime import timedelta, datetime
 
 import boto3
 import flask
 import pytest
 from flask_session.defaults import Defaults
 from flask_session.dynamodb import DynamoDBSession
+from itsdangerous import want_bytes
 
+from tests.utils import session_permanent, session_refresh_each_request
 
-class TestDynamoDBSession:
+from tests.abs_test import ABSTestSession
+
+
+class TestDynamoDBSession(ABSTestSession):
     """This requires package: boto3"""
 
     @contextmanager
@@ -41,25 +49,20 @@ def setup_dynamodb(self):
                     }
                 )
 
-    def test_dynamodb_default(self, app_utils):
-        with self.setup_dynamodb():
-            app = app_utils.create_app(
-                {
-                    "SESSION_TYPE": "dynamodb",
-                    "SESSION_DYNAMODB": self.client,
-                }
-            )
-
-            with app.test_request_context():
-                assert isinstance(flask.session, DynamoDBSession)
-                app_utils.test_session(app)
+    def retrieve_stored_session(self, key, app):
+        document = self.store.get_item(Key={"id": key}).get("Item")
+        if document and Decimal(datetime.utcnow().timestamp()) <= document.get(
+            "expiration"
+        ):
+            serialized_session_data = want_bytes(document.get("val").value)
+            return json.loads(want_bytes(serialized_session_data)) if serialized_session_data else {}
+        return None
 
     def test_dynamodb_with_existing_table(self, app_utils):
         """
         Setting the SESSION_DYNAMODB_TABLE_EXISTS to True for an
         existing table shouldn't change anything.
         """
-
         with self.setup_dynamodb():
             app = app_utils.create_app(
                 {
@@ -71,7 +74,7 @@ def test_dynamodb_with_existing_table(self, app_utils):
 
             with app.test_request_context():
                 assert isinstance(flask.session, DynamoDBSession)
-                app_utils.test_session(app)
+                self._default_test(app_utils, app)
 
     def test_dynamodb_with_existing_table_fails_if_table_doesnt_exist(self, app_utils):
         """Accessing a non-existent table should result in problems."""
@@ -92,4 +95,46 @@ def test_dynamodb_with_existing_table_fails_if_table_doesnt_exist(self, app_util
         )
         with app.test_request_context(), pytest.raises(AssertionError):
             assert isinstance(flask.session, DynamoDBSession)
-            app_utils.test_session(app)
+            self._default_test(app_utils, app)
+
+
+    @session_permanent
+    @session_refresh_each_request
+    def test_default(self, app_utils, _session_permanent,
+                     _session_refresh_each_request):
+        with self.setup_dynamodb():
+            app = app_utils.create_app(
+                {
+                    "SESSION_TYPE": "dynamodb",
+                    "SESSION_DYNAMODB": self.client,
+                    "SESSION_PERMANENT": _session_permanent,
+                    "SESSION_REFRESH_EACH_REQUEST": _session_refresh_each_request,
+
+                }
+            )
+
+            with app.test_request_context():
+                assert isinstance(flask.session, DynamoDBSession)
+                self._default_test(app_utils, app)
+
+    @session_permanent
+    @session_refresh_each_request
+    def test_lifetime(self, app_utils,
+                      _session_permanent,
+                      _session_refresh_each_request):
+        with self.setup_dynamodb():
+
+            app = app_utils.create_app(
+                {
+                    "SESSION_TYPE": "dynamodb",
+                    "SESSION_DYNAMODB": self.client,
+                    "SESSION_PERMANENT": _session_permanent,
+                    "SESSION_REFRESH_EACH_REQUEST": _session_refresh_each_request,
+                    "PERMANENT_SESSION_LIFETIME": timedelta(seconds=4),
+
+                }
+            )
+
+            with app.test_request_context():
+                assert isinstance(flask.session, DynamoDBSession)
+                self._test_lifetime(app, _session_permanent)