Fix Erroneous Audit Info Update

ixmp4/data/db/events.py

@@ -1,4 +1,5 @@
 import logging
+from contextlib import contextmanager
 from typing import TYPE_CHECKING, Any, Mapping, Sequence, cast
 
 from sqlalchemy import Connection, event, sql
@@ -20,15 +21,34 @@ class SqlaEventHandler(object):
 
     def __init__(self, backend: "SqlAlchemyBackend") -> None:
         self.backend = backend
-        event.listen(
-            self.backend.session, "do_orm_execute", self.receive_do_orm_execute
-        )
-        event.listen(
-            base.BaseModel, "before_insert", self.receive_before_insert, propagate=True
-        )
-        event.listen(
-            base.BaseModel, "before_update", self.receive_before_update, propagate=True
-        )
+        self.listeners = [
+            ((backend.session, "do_orm_execute", self.receive_do_orm_execute), {}),
+            (
+                (base.BaseModel, "before_insert", self.receive_before_insert),
+                {"propagate": True},
+            ),
+            (
+                (base.BaseModel, "before_update", self.receive_before_update),
+                {"propagate": True},
+            ),
+        ]
+        self.add_listeners()
+
+    def add_listeners(self):
+        for args, kwargs in self.listeners:
+            event.listen(*args, **kwargs)
+
+    def remove_listeners(self):
+        for args, kwargs in self.listeners:
+            if event.contains(*args):
+                event.remove(*args)
+
+    @contextmanager
+    def pause(self):
+        """Temporarily removes all event listeners for the enclosed scope."""
+        self.remove_listeners()
+        yield
+        self.add_listeners()
 
     def set_logger(self, state):
         self.logger = logging.getLogger(__name__ + "." + str(id(state)))

ixmp4/data/db/run/repository.py

@@ -149,12 +149,16 @@ def set_as_default_version(self, id: int) -> None:
             .where(
                 Run.model__id == run.model__id,
                 Run.scenario__id == run.scenario__id,
+                Run.is_default,
             )
             .values(is_default=False)
         )
 
-        self.session.execute(exc)
-        self.session.commit()
+        with self.backend.event_handler.pause():
+            # we dont want to trigger the
+            # updated_at fields for this operation.
+            self.session.execute(exc)
+            self.session.commit()
 
         run.is_default = True
         self.session.commit()

tests/data/test_run.py

@@ -110,14 +110,3 @@ def drop_audit_info(self, df):
             inplace=True,
             columns=["created_by", "created_at", "updated_by", "updated_at"],
         )
-
-    def test_audit_info(self, platform: ixmp4.Platform):
-        run = platform.backend.runs.create("Model", "Scenario")
-        platform.backend.runs.set_as_default_version(run.id)
-        platform.backend.runs.create("Model", "Scenario")
-
-        runs = platform.backend.runs.tabulate(default_only=False)
-        assert (runs["created_by"] == "@unknown").all()
-        # was updated by set_as_default_version
-        assert runs["updated_by"][0] == "@unknown"
-        assert runs["updated_by"][1] is None

tests/test_auth.py

@@ -228,6 +228,31 @@ def test_guards(self, sqlite_platform: ixmp4.Platform, user, truths):
                     with pytest.raises(Forbidden):
                         run.meta = {"meta": "test"}
 
+    def test_run_audit_info(self, db_platform: ixmp4.Platform):
+        backend = cast(SqlAlchemyBackend, db_platform.backend)
+
+        test_user = User(username="test_audit", is_verified=True, is_superuser=True)
+
+        run1 = backend.runs.create("Model 1", "Scenario 1")
+
+        backend.runs.create("Model 1", "Scenario 1")
+        backend.runs.set_as_default_version(run1.id)
+
+        with backend.auth(test_user, self.mock_manager, self.TEST_PLATFORMS[0]):
+            run3 = backend.runs.create("Model 1", "Scenario 1")
+            backend.runs.set_as_default_version(run3.id)
+
+        runs = backend.runs.tabulate(default_only=False)
+        assert runs["created_by"][0] == "@unknown"
+        assert runs["created_by"][1] == "@unknown"
+        assert runs["created_by"][2] == "test_audit"
+
+        # run1 was updated by set_as_default_version
+        # run2 was not
+        assert runs["updated_by"][0] == "@unknown"
+        assert runs["updated_by"][1] is None
+        assert runs["updated_by"][2] == "test_audit"
+
     @pytest.mark.parametrize(
         "model, platform_info, access",
         [