import filecmp
-import os
import shutil
from pathlib import Path
from threading import Thread
print("Consumed a perfectly valid file.") # noqa: T201
def slow_write_file(self, target, *, incomplete=False):
- with open(self.sample_file, "rb") as f:
+ with Path(self.sample_file).open("rb") as f:
pdf_bytes = f.read()
if incomplete:
pdf_bytes = pdf_bytes[: len(pdf_bytes) - 100]
- with open(target, "wb") as f:
+ with Path(target).open("wb") as f:
# this will take 2 seconds, since the file is about 20k.
print("Start writing file.") # noqa: T201
for b in chunked(1000, pdf_bytes):
def test_consume_file(self):
self.t_start()
- f = Path(os.path.join(self.dirs.consumption_dir, "my_file.pdf"))
+ f = Path(self.dirs.consumption_dir) / "my_file.pdf"
shutil.copy(self.sample_file, f)
self.wait_for_task_mock_call()
def test_consume_file_invalid_ext(self):
self.t_start()
- f = os.path.join(self.dirs.consumption_dir, "my_file.wow")
+ f = Path(self.dirs.consumption_dir) / "my_file.wow"
shutil.copy(self.sample_file, f)
self.wait_for_task_mock_call()
self.consume_file_mock.assert_not_called()
def test_consume_existing_file(self):
- f = Path(os.path.join(self.dirs.consumption_dir, "my_file.pdf"))
+ f = Path(self.dirs.consumption_dir) / "my_file.pdf"
shutil.copy(self.sample_file, f)
self.t_start()
self.t_start()
- fname = Path(os.path.join(self.dirs.consumption_dir, "my_file.pdf"))
+ fname = Path(self.dirs.consumption_dir) / "my_file.pdf"
self.slow_write_file(fname)
self.t_start()
- fname = Path(os.path.join(self.dirs.consumption_dir, "my_file.~df"))
- fname2 = Path(os.path.join(self.dirs.consumption_dir, "my_file.pdf"))
+ fname = Path(self.dirs.consumption_dir) / "my_file.~df"
+ fname2 = Path(self.dirs.consumption_dir) / "my_file.pdf"
self.slow_write_file(fname)
shutil.move(fname, fname2)
self.t_start()
- fname = Path(os.path.join(self.dirs.consumption_dir, "my_file.pdf"))
+ fname = Path(self.dirs.consumption_dir) / "my_file.pdf"
self.slow_write_file(fname, incomplete=True)
self.wait_for_task_mock_call()
shutil.copy(
self.sample_file,
- os.path.join(self.dirs.consumption_dir, ".DS_STORE"),
+ Path(self.dirs.consumption_dir) / ".DS_STORE",
)
shutil.copy(
self.sample_file,
- os.path.join(self.dirs.consumption_dir, "my_file.pdf"),
+ Path(self.dirs.consumption_dir) / "my_file.pdf",
)
shutil.copy(
self.sample_file,
- os.path.join(self.dirs.consumption_dir, "._my_file.pdf"),
+ Path(self.dirs.consumption_dir) / "._my_file.pdf",
)
shutil.copy(
self.sample_file,
- os.path.join(self.dirs.consumption_dir, "my_second_file.pdf"),
+ Path(self.dirs.consumption_dir) / "my_second_file.pdf",
)
shutil.copy(
self.sample_file,
- os.path.join(self.dirs.consumption_dir, "._my_second_file.pdf"),
+ Path(self.dirs.consumption_dir) / "._my_second_file.pdf",
)
sleep(5)
def test_is_ignored(self):
test_paths = [
{
- "path": os.path.join(self.dirs.consumption_dir, "foo.pdf"),
+ "path": (Path(self.dirs.consumption_dir) / "foo.pdf").as_posix(),
"ignore": False,
},
{
- "path": os.path.join(self.dirs.consumption_dir, "foo", "bar.pdf"),
+ "path": (
+ Path(self.dirs.consumption_dir) / "foo" / "bar.pdf"
+ ).as_posix(),
"ignore": False,
},
{
- "path": os.path.join(self.dirs.consumption_dir, ".DS_STORE"),
+ "path": (Path(self.dirs.consumption_dir) / ".DS_STORE").as_posix(),
"ignore": True,
},
{
- "path": os.path.join(self.dirs.consumption_dir, ".DS_Store"),
+ "path": (Path(self.dirs.consumption_dir) / ".DS_Store").as_posix(),
"ignore": True,
},
{
- "path": os.path.join(self.dirs.consumption_dir, ".stfolder", "foo.pdf"),
+ "path": (
+ Path(self.dirs.consumption_dir) / ".stfolder" / "foo.pdf"
+ ).as_posix(),
"ignore": True,
},
{
- "path": os.path.join(self.dirs.consumption_dir, ".stfolder.pdf"),
+ "path": (Path(self.dirs.consumption_dir) / ".stfolder.pdf").as_posix(),
"ignore": False,
},
{
- "path": os.path.join(
- self.dirs.consumption_dir,
- ".stversions",
- "foo.pdf",
- ),
+ "path": (
+ Path(self.dirs.consumption_dir) / ".stversions" / "foo.pdf"
+ ).as_posix(),
"ignore": True,
},
{
- "path": os.path.join(self.dirs.consumption_dir, ".stversions.pdf"),
+ "path": (
+ Path(self.dirs.consumption_dir) / ".stversions.pdf"
+ ).as_posix(),
"ignore": False,
},
{
- "path": os.path.join(self.dirs.consumption_dir, "._foo.pdf"),
+ "path": (Path(self.dirs.consumption_dir) / "._foo.pdf").as_posix(),
"ignore": True,
},
{
- "path": os.path.join(self.dirs.consumption_dir, "my_foo.pdf"),
+ "path": (Path(self.dirs.consumption_dir) / "my_foo.pdf").as_posix(),
"ignore": False,
},
{
- "path": os.path.join(self.dirs.consumption_dir, "._foo", "bar.pdf"),
+ "path": (
+ Path(self.dirs.consumption_dir) / "._foo" / "bar.pdf"
+ ).as_posix(),
"ignore": True,
},
{
- "path": os.path.join(
- self.dirs.consumption_dir,
- "@eaDir",
- "SYNO@.fileindexdb",
- "_1jk.fnm",
- ),
+ "path": (
+ Path(self.dirs.consumption_dir)
+ / "@eaDir"
+ / "SYNO@.fileindexdb"
+ / "_1jk.fnm"
+ ).as_posix(),
"ignore": True,
},
]
self.t_start()
- f = os.path.join(self.dirs.consumption_dir, "my_file.pdf")
+ f = Path(self.dirs.consumption_dir) / "my_file.pdf"
shutil.copy(self.sample_file, f)
self.wait_for_task_mock_call()
self.t_start()
- path = os.path.join(self.dirs.consumption_dir, *tag_names)
- os.makedirs(path, exist_ok=True)
- f = Path(os.path.join(path, "my_file.pdf"))
+ path = Path(self.dirs.consumption_dir) / "/".join(tag_names)
+ path.mkdir(parents=True, exist_ok=True)
+ f = path / "my_file.pdf"
# Wait at least inotify read_delay for recursive watchers
# to be created for the new directories
sleep(1)
import hashlib
import json
-import os
import shutil
import tempfile
from io import StringIO
call_command(*args)
- with open(os.path.join(self.target, "manifest.json")) as f:
+ with (self.target / "manifest.json").open() as f:
manifest = json.load(f)
return manifest
def test_exporter(self, *, use_filename_format=False):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
num_permission_objects = Permission.objects.count()
4,
)
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ self.assertIsFile((self.target / "manifest.json").as_posix())
self.assertEqual(
self._get_document_from_manifest(manifest, self.d1.id)["fields"]["title"],
for element in manifest:
if element["model"] == "documents.document":
- fname = os.path.join(
- self.target,
- element[document_exporter.EXPORTER_FILE_NAME],
- )
+ fname = (
+ self.target / element[document_exporter.EXPORTER_FILE_NAME]
+ ).as_posix()
self.assertIsFile(fname)
self.assertIsFile(
- os.path.join(
- self.target,
- element[document_exporter.EXPORTER_THUMBNAIL_NAME],
- ),
+ (
+ self.target / element[document_exporter.EXPORTER_THUMBNAIL_NAME]
+ ).as_posix(),
)
- with open(fname, "rb") as f:
+ with Path(fname).open("rb") as f:
checksum = hashlib.md5(f.read()).hexdigest()
self.assertEqual(checksum, element["fields"]["checksum"])
)
if document_exporter.EXPORTER_ARCHIVE_NAME in element:
- fname = os.path.join(
- self.target,
- element[document_exporter.EXPORTER_ARCHIVE_NAME],
- )
+ fname = (
+ self.target / element[document_exporter.EXPORTER_ARCHIVE_NAME]
+ ).as_posix()
self.assertIsFile(fname)
- with open(fname, "rb") as f:
+ with Path(fname).open("rb") as f:
checksum = hashlib.md5(f.read()).hexdigest()
self.assertEqual(checksum, element["fields"]["archive_checksum"])
self.assertEqual(len(messages), 0)
def test_exporter_with_filename_format(self):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
with override_settings(
self.test_exporter(use_filename_format=True)
def test_update_export_changed_time(self):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
self._do_export()
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ self.assertIsFile((self.target / "manifest.json").as_posix())
- st_mtime_1 = os.stat(os.path.join(self.target, "manifest.json")).st_mtime
+ st_mtime_1 = (self.target / "manifest.json").stat().st_mtime
with mock.patch(
"documents.management.commands.document_exporter.copy_file_with_basic_stats",
self._do_export()
m.assert_not_called()
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
- st_mtime_2 = os.stat(os.path.join(self.target, "manifest.json")).st_mtime
+ self.assertIsFile((self.target / "manifest.json").as_posix())
+ st_mtime_2 = (self.target / "manifest.json").stat().st_mtime
Path(self.d1.source_path).touch()
self._do_export()
self.assertEqual(m.call_count, 1)
- st_mtime_3 = os.stat(os.path.join(self.target, "manifest.json")).st_mtime
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ st_mtime_3 = (self.target / "manifest.json").stat().st_mtime
+ self.assertIsFile((self.target / "manifest.json").as_posix())
self.assertNotEqual(st_mtime_1, st_mtime_2)
self.assertNotEqual(st_mtime_2, st_mtime_3)
self._do_export(compare_json=True)
- st_mtime_4 = os.stat(os.path.join(self.target, "manifest.json")).st_mtime
+ st_mtime_4 = (self.target / "manifest.json").stat().st_mtime
self.assertEqual(st_mtime_3, st_mtime_4)
def test_update_export_changed_checksum(self):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
self._do_export()
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ self.assertIsFile((self.target / "manifest.json").as_posix())
with mock.patch(
"documents.management.commands.document_exporter.copy_file_with_basic_stats",
self._do_export()
m.assert_not_called()
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ self.assertIsFile((self.target / "manifest.json").as_posix())
self.d2.checksum = "asdfasdgf3"
self.d2.save()
self._do_export(compare_checksums=True)
self.assertEqual(m.call_count, 1)
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ self.assertIsFile((self.target / "manifest.json").as_posix())
def test_update_export_deleted_document(self):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
manifest = self._do_export()
self.assertTrue(len(manifest), 7)
doc_from_manifest = self._get_document_from_manifest(manifest, self.d3.id)
self.assertIsFile(
- os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME]),
+ (self.target / doc_from_manifest[EXPORTER_FILE_NAME]).as_posix(),
)
self.d3.delete()
self.d3.id,
)
self.assertIsFile(
- os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME]),
+ (self.target / doc_from_manifest[EXPORTER_FILE_NAME]).as_posix(),
)
manifest = self._do_export(delete=True)
self.assertIsNotFile(
- os.path.join(self.target, doc_from_manifest[EXPORTER_FILE_NAME]),
+ (self.target / doc_from_manifest[EXPORTER_FILE_NAME]).as_posix(),
)
self.assertTrue(len(manifest), 6)
@override_settings(FILENAME_FORMAT="{title}/{correspondent}")
def test_update_export_changed_location(self):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
self._do_export(use_filename_format=True)
- self.assertIsFile(os.path.join(self.target, "wow1", "c.pdf"))
+ self.assertIsFile((self.target / "wow1" / "c.pdf").as_posix())
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
+ self.assertIsFile((self.target / "manifest.json").as_posix())
self.d1.title = "new_title"
self.d1.save()
self._do_export(use_filename_format=True, delete=True)
- self.assertIsNotFile(os.path.join(self.target, "wow1", "c.pdf"))
- self.assertIsNotDir(os.path.join(self.target, "wow1"))
- self.assertIsFile(os.path.join(self.target, "new_title", "c.pdf"))
- self.assertIsFile(os.path.join(self.target, "manifest.json"))
- self.assertIsFile(os.path.join(self.target, "wow2", "none.pdf"))
+ self.assertIsNotFile((self.target / "wow1" / "c.pdf").as_posix())
+ self.assertIsNotDir((self.target / "wow1").as_posix())
+ self.assertIsFile((self.target / "new_title" / "c.pdf").as_posix())
+ self.assertIsFile((self.target / "manifest.json").as_posix())
+ self.assertIsFile((self.target / "wow2" / "none.pdf").as_posix())
self.assertIsFile(
- os.path.join(self.target, "wow2", "none_01.pdf"),
+ (self.target / "wow2" / "none_01.pdf").as_posix(),
)
def test_export_missing_files(self):
- Zipfile is created
- Zipfile contains exported files
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
args = ["document_exporter", self.target, "--zip"]
call_command(*args)
- expected_file = os.path.join(
- self.target,
- f"export-{timezone.localdate().isoformat()}.zip",
- )
+ expected_file = (
+ self.target / f"export-{timezone.localdate().isoformat()}.zip"
+ ).as_posix()
self.assertIsFile(expected_file)
- Zipfile is created
- Zipfile contains exported files
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
args = ["document_exporter", self.target, "--zip", "--use-filename-format"]
):
call_command(*args)
- expected_file = os.path.join(
- self.target,
- f"export-{timezone.localdate().isoformat()}.zip",
- )
+ expected_file = (
+ self.target / f"export-{timezone.localdate().isoformat()}.zip"
+ ).as_posix()
self.assertIsFile(expected_file)
- Zipfile contains exported files
- The existing file and directory in target are removed
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
# Create stuff in target directory
call_command(*args)
- expected_file = os.path.join(
- self.target,
- f"export-{timezone.localdate().isoformat()}.zip",
- )
+ expected_file = (
+ self.target / f"export-{timezone.localdate().isoformat()}.zip"
+ ).as_posix()
self.assertIsFile(expected_file)
self.assertIsNotFile(existing_file)
- Error is raised
"""
with tempfile.TemporaryDirectory() as tmp_dir:
- os.chmod(tmp_dir, 0o000)
+ Path(tmp_dir).chmod(0o000)
args = ["document_exporter", tmp_dir]
- Manifest.json doesn't contain information about archive files
- Documents can be imported again
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
manifest = self._do_export()
- Manifest.json doesn't contain information about thumbnails
- Documents can be imported again
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
manifest = self._do_export()
- Main manifest.json file doesn't contain information about documents
- Documents can be imported again
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
manifest = self._do_export(split_manifest=True)
THEN:
- Documents can be imported again
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
self._do_export(use_folder_prefix=True)
- ContentType & Permission objects are not deleted, db transaction rolled back
"""
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
num_content_type_objects = ContentType.objects.count()
self.assertEqual(Permission.objects.count(), num_permission_objects + 1)
def test_exporter_with_auditlog_disabled(self):
- shutil.rmtree(os.path.join(self.dirs.media_dir, "documents"))
+ shutil.rmtree(Path(self.dirs.media_dir) / "documents")
shutil.copytree(
- os.path.join(os.path.dirname(__file__), "samples", "documents"),
- os.path.join(self.dirs.media_dir, "documents"),
+ Path(__file__).parent / "samples" / "documents",
+ Path(self.dirs.media_dir) / "documents",
)
with override_settings(