fix(toolchain): fix toolchain session

admin_action_tools/tests/helpers.py

@@ -12,9 +12,13 @@
 
 
 class RequestSessionFactory(RequestFactory):
+    def __init__(self, session, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.session = session
+
     def request(self, **request):
         request = super().request(**request)
-        request.session = {}
+        request.session = self.session
         return request
 
 
@@ -32,7 +36,7 @@ def setUpTestData(cls):
     def setUp(self):
         cache.clear()
         self.client.force_login(self.superuser)
-        self.factory = RequestSessionFactory()
+        self.factory = RequestSessionFactory(self.client.session)
 
     def _assertManyToManyFormHtml(self, rendered_content, options, selected_ids):
         # Form data should be embedded and hidden on confirmation page

admin_action_tools/tests/unit/confirm-tool/test_confirm_actions.py

@@ -3,27 +3,22 @@
 from django.contrib.admin.options import IS_POPUP_VAR
 from django.contrib.admin.sites import AdminSite
 from django.contrib.auth.models import Permission, User
-from django.test import TestCase
 from django.urls import reverse
 
 from admin_action_tools.constants import CONFIRM_ACTION
-from admin_action_tools.tests.helpers import RequestSessionFactory
+from admin_action_tools.tests.helpers import AdminConfirmTestCase
 from tests.market.admin import ShopAdmin
 from tests.market.admin.inventory_admin import InventoryAdmin
 from tests.market.models import Inventory, Shop
 
 
-class TestConfirmActions(TestCase):
+class TestConfirmActions(AdminConfirmTestCase):
     @classmethod
     def setUpTestData(cls):
         cls.superuser = User.objects.create_superuser(  # nosec
             username="super", email="[email protected]", password="pass"
         )
 
-    def setUp(self):
-        self.client.force_login(self.superuser)
-        self.factory = RequestSessionFactory()
-
     def test_get_changelist_should_not_be_affected(self):
         response = self.client.get(reverse("admin:market_shop_changelist"))
         self.assertIsNotNone(response)

admin_action_tools/tests/unit/form-tool/test_form_action.py

@@ -1,25 +1,23 @@
 from django.contrib.auth.models import User
-from django.test import TestCase
 from django.urls import reverse
 
 from admin_action_tools.constants import CONFIRM_FORM
-from admin_action_tools.tests.helpers import RequestSessionFactory
+from admin_action_tools.tests.helpers import AdminConfirmTestCase
 from tests.factories import InventoryFactory, ShopFactory
 from tests.market.form import NoteActionForm
 
 CONFIRM_FORM_UNIQUE = f"{CONFIRM_FORM}_{NoteActionForm.__name__}"
 
 
-class TestFormAction(TestCase):
+class TestFormAction(AdminConfirmTestCase):
     @classmethod
     def setUpTestData(cls):
         cls.superuser = User.objects.create_superuser(  # nosec
             username="super", email="[email protected]", password="pass"
         )
 
     def setUp(self):
-        self.client.force_login(self.superuser)
-        self.factory = RequestSessionFactory()
+        super().setUp()
         self.shop = ShopFactory()
         self.inv = InventoryFactory(shop=self.shop, quantity=10)
 

admin_action_tools/tests/unit/test_toolchain.py

@@ -0,0 +1,22 @@
+from admin_action_tools.constants import CONFIRM_FORM
+from admin_action_tools.tests.helpers import AdminConfirmTestCase
+from admin_action_tools.toolchain import ToolChain
+from tests.market.form import NoteActionForm
+
+
+class TestToolchain(AdminConfirmTestCase):
+    def test_form_action(self):
+        request = self.factory.request()
+        name = f"toolchain{request.path}"
+        request.session[name] = {
+            "expire_at": "2012-11-02T15:14:31.000",
+            "history": ["tool1"],
+            "tool1": {"data": {"field1": True}, "metadata": {}},
+        }
+        toolchain = ToolChain(request)
+
+        # test toolchain reset
+        self.assertEqual(toolchain.get_history(), [])
+
+        # test data is save
+        self.assertEqual(request.session[name]["history"], [])

admin_action_tools/toolchain.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import functools
+from datetime import datetime, timedelta
 from typing import Dict, Optional, Tuple
 
 from django.http import HttpRequest
@@ -45,27 +46,53 @@ class ToolChain:
     def __init__(self, request: HttpRequest) -> None:
         self.request = request
         self.session = request.session
-        self.ensure_default()
-
-    def ensure_default(self):
-        self.session.setdefault("toolchain", {})
-        self.session["toolchain"].setdefault("history", [])
+        self.name = f"toolchain{request.path}"
+        self._get_data()
+        self.data.setdefault("history", [])
+
+    def _get_data(self):
+        old_data = self.session.get(self.name, {})
+        expire_at = old_data.get("expire_at")
+
+        if expire_at:
+            expire_at = datetime.fromisoformat(expire_at)
+
+        if not old_data:
+            self.data = {"expire_at": self._get_expiration()}
+            self._save()
+        elif expire_at and expire_at < datetime.now():
+            self.data = {"expire_at": self._get_expiration()}
+            self._save()
+        else:
+            self.data = old_data
+
+    def _update_expire_at(self):
+        self.data["expire_at"] = self._get_expiration()
+        self._save()
+
+    @staticmethod
+    def _get_expiration():
+        return (datetime.now() + timedelta(seconds=60)).isoformat()
+
+    def _save(self):
+        self.session[self.name] = self.data
+        self.session.modified = True
 
     def get_toolchain(self) -> Dict:
-        return self.session.get("toolchain", {})
+        return self.data
 
     def set_tool(self, tool_name: str, data: dict, metadata=None) -> None:
-        self.session["toolchain"]["history"].append(tool_name)
+        self.data["history"].append(tool_name)
         cleaned_data = self.__clean_data(data, metadata)
-        self.session["toolchain"].update({tool_name: cleaned_data})
-        self.session.modified = True
+        self.data.update({tool_name: cleaned_data})
+        self._save()
 
     def get_tool(self, tool_name: str) -> Tuple[Optional[dict], Optional[dict]]:
-        tool = self.session.get("toolchain", {}).get(tool_name, {})
+        tool = self.data.get(tool_name, {})
         return tool.get("data"), tool.get("metadata")
 
     def clear_tool_chain(self):
-        self.session.pop("toolchain", None)
+        self.session.pop(self.name, None)
 
     def is_rollback(self):
         return BACK in self.request.POST
@@ -74,17 +101,17 @@ def is_cancel(self):
         return CANCEL in self.request.POST
 
     def rollback(self):
-        tool_name = self.session["toolchain"]["history"].pop()
+        tool_name = self.data["history"].pop()
         data, _ = self.get_tool(tool_name)
-        del self.session["toolchain"][tool_name]
-        self.session.modified = True
+        del self.data[tool_name]
+        self._save()
         return data
 
     def is_first_tool(self):
-        self.ensure_default()
-        return not self.session["toolchain"]["history"]
+        return not self.data["history"]
 
     def get_next_step(self, tool_name: str) -> ToolAction:
+        self._update_expire_at()
         if self.is_cancel():
             return ToolAction.CANCEL
         if self.is_rollback():
@@ -105,5 +132,4 @@ def __clean_data(self, data, metadata):
         return {"data": data, "metadata": metadata}
 
     def get_history(self):
-        self.ensure_default()
-        return self.session["toolchain"]["history"]
+        return self.data["history"]