From 4e891bece097a596a01a221c6d5b074836517ca9 Mon Sep 17 00:00:00 2001 From: svader0 Date: Wed, 19 Aug 2026 11:50:36 -0500 Subject: [PATCH] fix(base_models): make the model save hook sequence atomic pre_save_logic() is allowed to write related rows, and the base model ran it before full_clean(). A save rejected by validation therefore kept the hook's writes while its own row was never updated, so the two disagreed from then on. Nothing rolled that back: the default database does not use ATOMIC_REQUESTS, and the callers that catch the error cannot undo a write another model made. Wrap the hook, the validation and the save in one transaction. The savepoint is taken only when a subclass overrides pre_save_logic, so the models that do not (Finding among them) keep their savepoint-free save and the import query-count contract is unchanged. --- dojo/base_models/base.py | 32 +++++---- unittests/test_location_save_atomicity.py | 84 +++++++++++++++++++++++ 2 files changed, 103 insertions(+), 13 deletions(-) create mode 100644 unittests/test_location_save_atomicity.py diff --git a/dojo/base_models/base.py b/dojo/base_models/base.py index f9bd658162..d77affa35f 100644 --- a/dojo/base_models/base.py +++ b/dojo/base_models/base.py @@ -2,6 +2,7 @@ import logging from typing import TypeVar +from django.db import transaction from django.db.models import DateTimeField, Manager, Model, QuerySet from django.utils.translation import gettext_lazy as _ @@ -60,19 +61,24 @@ def save(self, *args: list, skip_validation: bool | None = None, **kwargs: dict) from dojo.location.feature import locations_enabled # noqa: PLC0415 skip_validation = not locations_enabled() - # Run the pre save logic, if enabled - self.pre_save_logic() - # Call the validations - if not skip_validation: - try: - self.full_clean() - except Exception: - self.print_all_fields() - raise - # Run the post save logic, if enabled - self.post_save_logic() - # Call the base save method to save the model to the database - super().save(*args, **kwargs) + # pre_save_logic() may commit related rows, so anything raised after it has to take + # those writes back out. Subclasses that do not override the hook have nothing to + # undo, and a savepoint on every save is measurable on the import path. + rolls_back = type(self).pre_save_logic is not BaseModelWithoutTimeMeta.pre_save_logic + with transaction.atomic() if rolls_back else contextlib.nullcontext(): + # Run the pre save logic, if enabled + self.pre_save_logic() + # Call the validations + if not skip_validation: + try: + self.full_clean() + except Exception: + self.print_all_fields() + raise + # Run the post save logic, if enabled + self.post_save_logic() + # Call the base save method to save the model to the database + super().save(*args, **kwargs) def pre_save_logic(self) -> None: """Allow for some pre save operations by other classes.""" diff --git a/unittests/test_location_save_atomicity.py b/unittests/test_location_save_atomicity.py new file mode 100644 index 0000000000..71cfef9c7f --- /dev/null +++ b/unittests/test_location_save_atomicity.py @@ -0,0 +1,84 @@ +from django.urls import reverse + +from dojo.authorization.roles_permissions import Roles +from dojo.location.models import Location, LocationProductReference +from dojo.location.status import ProductLocationStatus +from dojo.models import ( + Dojo_User, + Product, + Product_Member, + Product_Type, + Role, + User, +) +from dojo.url.models import URL +from unittests.dojo_test_case import DojoTestCase, skip_unless_v3 + + +@skip_unless_v3 +class LocationSaveAtomicityTest(DojoTestCase): + + """ + A save that fails validation must leave no part of itself behind. + + AbstractLocation.pre_save_logic() writes the parent Location row, and the base model + runs it before full_clean(). A rejected endpoint rename told the user the edit failed + while the Location row kept the new value, so the row summarised a URL it did not store. + """ + + @classmethod + def setUpTestData(cls): + prod_type, _ = Product_Type.objects.get_or_create(name="LOC-Atomic PT") + + cls.product_a = Product.objects.create(name="LOC-Atomic Product A", description="A", prod_type=prod_type) + cls.product_b = Product.objects.create(name="LOC-Atomic Product B", description="B", prod_type=prod_type) + + # Legacy authorization is membership-based via authorized_users, so mirror the + # Product_Member row onto that M2M. + cls.alice = User.objects.create_user( + username="loc_atomic_alice", + password="not-a-real-secret", # noqa: S106 - test fixture user + ) + Product_Member.objects.create( + user=cls.alice, product=cls.product_a, role=Role.objects.get(id=Roles.Writer), + ) + cls.product_a.authorized_users.add(Dojo_User.objects.get(pk=cls.alice.pk)) + + cls.own_url = "https://own.example.test/mine" + cls.taken_url = "https://taken.example.test/admin" + cls.location_a = cls.associate(cls.own_url, cls.product_a) + cls.location_b = cls.associate(cls.taken_url, cls.product_b) + + @classmethod + def associate(cls, value, product): + location = URL.create_location_from_value(value).location + LocationProductReference.objects.create( + location=location, product=product, status=ProductLocationStatus.Active, + ) + return location + + def setUp(self): + super().setUp() + self.client.force_login(self.alice) + + def rename(self, location, host, path): + return self.client.post( + reverse("edit_endpoint", kwargs={"location_id": location.id}), + {"protocol": "https", "host": host, "path": path, + "port": "", "user_info": "", "query": "", "fragment": "", "tags": ""}, + ) + + def test_rejected_rename_leaves_both_rows_unchanged(self): + response = self.rename(self.location_a, "taken.example.test", "admin") + self.assertEqual(302, response.status_code) + + self.assertEqual(self.own_url, str(URL.objects.get(pk=self.location_a.url.pk))) + self.assertEqual(self.own_url, Location.objects.get(pk=self.location_a.pk).location_value) + self.assertEqual(self.taken_url, Location.objects.get(pk=self.location_b.pk).location_value) + + def test_accepted_rename_updates_both_rows(self): + self.rename(self.location_a, "renamed.example.test", "ok") + + renamed = "https://renamed.example.test/ok" + self.assertEqual(renamed, str(URL.objects.get(pk=self.location_a.url.pk))) + self.assertEqual(renamed, Location.objects.get(pk=self.location_a.pk).location_value)