Get default user for bulk ops from middleware (#677)

* Get default user for bulk ops from middleware

* add user on init

* Added user to bulk ops

* Added changes

* lint

* skip tests

Author: Ross Mechanic

CHANGES.rst

@@ -1,8 +1,12 @@
 Changes
 =======
-2.10.1 (2020-06-16)
-- added ``user_db_constraint`` param to history to avoid circular reference on delete (gh-676)
-- added ``clean_old_history`` management command (gh-675)
+
+Unreleased
+----------
+- Added ``clean_old_history`` management command (gh-675)
+- Added ``user_db_constraint`` param to history to avoid circular reference on delete (gh-676)
+- Leverages ``get_user`` from ``HistoricalRecords`` in order to set a fallback user on
+  bulk update and bulk create (gh-677)
 
 2.10.0 (2020-04-27)
 -------------------

simple_history/manager.py

@@ -120,9 +120,14 @@ def bulk_history_create(
 
         historical_instances = []
         for instance in objs:
+            history_user = getattr(
+                instance,
+                "_history_user",
+                default_user or self.model.get_default_history_user(instance),
+            )
             row = self.model(
                 history_date=getattr(instance, "_history_date", timezone.now()),
-                history_user=getattr(instance, "_history_user", default_user),
+                history_user=history_user,
                 history_change_reason=get_change_reason_from_object(instance)
                 or default_change_reason,
                 history_type=history_type,

simple_history/models.py

@@ -422,6 +422,13 @@ def get_prev_record(self):
                 .last()
             )
 
+        def get_default_history_user(instance):
+            """
+            Returns the user specified by `get_user` method for manually creating
+            historical objects
+            """
+            return self.get_history_user(instance)
+
         extra_fields = {
             "history_id": self._get_history_id_field(),
             "history_date": models.DateTimeField(),
@@ -441,6 +448,7 @@ def get_prev_record(self):
             "__str__": lambda self: "{} as of {}".format(
                 self.history_object, self.history_date
             ),
+            "get_default_history_user": staticmethod(get_default_history_user),
         }
 
         extra_fields.update(self._get_history_related_field(model))

simple_history/tests/tests/test_middleware.py

@@ -1,5 +1,7 @@
 from datetime import date
+from unittest import skipIf
 
+import django
 from django.test import TestCase, override_settings
 from django.urls import reverse
 
@@ -145,3 +147,91 @@ def test_bucket_member_is_set_on_create_view_when_logged_in(self):
         history = bucket_datas.first().history.all()
 
         self.assertListEqual([h.history_user_id for h in history], [member1.id])
+
+
+@override_settings(**middleware_override_settings)
+class MiddlewareBulkOpsTest(TestCase):
+    def setUp(self):
+        self.user = CustomUser.objects.create_superuser(
+            "user_login", "[email protected]", "pass"
+        )
+
+    def test_user_is_set_on_bulk_create_view_when_logged_in(self):
+        self.client.force_login(self.user)
+        self.client.post(reverse("poll-bulk-create"), data={})
+        polls = Poll.objects.all()
+        self.assertEqual(len(polls), 2)
+
+        poll_history = Poll.history.all()
+
+        self.assertCountEqual(
+            [ph.history_user_id for ph in poll_history], [self.user.id, self.user.id]
+        )
+
+    def test_user_is_not_set_on_bulk_create_view_not_logged_in(self):
+        self.client.post(reverse("poll-bulk-create"), data={})
+
+        polls = Poll.objects.all()
+        self.assertEqual(polls.count(), 2)
+
+        poll_history = Poll.history.all()
+
+        self.assertListEqual([ph.history_user_id for ph in poll_history], [None, None])
+
+    def test_request_user_is_overwritten_by_default_user_on_bulk_create_view(self,):
+        self.client.force_login(self.user)
+        self.client.post(reverse("poll-bulk-create-with-default-user"), data={})
+
+        polls = Poll.objects.all()
+        self.assertEqual(len(polls), 2)
+
+        poll_history = Poll.history.all()
+
+        self.assertFalse(any(ph.history_user_id == self.user.id for ph in poll_history))
+        self.assertFalse(any(ph.history_user_id is None for ph in poll_history))
+
+    @skipIf(django.VERSION < (2, 2,), reason="bulk_update does not exist before 2.2")
+    def test_user_is_set_on_bulk_update_view_when_logged_in(self):
+        self.client.force_login(self.user)
+        poll_1 = Poll.objects.create(question="Test question 1", pub_date=date.today())
+        poll_2 = Poll.objects.create(
+            question="Test question 2", pub_date=date(2020, 1, 1)
+        )
+
+        self.client.post(reverse("poll-bulk-update"), data={})
+
+        polls = Poll.objects.all()
+        self.assertEqual(2, len(polls))
+
+        self.assertEqual("1", poll_1.history.latest("history_date").question)
+        self.assertEqual("0", poll_2.history.latest("history_date").question)
+        self.assertEqual(
+            self.user.id, poll_1.history.latest("history_date").history_user_id
+        )
+        self.assertEqual(
+            self.user.id, poll_2.history.latest("history_date").history_user_id
+        )
+
+    @skipIf(django.VERSION < (2, 2,), reason="bulk_update does not exist before 2.2")
+    def test_user_is_not_set_on_bulk_update_view_when_not_logged_in(self):
+        poll_1 = Poll.objects.create(question="Test question 1", pub_date=date.today())
+        poll_2 = Poll.objects.create(
+            question="Test question 2", pub_date=date(2020, 1, 1)
+        )
+
+        self.client.post(reverse("poll-bulk-update"), data={})
+
+        self.assertIsNone(poll_1.history.latest("history_date").history_user_id)
+        self.assertIsNone(poll_2.history.latest("history_date").history_user_id)
+
+    @skipIf(django.VERSION < (2, 2,), reason="bulk_update does not exist before 2.2")
+    def test_request_user_is_overwritten_by_default_user_on_bulk_update(self):
+        self.client.force_login(self.user)
+        poll = Poll.objects.create(pub_date=date(2020, 1, 1), question="123")
+
+        self.client.post(reverse("poll-bulk-update-with-default-user"), data={})
+
+        self.assertIsNotNone(poll.history.latest("history_date").history_user_id)
+        self.assertNotEqual(
+            self.user.id, poll.history.latest("history_date").history_user_id
+        )

simple_history/tests/urls.py

@@ -12,6 +12,10 @@
     PollList,
     PollUpdate,
     PollWithHistoricalIPAddressCreate,
+    PollBulkCreateView,
+    PollBulkCreateWithDefaultUserView,
+    PollBulkUpdateView,
+    PollBulkUpdateWithDefaultUserView,
 )
 from . import other_admin
 
@@ -40,4 +44,16 @@
     url(r"^poll/(?P<pk>[0-9]+)/delete/$", PollDelete.as_view(), name="poll-delete"),
     url(r"^polls/(?P<pk>[0-9]+)/$", PollDetail.as_view(), name="poll-detail"),
     url(r"^polls/$", PollList.as_view(), name="poll-list"),
+    url(r"^polls-bulk-create/$", PollBulkCreateView.as_view(), name="poll-bulk-create"),
+    url(
+        r"^polls-bulk-create-default-user/$",
+        PollBulkCreateWithDefaultUserView.as_view(),
+        name="poll-bulk-create-with-default-user",
+    ),
+    url(r"^polls-bulk-update/$", PollBulkUpdateView.as_view(), name="poll-bulk-update"),
+    url(
+        r"^polls-bulk-update-default-user/$",
+        PollBulkUpdateWithDefaultUserView.as_view(),
+        name="poll-bulk-update-with-default-user",
+    ),
 ]

