import tempfile
from enum import Enum
from pathlib import Path
+from typing import TYPE_CHECKING
import magic
from django.conf import settings
"""
Confirm the input file still exists where it should
"""
- if not os.path.isfile(self.input_doc.original_file):
+ if TYPE_CHECKING:
+ assert isinstance(self.input_doc.original_file, Path), (
+ self.input_doc.original_file
+ )
+ if not self.input_doc.original_file.is_file():
self._fail(
ConsumerStatusShortMessage.FILE_NOT_FOUND,
f"Cannot consume {self.input_doc.original_file}: File not found.",
"""
Using the MD5 of the file, check this exact file doesn't already exist
"""
- with open(self.input_doc.original_file, "rb") as f:
+ with Path(self.input_doc.original_file).open("rb") as f:
checksum = hashlib.md5(f.read()).hexdigest()
existing_doc = Document.global_objects.filter(
Q(checksum=checksum) | Q(archive_checksum=checksum),
log_msg += " Note: existing document is in the trash."
if settings.CONSUMER_DELETE_DUPLICATES:
- os.unlink(self.input_doc.original_file)
+ Path(self.input_doc.original_file).unlink()
self._fail(
msg,
log_msg,
if not settings.PRE_CONSUME_SCRIPT:
return
- if not os.path.isfile(settings.PRE_CONSUME_SCRIPT):
+ if not Path(settings.PRE_CONSUME_SCRIPT).is_file():
self._fail(
ConsumerStatusShortMessage.PRE_CONSUME_SCRIPT_NOT_FOUND,
f"Configured pre-consume script "
if not settings.POST_CONSUME_SCRIPT:
return
- if not os.path.isfile(settings.POST_CONSUME_SCRIPT):
+ if not Path(settings.POST_CONSUME_SCRIPT).is_file():
self._fail(
ConsumerStatusShortMessage.POST_CONSUME_SCRIPT_NOT_FOUND,
f"Configured post-consume script "
document.thumbnail_path,
)
- if archive_path and os.path.isfile(archive_path):
+ if archive_path and Path(archive_path).is_file():
document.archive_filename = generate_unique_filename(
document,
archive_filename=True,
document.archive_path,
)
- with open(archive_path, "rb") as f:
+ with Path(archive_path).open("rb") as f:
document.archive_checksum = hashlib.md5(
f.read(),
).hexdigest()
self.unmodified_original.unlink()
# https://github.com/jonaswinkler/paperless-ng/discussions/1037
- shadow_file = os.path.join(
- os.path.dirname(self.input_doc.original_file),
- "._" + os.path.basename(self.input_doc.original_file),
+ shadow_file = (
+ Path(self.input_doc.original_file).parent
+ / f"._{Path(self.input_doc.original_file).name}"
)
- if os.path.isfile(shadow_file):
+ if Path(shadow_file).is_file():
self.log.debug(f"Deleting file {shadow_file}")
- os.unlink(shadow_file)
+ Path(shadow_file).unlink()
except Exception as e:
self._fail(
create_date = date
self.log.debug(f"Creation date from parse_date: {create_date}")
else:
- stats = os.stat(self.input_doc.original_file)
+ stats = Path(self.input_doc.original_file).stat()
create_date = timezone.make_aware(
datetime.datetime.fromtimestamp(stats.st_mtime),
)
) # adds to document
def _write(self, storage_type, source, target):
- with open(source, "rb") as read_file, open(target, "wb") as write_file:
+ with (
+ Path(source).open("rb") as read_file,
+ Path(target).open("wb") as write_file,
+ ):
write_file.write(read_file.read())
# Attempt to copy file's original stats, but it's ok if we can't
import datetime
-import os
import shutil
import tempfile
import uuid
from datetime import date
from datetime import timedelta
from pathlib import Path
+from typing import TYPE_CHECKING
from unittest import mock
import celery
content = b"This is a test"
content_thumbnail = b"thumbnail content"
- with open(filename, "wb") as f:
+ with Path(filename).open("wb") as f:
f.write(content)
doc = Document.objects.create(
title="none",
- filename=os.path.basename(filename),
+ filename=Path(filename).name,
mime_type="application/pdf",
)
- with open(
- os.path.join(self.dirs.thumbnail_dir, f"{doc.pk:07d}.webp"),
- "wb",
- ) as f:
+ if TYPE_CHECKING:
+ assert isinstance(self.dirs.thumbnail_dir, Path), self.dirs.thumbnail_dir
+ with (self.dirs.thumbnail_dir / f"{doc.pk:07d}.webp").open("wb") as f:
f.write(content_thumbnail)
response = self.client.get(f"/api/documents/{doc.pk}/download/")
content = b"This is a test"
content_thumbnail = b"thumbnail content"
- with open(filename, "wb") as f:
+ with Path(filename).open("wb") as f:
f.write(content)
user1 = User.objects.create_user(username="test1")
doc = Document.objects.create(
title="none",
- filename=os.path.basename(filename),
+ filename=Path(filename).name,
mime_type="application/pdf",
owner=user1,
)
- with open(
- os.path.join(self.dirs.thumbnail_dir, f"{doc.pk:07d}.webp"),
- "wb",
- ) as f:
+ with (Path(self.dirs.thumbnail_dir) / f"{doc.pk:07d}.webp").open("wb") as f:
f.write(content_thumbnail)
response = self.client.get(f"/api/documents/{doc.pk}/download/")
mime_type="application/pdf",
)
- with open(doc.source_path, "wb") as f:
+ with Path(doc.source_path).open("wb") as f:
f.write(content)
- with open(doc.archive_path, "wb") as f:
+ with Path(doc.archive_path).open("wb") as f:
f.write(content_archive)
response = self.client.get(f"/api/documents/{doc.pk}/download/")
def test_document_actions_not_existing_file(self):
doc = Document.objects.create(
title="none",
- filename=os.path.basename("asd"),
+ filename=Path("asd").name,
mime_type="application/pdf",
)
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"documenst": f},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.zip"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.zip").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "title": "my custom title"},
)
c = Correspondent.objects.create(name="test-corres")
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "correspondent": c.id},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "correspondent": 3456},
)
dt = DocumentType.objects.create(name="invoice")
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "document_type": dt.id},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "document_type": 34578},
)
sp = StoragePath.objects.create(name="invoices")
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "storage_path": sp.id},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "storage_path": 34578},
t1 = Tag.objects.create(name="tag1")
t2 = Tag.objects.create(name="tag2")
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "tags": [t2.id, t1.id]},
t1 = Tag.objects.create(name="tag1")
t2 = Tag.objects.create(name="tag2")
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "tags": [t2.id, t1.id, 734563]},
0,
tzinfo=zoneinfo.ZoneInfo("America/Los_Angeles"),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "created": created},
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f, "archive_serial_number": 500},
data_type=CustomField.FieldDataType.STRING,
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "simple.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{
id=str(uuid.uuid4()),
)
- with open(
- os.path.join(os.path.dirname(__file__), "samples", "invalid_pdf.pdf"),
- "rb",
- ) as f:
+ with (Path(__file__).parent / "samples" / "invalid_pdf.pdf").open("rb") as f:
response = self.client.post(
"/api/documents/post_document/",
{"document": f},
archive_filename="archive.pdf",
)
- source_file = os.path.join(
- os.path.dirname(__file__),
- "samples",
- "documents",
- "thumbnails",
- "0000001.webp",
+ source_file: Path = (
+ Path(__file__).parent
+ / "samples"
+ / "documents"
+ / "thumbnails"
+ / "0000001.webp"
)
- archive_file = os.path.join(os.path.dirname(__file__), "samples", "simple.pdf")
+ archive_file: Path = Path(__file__).parent / "samples" / "simple.pdf"
shutil.copy(source_file, doc.source_path)
shutil.copy(archive_file, doc.archive_path)
self.assertGreater(len(meta["archive_metadata"]), 0)
self.assertEqual(meta["media_filename"], "file.pdf")
self.assertEqual(meta["archive_media_filename"], "archive.pdf")
- self.assertEqual(meta["original_size"], os.stat(source_file).st_size)
- self.assertEqual(meta["archive_size"], os.stat(archive_file).st_size)
+ self.assertEqual(meta["original_size"], Path(source_file).stat().st_size)
+ self.assertEqual(meta["archive_size"], Path(archive_file).stat().st_size)
response = self.client.get(f"/api/documents/{doc.pk}/metadata/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
mime_type="application/pdf",
)
- shutil.copy(
- os.path.join(os.path.dirname(__file__), "samples", "simple.pdf"),
- doc.source_path,
- )
+ shutil.copy(Path(__file__).parent / "samples" / "simple.pdf", doc.source_path)
response = self.client.get(f"/api/documents/{doc.pk}/metadata/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_get_logs(self):
log_data = "test\ntest2\n"
- with open(os.path.join(settings.LOGGING_DIR, "mail.log"), "w") as f:
+ with (Path(settings.LOGGING_DIR) / "mail.log").open("w") as f:
f.write(log_data)
- with open(os.path.join(settings.LOGGING_DIR, "paperless.log"), "w") as f:
+ with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f:
f.write(log_data)
response = self.client.get("/api/logs/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_get_logs_only_when_exist(self):
log_data = "test\ntest2\n"
- with open(os.path.join(settings.LOGGING_DIR, "paperless.log"), "w") as f:
+ with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f:
f.write(log_data)
response = self.client.get("/api/logs/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
def test_get_log(self):
log_data = "test\ntest2\n"
- with open(os.path.join(settings.LOGGING_DIR, "paperless.log"), "w") as f:
+ with (Path(settings.LOGGING_DIR) / "paperless.log").open("w") as f:
f.write(log_data)
response = self.client.get("/api/logs/paperless/")
self.assertEqual(response.status_code, status.HTTP_200_OK)
projects / thirdparty / paperless-ngx.git / commitdiff
