feat(BACKEND): allow bypass of revision protection

Add context_manager  `unprotected_content_revision` allowing to bypass revision deletion protection.

Ref: #2589

backend/tracim_backend/models/revision_protection.py

@@ -11,20 +11,22 @@
 from tracim_backend.models.data import Content
 from tracim_backend.models.data import ContentRevisionRO
 from tracim_backend.models.meta import DeclarativeBase
+from tracim_backend.models.tracim_session import TracimSession
 
 
 def prevent_content_revision_delete(
-    session: Session, flush_context: UOWTransaction, instances: [DeclarativeBase]
+    session: TracimSession, flush_context: UOWTransaction, instances: [DeclarativeBase]
 ) -> None:
-    for instance in session.deleted:
-        if isinstance(instance, ContentRevisionRO) and instance.revision_id is not None:
-            raise ContentRevisionDeleteError(
-                "ContentRevision is not deletable. "
-                + "You must make a new revision with"
-                + "is_deleted set to True. Look at "
-                + "tracim.model.new_revision context "
-                + "manager to make a new revision"
-            )
+    if not session.get_allowed_revision_deletion():
+        for instance in session.deleted:
+            if isinstance(instance, ContentRevisionRO) and instance.revision_id is not None:
+                raise ContentRevisionDeleteError(
+                    "ContentRevision is not deletable. "
+                    + "You must make a new revision with"
+                    + "is_deleted set to True. Look at "
+                    + "tracim.model.new_revision context "
+                    + "manager to make a new revision"
+                )
 
 
 class RevisionsIntegrity(object):

backend/tracim_backend/models/setup_models.py

@@ -21,6 +21,7 @@
 from tracim_backend.models.data import Content  # noqa: F401
 from tracim_backend.models.data import ContentRevisionRO  # noqa: F401
 from tracim_backend.models.meta import DeclarativeBase  # noqa: F401
+from tracim_backend.models.tracim_session import TracimSession
 
 if typing.TYPE_CHECKING:
     # INFO - G.M - 2019-05-03 - import for type-checking only, setted here to
@@ -46,13 +47,13 @@ def get_engine(app_config: "CFG", prefix="sqlalchemy.", **kwargs) -> Engine:
 
 
 def get_session_factory(engine) -> sessionmaker:
-    factory = sessionmaker(expire_on_commit=False)
+    factory = sessionmaker(expire_on_commit=False, class_=TracimSession)
     factory.configure(bind=engine)
     return factory
 
 
 def get_scoped_session_factory(engine) -> scoped_session:
-    factory = scoped_session(sessionmaker(expire_on_commit=False))
+    factory = scoped_session(sessionmaker(expire_on_commit=False, class_=TracimSession))
     factory.configure(bind=engine)
     return factory
 

backend/tracim_backend/models/tracim_session.py

@@ -0,0 +1,35 @@
+from contextlib import contextmanager
+import typing
+
+from sqlalchemy.orm import Session
+
+
+class TracimSession(Session):
+    """
+    Subclass of Sqlalchemy session to add tracim specific stuff
+    """
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self._allow_revision_deletion = False  # type: bool
+
+    def set_allowed_revision_deletion(self, value: bool) -> None:
+        self._allow_revision_deletion = value
+
+    def get_allowed_revision_deletion(self) -> bool:
+        return self._allow_revision_deletion
+
+
+@contextmanager
+def unprotected_content_revision(
+    session: TracimSession,
+) -> typing.Generator[TracimSession, None, None]:
+    """
+    allow to bypass protection on tracim revisions
+    """
+    original_allow_revision_deletion_status = session.get_allowed_revision_deletion()
+    try:
+        session.set_allowed_revision_deletion(True)
+        yield session
+    finally:
+        session.set_allowed_revision_deletion(original_allow_revision_deletion_status)

backend/tracim_backend/tests/models/test_content.py

@@ -7,12 +7,14 @@
 from sqlalchemy.sql.elements import and_
 import transaction
 
+from tracim_backend.exceptions import ContentRevisionDeleteError
 from tracim_backend.exceptions import ContentRevisionUpdateError
 from tracim_backend.models.data import ActionDescription
 from tracim_backend.models.data import Content
 from tracim_backend.models.data import ContentRevisionRO
 from tracim_backend.models.data import Workspace
 from tracim_backend.models.revision_protection import new_revision
+from tracim_backend.models.tracim_session import unprotected_content_revision
 from tracim_backend.tests.fixtures import *  # noqa:F401,F403
 
 
@@ -151,6 +153,51 @@ def test_unit__update__err__without_prepare(self, admin_user, session, content_t
             content.description = "FOO"
         # Raise ContentRevisionUpdateError because revision can't be updated
 
+    def test_unit__delete_revision__err__nominal_case(self, admin_user, session, content_type_list):
+        # file creation
+        workspace = Workspace(label="TEST_WORKSPACE_1", owner=admin_user)
+        session.add(workspace)
+        session.flush()
+        content = Content(
+            owner=admin_user,
+            workspace=workspace,
+            type=content_type_list.Page.slug,
+            label="TEST_CONTENT_1",
+            description="TEST_CONTENT_DESCRIPTION_1",
+            revision_type=ActionDescription.CREATION,
+            is_deleted=False,
+            is_archived=False,
+        )
+        session.add(content)
+        session.flush()
+        session.delete(content.revisions[0])
+        # Raise ContentRevisionDeleteError because revision can't be deleted
+        with pytest.raises(ContentRevisionDeleteError):
+            session.flush()
+
+    def test_unit__delete_revision__ok__with_unsafe_context(
+        self, admin_user, session, content_type_list
+    ):
+
+        with unprotected_content_revision(session) as unsafe_session:
+            workspace = Workspace(label="TEST_WORKSPACE_1", owner=admin_user)
+            unsafe_session.add(workspace)
+            unsafe_session.flush()
+            content = Content(
+                owner=admin_user,
+                workspace=workspace,
+                type=content_type_list.Page.slug,
+                label="TEST_CONTENT_1",
+                description="TEST_CONTENT_DESCRIPTION_1",
+                revision_type=ActionDescription.CREATION,
+                is_deleted=False,
+                is_archived=False,
+            )
+            unsafe_session.add(content)
+            unsafe_session.flush()
+            unsafe_session.delete(content.revisions[0])
+            unsafe_session.flush()
+
     def test_unit__content_depot_file__ok__nominal_case(
         self, admin_user, session, content_type_list
     ):