simple_history/tests/view.py

@@ -1,4 +1,8 @@
+from datetime import date
+
+from django.http import HttpResponse
 from django.urls import reverse_lazy
+from django.views import View
 from django.views.generic import (
     CreateView,
     DeleteView,
@@ -7,18 +11,71 @@
     UpdateView,
 )
 
+from simple_history.tests.custom_user.models import CustomUser
 from simple_history.tests.models import (
     BucketDataRegisterRequestUser,
     Poll,
     PollWithHistoricalIPAddress,
 )
+from simple_history.utils import bulk_create_with_history, bulk_update_with_history
 
 
 class PollCreate(CreateView):
     model = Poll
     fields = ["question", "pub_date"]
 
 
+class PollBulkCreateView(View):
+    def post(self, request, *args, **kwargs):
+        poll_info_list = [
+            {"question": "1", "pub_date": date(2020, 1, 1)},
+            {"question": "2", "pub_date": date(2020, 1, 2)},
+        ]
+        polls_to_create = [Poll(**poll_info) for poll_info in poll_info_list]
+        bulk_create_with_history(polls_to_create, Poll)
+        return HttpResponse(status=200)
+
+
+class PollBulkCreateWithDefaultUserView(View):
+    def post(self, request, *args, **kwargs):
+        default_user = CustomUser.objects.create_superuser(
+            "test_user", "[email protected]", "pass"
+        )
+        # Bulk create objects with history
+        poll_info_list = [
+            {"question": "1", "pub_date": date(2020, 1, 1)},
+            {"question": "2", "pub_date": date(2020, 1, 2)},
+        ]
+        polls_to_create = [Poll(**poll_info) for poll_info in poll_info_list]
+        bulk_create_with_history(polls_to_create, Poll, default_user=default_user)
+        return HttpResponse(status=200)
+
+
+class PollBulkUpdateView(View):
+    def post(self, request, *args, **kwargs):
+        polls = Poll.objects.order_by("pub_date")
+        for i, poll in enumerate(polls):
+            poll.question = str(i)
+
+        bulk_update_with_history(polls, fields=["question"], model=Poll)
+        return HttpResponse(status=201)
+
+
+class PollBulkUpdateWithDefaultUserView(View):
+    def post(self, request, *args, **kwargs):
+        default_user = CustomUser.objects.create_superuser(
+            "test_user", "[email protected]", "pass"
+        )
+        polls = Poll.objects.all()
+        for i, poll in enumerate(polls):
+            poll.question = str(i)
+
+        bulk_update_with_history(
+            polls, fields=["question"], model=Poll, default_user=default_user
+        )
+        return HttpResponse(status=201)
+
+
 class PollWithHistoricalIPAddressCreate(CreateView):
     model = PollWithHistoricalIPAddress
     fields = ["question", "pub_date"]