Resolves test issues with Python 3.12 (#6902)

This commit is contained in:
Trenton H 2024-06-03 12:33:46 -07:00 committed by GitHub
parent de7c22e8d6
commit 6d2ae3df1f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 67 additions and 52 deletions

View File

@ -485,56 +485,65 @@ class ConsumerPlugin(
Return the document object if it was successfully created.
"""
self._send_progress(
0,
100,
ProgressStatusOptions.STARTED,
ConsumerStatusShortMessage.NEW_FILE,
)
tempdir = None
# Make sure that preconditions for consuming the file are met.
self.pre_check_file_exists()
self.pre_check_directories()
self.pre_check_duplicate()
self.pre_check_asn_value()
self.log.info(f"Consuming {self.filename}")
# For the actual work, copy the file into a tempdir
tempdir = tempfile.TemporaryDirectory(
prefix="paperless-ngx",
dir=settings.SCRATCH_DIR,
)
self.working_copy = Path(tempdir.name) / Path(self.filename)
copy_file_with_basic_stats(self.input_doc.original_file, self.working_copy)
# Determine the parser class.
mime_type = magic.from_file(self.working_copy, mime=True)
self.log.debug(f"Detected mime type: {mime_type}")
# Based on the mime type, get the parser for that type
parser_class: Optional[type[DocumentParser]] = get_parser_class_for_mime_type(
mime_type,
)
if not parser_class:
tempdir.cleanup()
self._fail(
ConsumerStatusShortMessage.UNSUPPORTED_TYPE,
f"Unsupported mime type {mime_type}",
try:
self._send_progress(
0,
100,
ProgressStatusOptions.STARTED,
ConsumerStatusShortMessage.NEW_FILE,
)
# Notify all listeners that we're going to do some work.
# Make sure that preconditions for consuming the file are met.
document_consumption_started.send(
sender=self.__class__,
filename=self.working_copy,
logging_group=self.logging_group,
)
self.pre_check_file_exists()
self.pre_check_directories()
self.pre_check_duplicate()
self.pre_check_asn_value()
self.run_pre_consume_script()
self.log.info(f"Consuming {self.filename}")
# For the actual work, copy the file into a tempdir
tempdir = tempfile.TemporaryDirectory(
prefix="paperless-ngx",
dir=settings.SCRATCH_DIR,
)
self.working_copy = Path(tempdir.name) / Path(self.filename)
copy_file_with_basic_stats(self.input_doc.original_file, self.working_copy)
# Determine the parser class.
mime_type = magic.from_file(self.working_copy, mime=True)
self.log.debug(f"Detected mime type: {mime_type}")
# Based on the mime type, get the parser for that type
parser_class: Optional[type[DocumentParser]] = (
get_parser_class_for_mime_type(
mime_type,
)
)
if not parser_class:
tempdir.cleanup()
self._fail(
ConsumerStatusShortMessage.UNSUPPORTED_TYPE,
f"Unsupported mime type {mime_type}",
)
# Notify all listeners that we're going to do some work.
document_consumption_started.send(
sender=self.__class__,
filename=self.working_copy,
logging_group=self.logging_group,
)
self.run_pre_consume_script()
except:
if tempdir:
tempdir.cleanup()
raise
def progress_callback(current_progress, max_progress): # pragma: no cover
# recalculate progress to be within 20 and 80
@ -593,6 +602,9 @@ class ConsumerPlugin(
archive_path = document_parser.get_archive_path()
except ParseError as e:
document_parser.cleanup()
if tempdir:
tempdir.cleanup()
self._fail(
str(e),
f"Error occurred while consuming document {self.filename}: {e}",
@ -601,7 +613,8 @@ class ConsumerPlugin(
)
except Exception as e:
document_parser.cleanup()
tempdir.cleanup()
if tempdir:
tempdir.cleanup()
self._fail(
str(e),
f"Unexpected error while consuming document {self.filename}: {e}",

View File

@ -70,12 +70,13 @@ class TestApiAppConfig(DirectoriesMixin, APITestCase):
config.app_logo = "/logo/example.jpg"
config.save()
response = self.client.get("/api/ui_settings/", format="json")
self.assertDictContainsSubset(
self.assertDictEqual(
response.data["settings"],
{
"app_title": config.app_title,
"app_logo": config.app_logo,
},
response.data["settings"],
}
| response.data["settings"],
)
def test_api_update_config(self):

View File

@ -339,11 +339,12 @@ class TestDBSettings(TestCase):
):
databases = _parse_db_settings()
self.assertDictContainsSubset(
{
self.assertDictEqual(
databases["default"]["OPTIONS"],
databases["default"]["OPTIONS"]
| {
"connect_timeout": 10.0,
},
databases["default"]["OPTIONS"],
)
self.assertDictEqual(
{