Add populate_tenant_history management command

tcms_tenants/management/commands/populate_tenant_history.py

@@ -0,0 +1,58 @@
+# Copyright (c) 2026 Alexander Todorov <atodorov@otb.bg>
+#
+# Licensed under GNU Affero General Public License v3 or later (AGPLv3+)
+# https://www.gnu.org/licenses/agpl-3.0.html
+
+from django.utils import timezone
+from django_tenants.utils import get_tenant_model, tenant_context
+from simple_history.management.commands.populate_history import (
+    Command as PopulateHistoryCommand,
+)
+
+
+class Command(PopulateHistoryCommand):
+    help = (
+        "Populates the corresponding HistoricalRecords field with "
+        "the current state of all instances in a model for all tenants"
+    )
+
+    def add_arguments(self, parser):
+        super().add_arguments(parser)
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            default=False,
+            help="Don't create anything, just report what would be populated",
+        )
+
+    def handle(  # pylint: disable=attribute-defined-outside-init
+        self, *args, **options
+    ):
+        self.verbosity = options["verbosity"]
+        self.dry_run = options["dry_run"]
+
+        output = None
+        if self.verbosity:
+            output = self.stdout
+
+        for tenant in get_tenant_model().objects.all():
+            if output:
+                output.write(
+                    f"\n\n {timezone.now()} === Populating history"
+                    f" for tenant '{tenant.schema_name}' === dry run: {self.dry_run} ==="
+                )
+
+            with tenant_context(tenant):
+                super().handle(*args, **options)
+
+            if output:
+                output.write(
+                    f"\n\n {timezone.now()} === End populating history"
+                    f" for tenant '{tenant.schema_name}' ==="
+                )
+
+    def _bulk_history_create(self, model, batch_size):
+        if self.dry_run:
+            return
+
+        super()._bulk_history_create(model, batch_size)

tcms_tenants/management/commands/upgrade_tenants.py

@@ -27,26 +27,34 @@ def handle(self, *args, **kwargs):
         self.stdout.write("""To finish the upgrade process, the following
 management commands will be executed:
 
+populate_tenant_history
 refresh_tenant_permissions
 remove_stale_tenant_attachments
 remove_stale_tenant_comments
             """)
 
-        self.stdout.write("\n1. Refreshing tenant permissions:")
+        self.stdout.write("\n1. Populating history:")
+        call_command(
+            "populate_tenant_history",
+            "--auto",
+            verbosity=kwargs["verbosity"],
+        )
+
+        self.stdout.write("\n2. Refreshing tenant permissions:")
         call_command(
             "refresh_tenant_permissions",
             verbosity=kwargs["verbosity"],
             interactive=kwargs["interactive"],
         )
 
-        self.stdout.write("\n2. Removing stale attachments:")
+        self.stdout.write("\n3. Removing stale attachments:")
         call_command(
             "remove_stale_tenant_attachments",
             verbosity=kwargs["verbosity"],
             answer=answer,
         )
 
-        self.stdout.write("\n3. Removing stale comments:")
+        self.stdout.write("\n4. Removing stale comments:")
         call_command(
             "remove_stale_tenant_comments",
             verbosity=kwargs["verbosity"],

tcms_tenants/tests/test_populate_tenant_history.py

@@ -0,0 +1,96 @@
+# Copyright (c) 2026 Alexander Todorov <atodorov@otb.bg>
+#
+# Licensed under GNU Affero General Public License v3 or later (AGPLv3+)
+# https://www.gnu.org/licenses/agpl-3.0.html
+
+# pylint: disable=too-many-ancestors
+
+from io import StringIO
+
+from django.core.management import call_command
+
+from django_tenants.utils import tenant_context
+from simple_history.utils import get_history_model_for_model
+
+from tcms.testcases.models import TestCase
+from tcms.tests.factories import TestCaseFactory
+
+from tcms_tenants.tests import TenantGroupsTestCase, tenants_with_schema
+
+
+class PopulateTenantHistoryTestCase(TenantGroupsTestCase):
+    model = "testcases.TestCase"
+    instances_per_tenant = 3
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+        for tenant in tenants_with_schema():
+            with tenant_context(tenant):
+                for _ in range(cls.instances_per_tenant):
+                    test_case = TestCaseFactory()
+                    test_case.save()  # creates a history record
+                    test_case.emailing.notify_on_case_update = False
+                    test_case.emailing.save()
+
+                # emulate objects which existed before history was enabled
+                get_history_model_for_model(TestCase).objects.all().delete()
+
+    def test_populates_history_for_all_tenants(self):
+        tenants = tenants_with_schema()
+        self.assertGreaterEqual(len(tenants), 1)
+
+        history_model = get_history_model_for_model(TestCase)
+        for tenant in tenants:
+            with tenant_context(tenant):
+                self.assertTrue(TestCase.objects.exists())
+                self.assertFalse(history_model.objects.exists())
+
+        out = StringIO()
+        call_command(
+            "populate_tenant_history",
+            self.model,
+            verbosity=1,
+            stdout=out,
+        )
+
+        output = out.getvalue()
+
+        for tenant in tenants:
+            with tenant_context(tenant):
+                self.assertEqual(
+                    history_model.objects.count(), TestCase.objects.count()
+                )
+
+            for banner in ("Populating history", "End populating history"):
+                self.assertIn(
+                    f"=== {banner} for tenant '{tenant.schema_name}' ===",
+                    output,
+                )
+
+    def test_dry_run_does_not_populate_history(self):
+        tenants = tenants_with_schema()
+        self.assertGreaterEqual(len(tenants), 1)
+
+        history_model = get_history_model_for_model(TestCase)
+
+        out = StringIO()
+        call_command(
+            "populate_tenant_history",
+            self.model,
+            dry_run=True,
+            verbosity=1,
+            stdout=out,
+        )
+
+        output = out.getvalue()
+
+        for tenant in tenants:
+            with tenant_context(tenant):
+                self.assertFalse(history_model.objects.exists())
+
+            self.assertIn(
+                f"for tenant '{tenant.schema_name}' === dry run: True ===",
+                output,
+            )