Chore: Run read-only migration tests through one migration cycle per class

Each TestMigrations test migrates back, seeds, migrates forward and returns to
the latest migration, which is slow. The fulltext query
prefix, sha256 checksum and skip archive file classes only read the migrated
data, so they now opt in to migrate_once: the first test runs the migration,
the rest reuse its state, and the schema is restored and the tables flushed in
tearDownClass.
This commit is contained in:
stumpylog
2026-09-21 11:45:28 -07:00
parent f502cd5e34
commit 8849aff5d9
4 changed files with 83 additions and 20 deletions
@@ -8,6 +8,7 @@ pytestmark = pytest.mark.search
class TestMigrateFulltextQueryFieldPrefixes(TestMigrations):
migrate_from = "0016_sha256_checksums"
migrate_to = "0017_migrate_fulltext_query_field_prefixes"
migrate_once = True
def setUpBeforeMigration(self, apps) -> None:
User = apps.get_model("auth", "User")
@@ -1,5 +1,4 @@
import hashlib
import shutil
import tempfile
from pathlib import Path
@@ -19,19 +18,22 @@ class TestSha256ChecksumDataMigration(TestMigrations):
migrate_from = "0015_document_version_index_and_more"
migrate_to = "0016_sha256_checksums"
migrate_once = True
reset_sequences = True
ORIGINAL_CONTENT = b"original file content for sha256 migration test"
ARCHIVE_CONTENT = b"archive file content for sha256 migration test"
def setUpBeforeMigration(self, apps) -> None:
self._originals_dir = Path(tempfile.mkdtemp())
self._archive_dir = Path(tempfile.mkdtemp())
self._settings_override = override_settings(
ORIGINALS_DIR=self._originals_dir,
ARCHIVE_DIR=self._archive_dir,
@classmethod
def setUpClass(cls) -> None:
super().setUpClass()
originals_dir = Path(cls.enterClassContext(tempfile.TemporaryDirectory()))
archive_dir = Path(cls.enterClassContext(tempfile.TemporaryDirectory()))
cls.enterClassContext(
override_settings(ORIGINALS_DIR=originals_dir, ARCHIVE_DIR=archive_dir),
)
self._settings_override.enable()
def setUpBeforeMigration(self, apps) -> None:
Document = apps.get_model("documents", "Document")
# doc1: original file present, no archive
@@ -85,8 +87,9 @@ class TestSha256ChecksumDataMigration(TestMigrations):
archive_checksum=None,
).pk
def _fixture_teardown(self) -> None:
super()._fixture_teardown()
@classmethod
def tearDownClass(cls) -> None:
super().tearDownClass()
# Django's SQLite backend returns [] from sequence_reset_sql(), so
# reset_sequences=True flushes rows but never clears sqlite_sequence.
# Explicitly delete the entry so subsequent tests start from pk=1.
@@ -96,12 +99,6 @@ class TestSha256ChecksumDataMigration(TestMigrations):
"DELETE FROM sqlite_sequence WHERE name='documents_document'",
)
def tearDown(self) -> None:
super().tearDown()
self._settings_override.disable()
shutil.rmtree(self._originals_dir, ignore_errors=True)
shutil.rmtree(self._archive_dir, ignore_errors=True)
def test_original_checksum_updated_to_sha256_when_file_exists(self) -> None:
Document = self.apps.get_model("documents", "Document")
doc = Document.objects.get(pk=self.doc1_id)
+68 -4
View File
@@ -159,6 +159,16 @@ class ConsumeTaskMixin:
class TestMigrations(TransactionTestCase):
"""Run a migration on seeded data, then let the tests inspect the result.
By default every test migrates back, seeds, migrates forward and returns to
the latest migration, which costs several seconds. A class whose tests only
read the migrated data can set ``migrate_once`` to pay that once per class:
the migration runs for the first test, the database is left alone between
tests, and it is restored and flushed when the class finishes. Such tests
must not write to the database.
"""
@property
def app(self):
return apps.get_containing_app_config(type(self).__module__).name
@@ -166,10 +176,38 @@ class TestMigrations(TransactionTestCase):
migrate_from = None
dependencies = None
migrate_to = None
migrate_once = False
_once_owner: "TestMigrations | None" = None
_once_state: dict[str, Any] | None = None
_once_finishing = False
def setUp(self) -> None:
super().setUp()
cls = type(self)
if self.migrate_once:
if cls._once_state is not None:
vars(self).update(cls._once_state)
return
if cls._once_owner is not None:
raise RuntimeError(
f"The migration in '{cls.__name__}' failed for an earlier test",
)
# Recorded before migrating so a failed migration is still restored
cls._once_owner = self
before = dict(vars(self))
self._migrate()
if self.migrate_once:
cls._once_state = {
name: value
for name, value in vars(self).items()
if name not in before or before[name] is not value
}
def _migrate(self) -> None:
assert self.migrate_from and self.migrate_to, (
f"TestCase '{type(self).__name__}' must define migrate_from and migrate_to properties"
)
@@ -197,19 +235,45 @@ class TestMigrations(TransactionTestCase):
def setUpBeforeMigration(self, apps) -> None:
pass
def _migrate_to_latest(self) -> None:
executor = MigrationExecutor(connection)
executor.loader.build_graph()
targets = executor.loader.graph.leaf_nodes()
executor.migrate(targets)
def tearDown(self) -> None:
"""
Ensure the database schema is restored to the latest migration after
each migration test, so subsequent tests run against HEAD.
"""
if self.migrate_once and not self._once_finishing:
return
try:
executor = MigrationExecutor(connection)
executor.loader.build_graph()
targets = executor.loader.graph.leaf_nodes()
executor.migrate(targets)
self._migrate_to_latest()
finally:
super().tearDown()
def _fixture_teardown(self) -> None:
# Django flushes every table after each test, which would discard the
# data the remaining tests of a migrate_once class still need
if self.migrate_once and not self._once_finishing:
return
super()._fixture_teardown()
@classmethod
def tearDownClass(cls) -> None:
owner = cls._once_owner
try:
if owner is not None:
cls._once_finishing = True
owner.tearDown()
owner._fixture_teardown()
finally:
cls._once_owner = None
cls._once_state = None
cls._once_finishing = False
super().tearDownClass()
class SampleDirMixin:
SAMPLE_DIR = Path(__file__).parent / "samples"
@@ -4,6 +4,7 @@ from documents.tests.utils import TestMigrations
class TestMigrateSkipArchiveFile(TestMigrations):
migrate_from = "0007_optimize_integer_field_sizes"
migrate_to = "0008_replace_skip_archive_file"
migrate_once = True
def setUpBeforeMigration(self, apps):
ApplicationConfiguration = apps.get_model(