mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-09-27 11:50:31 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8849aff5d9 |
@@ -8,6 +8,7 @@ pytestmark = pytest.mark.search
|
|||||||
class TestMigrateFulltextQueryFieldPrefixes(TestMigrations):
|
class TestMigrateFulltextQueryFieldPrefixes(TestMigrations):
|
||||||
migrate_from = "0016_sha256_checksums"
|
migrate_from = "0016_sha256_checksums"
|
||||||
migrate_to = "0017_migrate_fulltext_query_field_prefixes"
|
migrate_to = "0017_migrate_fulltext_query_field_prefixes"
|
||||||
|
migrate_once = True
|
||||||
|
|
||||||
def setUpBeforeMigration(self, apps) -> None:
|
def setUpBeforeMigration(self, apps) -> None:
|
||||||
User = apps.get_model("auth", "User")
|
User = apps.get_model("auth", "User")
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
import hashlib
|
import hashlib
|
||||||
import shutil
|
|
||||||
import tempfile
|
import tempfile
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
@@ -19,19 +18,22 @@ class TestSha256ChecksumDataMigration(TestMigrations):
|
|||||||
|
|
||||||
migrate_from = "0015_document_version_index_and_more"
|
migrate_from = "0015_document_version_index_and_more"
|
||||||
migrate_to = "0016_sha256_checksums"
|
migrate_to = "0016_sha256_checksums"
|
||||||
|
migrate_once = True
|
||||||
reset_sequences = True
|
reset_sequences = True
|
||||||
|
|
||||||
ORIGINAL_CONTENT = b"original file content for sha256 migration test"
|
ORIGINAL_CONTENT = b"original file content for sha256 migration test"
|
||||||
ARCHIVE_CONTENT = b"archive file content for sha256 migration test"
|
ARCHIVE_CONTENT = b"archive file content for sha256 migration test"
|
||||||
|
|
||||||
def setUpBeforeMigration(self, apps) -> None:
|
@classmethod
|
||||||
self._originals_dir = Path(tempfile.mkdtemp())
|
def setUpClass(cls) -> None:
|
||||||
self._archive_dir = Path(tempfile.mkdtemp())
|
super().setUpClass()
|
||||||
self._settings_override = override_settings(
|
originals_dir = Path(cls.enterClassContext(tempfile.TemporaryDirectory()))
|
||||||
ORIGINALS_DIR=self._originals_dir,
|
archive_dir = Path(cls.enterClassContext(tempfile.TemporaryDirectory()))
|
||||||
ARCHIVE_DIR=self._archive_dir,
|
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")
|
Document = apps.get_model("documents", "Document")
|
||||||
|
|
||||||
# doc1: original file present, no archive
|
# doc1: original file present, no archive
|
||||||
@@ -85,8 +87,9 @@ class TestSha256ChecksumDataMigration(TestMigrations):
|
|||||||
archive_checksum=None,
|
archive_checksum=None,
|
||||||
).pk
|
).pk
|
||||||
|
|
||||||
def _fixture_teardown(self) -> None:
|
@classmethod
|
||||||
super()._fixture_teardown()
|
def tearDownClass(cls) -> None:
|
||||||
|
super().tearDownClass()
|
||||||
# Django's SQLite backend returns [] from sequence_reset_sql(), so
|
# Django's SQLite backend returns [] from sequence_reset_sql(), so
|
||||||
# reset_sequences=True flushes rows but never clears sqlite_sequence.
|
# reset_sequences=True flushes rows but never clears sqlite_sequence.
|
||||||
# Explicitly delete the entry so subsequent tests start from pk=1.
|
# 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'",
|
"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:
|
def test_original_checksum_updated_to_sha256_when_file_exists(self) -> None:
|
||||||
Document = self.apps.get_model("documents", "Document")
|
Document = self.apps.get_model("documents", "Document")
|
||||||
doc = Document.objects.get(pk=self.doc1_id)
|
doc = Document.objects.get(pk=self.doc1_id)
|
||||||
|
|||||||
@@ -159,6 +159,16 @@ class ConsumeTaskMixin:
|
|||||||
|
|
||||||
|
|
||||||
class TestMigrations(TransactionTestCase):
|
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
|
@property
|
||||||
def app(self):
|
def app(self):
|
||||||
return apps.get_containing_app_config(type(self).__module__).name
|
return apps.get_containing_app_config(type(self).__module__).name
|
||||||
@@ -166,10 +176,38 @@ class TestMigrations(TransactionTestCase):
|
|||||||
migrate_from = None
|
migrate_from = None
|
||||||
dependencies = None
|
dependencies = None
|
||||||
migrate_to = 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:
|
def setUp(self) -> None:
|
||||||
super().setUp()
|
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, (
|
assert self.migrate_from and self.migrate_to, (
|
||||||
f"TestCase '{type(self).__name__}' must define migrate_from and migrate_to properties"
|
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:
|
def setUpBeforeMigration(self, apps) -> None:
|
||||||
pass
|
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:
|
def tearDown(self) -> None:
|
||||||
"""
|
"""
|
||||||
Ensure the database schema is restored to the latest migration after
|
Ensure the database schema is restored to the latest migration after
|
||||||
each migration test, so subsequent tests run against HEAD.
|
each migration test, so subsequent tests run against HEAD.
|
||||||
"""
|
"""
|
||||||
|
if self.migrate_once and not self._once_finishing:
|
||||||
|
return
|
||||||
try:
|
try:
|
||||||
executor = MigrationExecutor(connection)
|
self._migrate_to_latest()
|
||||||
executor.loader.build_graph()
|
|
||||||
targets = executor.loader.graph.leaf_nodes()
|
|
||||||
executor.migrate(targets)
|
|
||||||
finally:
|
finally:
|
||||||
super().tearDown()
|
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:
|
class SampleDirMixin:
|
||||||
SAMPLE_DIR = Path(__file__).parent / "samples"
|
SAMPLE_DIR = Path(__file__).parent / "samples"
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ from documents.tests.utils import TestMigrations
|
|||||||
class TestMigrateSkipArchiveFile(TestMigrations):
|
class TestMigrateSkipArchiveFile(TestMigrations):
|
||||||
migrate_from = "0007_optimize_integer_field_sizes"
|
migrate_from = "0007_optimize_integer_field_sizes"
|
||||||
migrate_to = "0008_replace_skip_archive_file"
|
migrate_to = "0008_replace_skip_archive_file"
|
||||||
|
migrate_once = True
|
||||||
|
|
||||||
def setUpBeforeMigration(self, apps):
|
def setUpBeforeMigration(self, apps):
|
||||||
ApplicationConfiguration = apps.get_model(
|
ApplicationConfiguration = apps.get_model(
|
||||||
|
|||||||
Reference in New Issue
Block a user