mirror of
				https://github.com/paperless-ngx/paperless-ngx.git
				synced 2025-11-04 03:27:12 -05:00 
			
		
		
		
	fix & test a migration
This commit is contained in:
		
							parent
							
								
									e79c45c98d
								
							
						
					
					
						commit
						3f94fc2618
					
				@ -11,6 +11,7 @@ from paperless.db import GnuPG
 | 
			
		||||
STORAGE_TYPE_UNENCRYPTED = "unencrypted"
 | 
			
		||||
STORAGE_TYPE_GPG = "gpg"
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def source_path(self):
 | 
			
		||||
    if self.filename:
 | 
			
		||||
        fname = str(self.filename)
 | 
			
		||||
 | 
			
		||||
							
								
								
									
										129
									
								
								src/documents/tests/test_migrations.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										129
									
								
								src/documents/tests/test_migrations.py
									
									
									
									
									
										Normal file
									
								
							@ -0,0 +1,129 @@
 | 
			
		||||
import os
 | 
			
		||||
import shutil
 | 
			
		||||
from pathlib import Path
 | 
			
		||||
 | 
			
		||||
from django.apps import apps
 | 
			
		||||
from django.conf import settings
 | 
			
		||||
from django.db import connection
 | 
			
		||||
from django.db.migrations.executor import MigrationExecutor
 | 
			
		||||
from django.test import TestCase, TransactionTestCase, override_settings
 | 
			
		||||
 | 
			
		||||
from documents.models import Document
 | 
			
		||||
from documents.parsers import get_default_file_extension
 | 
			
		||||
from documents.tests.utils import DirectoriesMixin
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
class TestMigrations(TransactionTestCase):
 | 
			
		||||
 | 
			
		||||
    @property
 | 
			
		||||
    def app(self):
 | 
			
		||||
        return apps.get_containing_app_config(type(self).__module__).name
 | 
			
		||||
 | 
			
		||||
    migrate_from = None
 | 
			
		||||
    migrate_to = None
 | 
			
		||||
 | 
			
		||||
    def setUp(self):
 | 
			
		||||
        super(TestMigrations, self).setUp()
 | 
			
		||||
 | 
			
		||||
        assert self.migrate_from and self.migrate_to, \
 | 
			
		||||
            "TestCase '{}' must define migrate_from and migrate_to     properties".format(type(self).__name__)
 | 
			
		||||
        self.migrate_from = [(self.app, self.migrate_from)]
 | 
			
		||||
        self.migrate_to = [(self.app, self.migrate_to)]
 | 
			
		||||
        executor = MigrationExecutor(connection)
 | 
			
		||||
        old_apps = executor.loader.project_state(self.migrate_from).apps
 | 
			
		||||
 | 
			
		||||
        # Reverse to the original migration
 | 
			
		||||
        executor.migrate(self.migrate_from)
 | 
			
		||||
 | 
			
		||||
        self.setUpBeforeMigration(old_apps)
 | 
			
		||||
 | 
			
		||||
        # Run the migration to test
 | 
			
		||||
        executor = MigrationExecutor(connection)
 | 
			
		||||
        executor.loader.build_graph()  # reload.
 | 
			
		||||
        executor.migrate(self.migrate_to)
 | 
			
		||||
 | 
			
		||||
        self.apps = executor.loader.project_state(self.migrate_to).apps
 | 
			
		||||
 | 
			
		||||
    def setUpBeforeMigration(self, apps):
 | 
			
		||||
        pass
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
STORAGE_TYPE_UNENCRYPTED = "unencrypted"
 | 
			
		||||
STORAGE_TYPE_GPG = "gpg"
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def source_path_before(self):
 | 
			
		||||
    if self.filename:
 | 
			
		||||
        fname = str(self.filename)
 | 
			
		||||
    else:
 | 
			
		||||
        fname = "{:07}.{}".format(self.pk, self.file_type)
 | 
			
		||||
        if self.storage_type == STORAGE_TYPE_GPG:
 | 
			
		||||
            fname += ".gpg"
 | 
			
		||||
 | 
			
		||||
    return os.path.join(
 | 
			
		||||
        settings.ORIGINALS_DIR,
 | 
			
		||||
        fname
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def file_type_after(self):
 | 
			
		||||
    return get_default_file_extension(self.mime_type)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def source_path_after(doc):
 | 
			
		||||
    if doc.filename:
 | 
			
		||||
        fname = str(doc.filename)
 | 
			
		||||
    else:
 | 
			
		||||
        fname = "{:07}{}".format(doc.pk, file_type_after(doc))
 | 
			
		||||
        if doc.storage_type == STORAGE_TYPE_GPG:
 | 
			
		||||
            fname += ".gpg"  # pragma: no cover
 | 
			
		||||
 | 
			
		||||
    return os.path.join(
 | 
			
		||||
        settings.ORIGINALS_DIR,
 | 
			
		||||
        fname
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@override_settings(PASSPHRASE="test")
 | 
			
		||||
class TestMigrateMimeType(DirectoriesMixin, TestMigrations):
 | 
			
		||||
 | 
			
		||||
    migrate_from = '1002_auto_20201111_1105'
 | 
			
		||||
    migrate_to = '1003_mime_types'
 | 
			
		||||
 | 
			
		||||
    def setUpBeforeMigration(self, apps):
 | 
			
		||||
        Document = apps.get_model("documents", "Document")
 | 
			
		||||
        doc = Document.objects.create(title="test", file_type="pdf", filename="file1.pdf")
 | 
			
		||||
        self.doc_id = doc.id
 | 
			
		||||
        shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), source_path_before(doc))
 | 
			
		||||
 | 
			
		||||
        doc2 = Document.objects.create(checksum="B", file_type="pdf", storage_type=STORAGE_TYPE_GPG)
 | 
			
		||||
        self.doc2_id = doc2.id
 | 
			
		||||
        shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "documents", "originals", "0000002.pdf.gpg"), source_path_before(doc2))
 | 
			
		||||
 | 
			
		||||
    def testMimeTypesMigrated(self):
 | 
			
		||||
        Document = self.apps.get_model('documents', 'Document')
 | 
			
		||||
 | 
			
		||||
        doc = Document.objects.get(id=self.doc_id)
 | 
			
		||||
        self.assertEqual(doc.mime_type, "application/pdf")
 | 
			
		||||
 | 
			
		||||
        doc2 = Document.objects.get(id=self.doc2_id)
 | 
			
		||||
        self.assertEqual(doc2.mime_type, "application/pdf")
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@override_settings(PASSPHRASE="test")
 | 
			
		||||
class TestMigrateMimeTypeBackwards(DirectoriesMixin, TestMigrations):
 | 
			
		||||
 | 
			
		||||
    migrate_from = '1003_mime_types'
 | 
			
		||||
    migrate_to = '1002_auto_20201111_1105'
 | 
			
		||||
 | 
			
		||||
    def setUpBeforeMigration(self, apps):
 | 
			
		||||
        Document = apps.get_model("documents", "Document")
 | 
			
		||||
        doc = Document.objects.create(title="test", mime_type="application/pdf", filename="file1.pdf")
 | 
			
		||||
        self.doc_id = doc.id
 | 
			
		||||
        shutil.copy(os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"), source_path_after(doc))
 | 
			
		||||
 | 
			
		||||
    def testMimeTypesReverted(self):
 | 
			
		||||
        Document = self.apps.get_model('documents', 'Document')
 | 
			
		||||
 | 
			
		||||
        doc = Document.objects.get(id=self.doc_id)
 | 
			
		||||
        self.assertEqual(doc.file_type, "pdf")
 | 
			
		||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user