555 lines
25 KiB
Python
555 lines
25 KiB
Python
import contextlib
|
|
import datetime
|
|
import os
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import struct
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import unifydrive2nas as app
|
|
|
|
|
|
def atom(kind, payload):
|
|
return struct.pack(">I4s", len(payload) + 8, kind) + payload
|
|
|
|
|
|
def atoms_in(content, start=0, end=None):
|
|
if end is None:
|
|
end = len(content)
|
|
result = []
|
|
position = start
|
|
while position + 8 <= end:
|
|
size, kind = struct.unpack(">I4s", content[position:position + 8])
|
|
result.append((kind, position + 8, position + size))
|
|
position += size
|
|
return result
|
|
|
|
|
|
def tiff_with_exif_datetime(text, tag=0x9003):
|
|
return tiff_with_exif_datetimes(((tag, text),))
|
|
|
|
|
|
def tiff_with_exif_datetimes(values):
|
|
encoded_values = [(tag, text.encode("ascii") + b"\x00") for tag, text in values]
|
|
# IFD0 の ExifIFDPointer は TIFF 先頭から 26、値は Exif IFD の後に置く。
|
|
ifd0 = struct.pack("<H", 1)
|
|
ifd0 += struct.pack("<HHI", 0x8769, 4, 1) + struct.pack("<I", 26)
|
|
ifd0 += struct.pack("<I", 0)
|
|
value_offset = 26 + 2 + 12 * len(encoded_values) + 4
|
|
exif_ifd = struct.pack("<H", len(encoded_values))
|
|
payload = b""
|
|
for tag, encoded in encoded_values:
|
|
exif_ifd += struct.pack("<HHI", tag, 2, len(encoded)) + struct.pack("<I", value_offset)
|
|
payload += encoded
|
|
value_offset += len(encoded)
|
|
exif_ifd += struct.pack("<I", 0)
|
|
return b"II*\x00" + struct.pack("<I", 8) + ifd0 + exif_ifd + payload
|
|
|
|
|
|
def jpeg_with_exif(text):
|
|
exif = b"Exif\x00\x00" + tiff_with_exif_datetime(text)
|
|
return b"\xff\xd8\xff\xe1" + struct.pack(">H", len(exif) + 2) + exif + b"\xff\xd9"
|
|
|
|
|
|
def jfif_without_exif():
|
|
payload = b"JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"
|
|
return b"\xff\xd8\xff\xe0" + struct.pack(">H", len(payload) + 2) + payload + b"\xff\xd9"
|
|
|
|
|
|
def png_with_exif(text):
|
|
signature = b"\x89PNG\r\n\x1a\n"
|
|
exif = tiff_with_exif_datetime(text)
|
|
# exifread は日時テストでは CRC を参照しないため最小値でよい。
|
|
return signature + struct.pack(">I", len(exif)) + b"eXIf" + exif + b"\x00\x00\x00\x00"
|
|
|
|
|
|
def movie_with_metadata(
|
|
apple_text=None,
|
|
mvhd_seconds=None,
|
|
mvhd_version=0,
|
|
quicktime_meta=False,
|
|
metadata_key=b"com.apple.quicktime.creationdate",
|
|
item_index=1,
|
|
):
|
|
children = b""
|
|
if mvhd_seconds is not None:
|
|
if mvhd_version == 0:
|
|
payload = b"\x00\x00\x00\x00" + struct.pack(">I", mvhd_seconds)
|
|
else:
|
|
payload = b"\x01\x00\x00\x00" + struct.pack(">Q", mvhd_seconds)
|
|
children += atom(b"mvhd", payload)
|
|
if apple_text is not None:
|
|
key = metadata_key
|
|
keys = atom(b"keys", b"\x00\x00\x00\x00" + struct.pack(">I", 1) + struct.pack(">I4s", len(key) + 8, b"mdta") + key)
|
|
data = atom(b"data", b"\x00\x00\x00\x01" + b"\x00\x00\x00\x00" + apple_text.encode("utf-8"))
|
|
item = atom(struct.pack(">I", item_index), data)
|
|
ilst = atom(b"ilst", item)
|
|
if quicktime_meta:
|
|
# 実際の iPhone MOV と同じく、version/flags 無しで hdlr から始める。
|
|
meta_payload = atom(b"hdlr", b"\x00" * 26) + keys + ilst
|
|
else:
|
|
meta_payload = b"\x00\x00\x00\x00" + keys + ilst
|
|
children += atom(b"meta", meta_payload)
|
|
return atom(b"ftyp", b"qt \x00\x00\x00\x00") + atom(b"moov", children)
|
|
|
|
|
|
class Workspace:
|
|
def __init__(self, testcase):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
testcase.addCleanup(self.temporary.cleanup)
|
|
root = Path(self.temporary.name)
|
|
self.source = root / "source"
|
|
self.destination = root / "destination"
|
|
self.database = root / "state" / "copied.sqlite3"
|
|
self.source.mkdir()
|
|
|
|
def config(self, initialize=False, dry_run=False):
|
|
return app.Config(
|
|
source=self.source,
|
|
destination=self.destination,
|
|
database=self.database,
|
|
initialize=initialize,
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
def run(self, **kwargs):
|
|
return app.run(self.config(**kwargs), mount_checker=lambda unused: True)
|
|
|
|
|
|
class DateDetectionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.workspace = Workspace(self)
|
|
|
|
def write(self, name, content, mtime=None):
|
|
path = self.workspace.source / name
|
|
path.write_bytes(content)
|
|
if mtime is not None:
|
|
os.utime(str(path), (mtime, mtime))
|
|
return path
|
|
|
|
def test_exif_original_is_used_for_jpeg(self):
|
|
path = self.write("photo.jpg", jpeg_with_exif("2024:02:03 04:05:06"))
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("exif", source)
|
|
self.assertEqual("20240203", value.strftime("%Y%m%d"))
|
|
|
|
def test_exif_is_read_from_png_chunk(self):
|
|
path = self.write("screen.PNG", png_with_exif("2023:12:31 23:59:58"))
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("exif", source)
|
|
self.assertEqual("20231231", value.strftime("%Y%m%d"))
|
|
|
|
def test_digitized_date_is_used_when_original_is_absent(self):
|
|
exif = b"Exif\x00\x00" + tiff_with_exif_datetime("2020:06:07 08:09:10", tag=0x9004)
|
|
content = b"\xff\xd8\xff\xe1" + struct.pack(">H", len(exif) + 2) + exif + b"\xff\xd9"
|
|
path = self.write("digitized.jpg", content)
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("exif", source)
|
|
self.assertEqual("20200607", value.strftime("%Y%m%d"))
|
|
|
|
def test_original_date_precedes_digitized_when_both_exist(self):
|
|
exif = b"Exif\x00\x00" + tiff_with_exif_datetimes(
|
|
(
|
|
(0x9003, "2024:01:02 03:04:05"),
|
|
(0x9004, "1999:12:31 23:59:59"),
|
|
)
|
|
)
|
|
content = b"\xff\xd8\xff\xe1" + struct.pack(">H", len(exif) + 2) + exif + b"\xff\xd9"
|
|
path = self.write("both.jpg", content)
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("exif", source)
|
|
self.assertEqual("20240102 030405", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_apple_date_precedes_mvhd_and_is_converted_to_jst(self):
|
|
seconds = int((datetime.datetime(2020, 1, 1, tzinfo=app.UTC) - app.QUICKTIME_EPOCH).total_seconds())
|
|
path = self.write("clip.mov", movie_with_metadata("2025-11-08T20:46:34+0900", seconds))
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("apple", source)
|
|
self.assertEqual("20251108 204634", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_apple_date_in_quicktime_meta_without_fullbox_header(self):
|
|
seconds = int((datetime.datetime(2020, 1, 1, tzinfo=app.UTC) - app.QUICKTIME_EPOCH).total_seconds())
|
|
path = self.write(
|
|
"iphone.mov",
|
|
movie_with_metadata(
|
|
"2025-11-08T20:46:34+0900", seconds, quicktime_meta=True
|
|
),
|
|
)
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("apple", source)
|
|
self.assertEqual("20251108 204634", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_quicktime_meta_fixture_matches_observed_atom_layout(self):
|
|
apple_text = "2025-11-08T20:46:34+0900"
|
|
content = movie_with_metadata(apple_text, quicktime_meta=True)
|
|
top = atoms_in(content)
|
|
moov = next(item for item in top if item[0] == b"moov")
|
|
meta = next(item for item in atoms_in(content, moov[1], moov[2]) if item[0] == b"meta")
|
|
children = atoms_in(content, meta[1], meta[2])
|
|
self.assertEqual([b"hdlr", b"keys", b"ilst"], [item[0] for item in children])
|
|
|
|
keys = children[1]
|
|
self.assertEqual(b"\x00\x00\x00\x00", content[keys[1]:keys[1] + 4])
|
|
self.assertEqual(1, struct.unpack(">I", content[keys[1] + 4:keys[1] + 8])[0])
|
|
entry_size, namespace = struct.unpack(">I4s", content[keys[1] + 8:keys[1] + 16])
|
|
self.assertEqual(b"mdta", namespace)
|
|
self.assertEqual(
|
|
b"com.apple.quicktime.creationdate",
|
|
content[keys[1] + 16:keys[1] + 8 + entry_size],
|
|
)
|
|
|
|
ilst = children[2]
|
|
item = atoms_in(content, ilst[1], ilst[2])[0]
|
|
self.assertEqual(struct.pack(">I", 1), item[0])
|
|
data = atoms_in(content, item[1], item[2])[0]
|
|
self.assertEqual(b"data", data[0])
|
|
self.assertEqual(b"\x00\x00\x00\x01\x00\x00\x00\x00", content[data[1]:data[1] + 8])
|
|
self.assertEqual(apple_text.encode("utf-8"), content[data[1] + 8:data[2]])
|
|
|
|
def test_missing_apple_key_or_ilst_item_falls_back_to_mvhd_in_both_meta_forms(self):
|
|
instant = datetime.datetime(2024, 5, 6, 7, 8, 9, tzinfo=app.UTC)
|
|
seconds = int((instant - app.QUICKTIME_EPOCH).total_seconds())
|
|
cases = (
|
|
("missing-key", {"metadata_key": b"com.apple.quicktime.location.ISO6709"}),
|
|
("missing-item", {"item_index": 2}),
|
|
)
|
|
for quicktime_meta in (False, True):
|
|
for label, options in cases:
|
|
with self.subTest(quicktime_meta=quicktime_meta, case=label):
|
|
path = self.write(
|
|
"{0}-{1}.mov".format("quicktime" if quicktime_meta else "fullbox", label),
|
|
movie_with_metadata(
|
|
"2025-11-08T20:46:34+0900",
|
|
seconds,
|
|
quicktime_meta=quicktime_meta,
|
|
**options
|
|
),
|
|
)
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mvhd", source)
|
|
self.assertEqual("20240506 160809", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_apple_date_uses_its_own_non_jst_offset(self):
|
|
seconds = int((datetime.datetime(2020, 1, 1, tzinfo=app.UTC) - app.QUICKTIME_EPOCH).total_seconds())
|
|
path = self.write("offset.mov", movie_with_metadata("2025-01-01T20:30:00-0500", seconds))
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("apple", source)
|
|
self.assertEqual("20250102 103000", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_mvhd_versions_zero_and_one_are_supported(self):
|
|
instant = datetime.datetime(2024, 12, 31, 18, 0, tzinfo=app.UTC)
|
|
seconds = int((instant - app.QUICKTIME_EPOCH).total_seconds())
|
|
for version in (0, 1):
|
|
with self.subTest(version=version):
|
|
path = self.write("clip{0}.mp4".format(version), movie_with_metadata(mvhd_seconds=seconds, mvhd_version=version))
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mvhd", source)
|
|
self.assertEqual("20250101 030000", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_mvhd_jst_day_boundary(self):
|
|
cases = (
|
|
(datetime.datetime(2024, 12, 31, 14, 59, 59, tzinfo=app.UTC), "20241231 235959"),
|
|
(datetime.datetime(2024, 12, 31, 15, 0, 0, tzinfo=app.UTC), "20250101 000000"),
|
|
)
|
|
for index, (instant, expected) in enumerate(cases):
|
|
with self.subTest(instant=instant):
|
|
seconds = int((instant - app.QUICKTIME_EPOCH).total_seconds())
|
|
path = self.write("boundary{0}.mp4".format(index), movie_with_metadata(mvhd_seconds=seconds))
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mvhd", source)
|
|
self.assertEqual(expected, value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_metadata_parser_exceptions_fall_through_candidates(self):
|
|
expected = datetime.datetime(2022, 5, 7, 0, 30, tzinfo=app.JST)
|
|
path = self.write("error.mov", b"broken", expected.timestamp())
|
|
mvhd = datetime.datetime(2020, 2, 3, 4, 5, tzinfo=app.JST)
|
|
with mock.patch.object(app, "parse_exif_datetime", side_effect=ValueError("bad exif")):
|
|
with mock.patch.object(app, "parse_video_datetimes", return_value=(None, mvhd)):
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual((mvhd, "mvhd"), (value, source))
|
|
|
|
with mock.patch.object(app, "parse_exif_datetime", side_effect=ValueError("bad exif")):
|
|
with mock.patch.object(app, "parse_video_datetimes", side_effect=ValueError("bad movie")):
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mtime", source)
|
|
self.assertEqual("20220507 003000", value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_zero_mvhd_and_broken_file_fall_back_to_mtime(self):
|
|
expected = datetime.datetime(2022, 5, 6, 12, 0, tzinfo=app.JST)
|
|
cases = (
|
|
("zero-v0.mp4", movie_with_metadata(mvhd_seconds=0, mvhd_version=0)),
|
|
("zero-v1.mp4", movie_with_metadata(mvhd_seconds=0, mvhd_version=1)),
|
|
("broken.mov", b"broken"),
|
|
)
|
|
for name, content in cases:
|
|
with self.subTest(name=name):
|
|
path = self.write(name, content, expected.timestamp())
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mtime", source)
|
|
self.assertEqual("20220506", value.strftime("%Y%m%d"))
|
|
|
|
def test_mtime_is_interpreted_in_jst_across_day_boundary(self):
|
|
cases = (
|
|
(datetime.datetime(2024, 12, 31, 14, 59, 59, tzinfo=app.UTC), "20241231 235959"),
|
|
(datetime.datetime(2024, 12, 31, 15, 0, 0, tzinfo=app.UTC), "20250101 000000"),
|
|
)
|
|
for index, (instant, expected) in enumerate(cases):
|
|
with self.subTest(instant=instant):
|
|
path = self.write("mtime{0}.jpg".format(index), jfif_without_exif(), instant.timestamp())
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mtime", source)
|
|
self.assertEqual(expected, value.strftime("%Y%m%d %H%M%S"))
|
|
|
|
def test_jfif_without_exif_falls_back_to_mtime(self):
|
|
expected = datetime.datetime(2021, 7, 8, 1, 2, tzinfo=app.JST)
|
|
path = self.write("plain.jpeg", jfif_without_exif(), expected.timestamp())
|
|
value, source = app.determine_datetime(path, path.stat())
|
|
self.assertEqual("mtime", source)
|
|
self.assertEqual("20210708", value.strftime("%Y%m%d"))
|
|
|
|
|
|
class CopyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.workspace = Workspace(self)
|
|
self.timestamp = datetime.datetime(2026, 10, 12, 9, 0, tzinfo=app.JST).timestamp()
|
|
|
|
def write(self, relative, content=b"content"):
|
|
path = self.workspace.source / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(content)
|
|
os.utime(str(path), (self.timestamp, self.timestamp))
|
|
return path
|
|
|
|
def copied_path(self, name):
|
|
return self.workspace.destination / "2026" / "20261012" / name
|
|
|
|
def test_exclusions_and_case_insensitive_extensions(self):
|
|
self.write("visible/PHOTO.JpG")
|
|
self.write("visible/raw.ARW")
|
|
self.write("visible/note.txt")
|
|
self.write(".hidden/photo.jpg")
|
|
self.write("visible/.nested/deeper/photo.jpg")
|
|
self.write("visible/.partial.JPG.z0123456789abcdef0123456789abcdef")
|
|
self.write("visible/@eaDir/thumb.jpg")
|
|
summary = self.workspace.run()
|
|
self.assertEqual((1, 1), (summary.targets, summary.copied))
|
|
self.assertTrue(self.copied_path("PHOTO.JpG").is_file())
|
|
|
|
def test_all_supported_extensions_are_case_insensitive(self):
|
|
names = ("a.JPG", "b.Jpeg", "c.HEIC", "d.HeIf", "e.PNG", "f.MOV", "g.Mp4")
|
|
for name in names:
|
|
self.write("album/" + name)
|
|
summary = self.workspace.run()
|
|
self.assertEqual((len(names), len(names)), (summary.targets, summary.copied))
|
|
self.assertEqual(sorted(names), sorted(path.name for path in self.copied_path("a.JPG").parent.iterdir()))
|
|
|
|
def test_jpeg_content_is_detected_even_with_png_extension(self):
|
|
source = self.write("mismatch.PNG", jpeg_with_exif("2024:03:04 05:06:07"))
|
|
value, date_source = app.determine_datetime(source, source.stat())
|
|
self.assertEqual("exif", date_source)
|
|
self.assertEqual("20240304", value.strftime("%Y%m%d"))
|
|
|
|
def test_same_name_same_content_is_recorded_without_copy(self):
|
|
self.write("album/photo.jpg", b"same")
|
|
existing = self.copied_path("photo.jpg")
|
|
existing.parent.mkdir(parents=True)
|
|
existing.write_bytes(b"same")
|
|
summary = self.workspace.run()
|
|
self.assertEqual(1, summary.identical)
|
|
self.assertEqual(0, summary.copied)
|
|
self.assertEqual(b"same", existing.read_bytes())
|
|
|
|
def test_same_name_different_content_gets_sequence_number(self):
|
|
self.write("album/photo.jpg", b"new")
|
|
existing = self.copied_path("photo.jpg")
|
|
existing.parent.mkdir(parents=True)
|
|
existing.write_bytes(b"old")
|
|
summary = self.workspace.run()
|
|
self.assertEqual((1, 1), (summary.copied, summary.numbered))
|
|
self.assertEqual(b"old", existing.read_bytes())
|
|
self.assertEqual(b"new", self.copied_path("photo_1.jpg").read_bytes())
|
|
|
|
def test_sequence_number_skips_multiple_different_collisions(self):
|
|
self.write("album/photo.jpg", b"new")
|
|
directory = self.copied_path("photo.jpg").parent
|
|
directory.mkdir(parents=True)
|
|
self.copied_path("photo.jpg").write_bytes(b"old-zero")
|
|
self.copied_path("photo_1.jpg").write_bytes(b"old-one")
|
|
summary = self.workspace.run()
|
|
self.assertEqual((1, 1), (summary.copied, summary.numbered))
|
|
self.assertEqual(b"new", self.copied_path("photo_2.jpg").read_bytes())
|
|
|
|
def test_second_run_does_not_copy_again(self):
|
|
self.write("photo.jpg")
|
|
first = self.workspace.run()
|
|
second = self.workspace.run()
|
|
self.assertEqual(1, first.copied)
|
|
self.assertEqual((1, 0, 0), (second.targets, second.new, second.copied))
|
|
|
|
def test_size_or_mtime_change_makes_source_new_again(self):
|
|
source = self.write("photo.jpg", b"first")
|
|
self.workspace.run()
|
|
|
|
source.write_bytes(b"second-content")
|
|
os.utime(str(source), ns=(source.stat().st_atime_ns, source.stat().st_mtime_ns + 1_000_000_000))
|
|
changed_content = self.workspace.run()
|
|
self.assertEqual((1, 1, 1), (changed_content.new, changed_content.copied, changed_content.numbered))
|
|
self.assertEqual(b"second-content", self.copied_path("photo_1.jpg").read_bytes())
|
|
|
|
current = source.stat()
|
|
os.utime(str(source), ns=(current.st_atime_ns, current.st_mtime_ns + 1_000_000_000))
|
|
changed_mtime = self.workspace.run()
|
|
self.assertEqual((1, 0), (changed_mtime.new, changed_mtime.copied))
|
|
self.assertEqual(1, changed_mtime.identical)
|
|
with contextlib.closing(sqlite3.connect(str(self.workspace.database))) as connection:
|
|
self.assertEqual(3, connection.execute("SELECT COUNT(*) FROM copied_files").fetchone()[0])
|
|
|
|
def test_database_records_key_destination_and_date_source(self):
|
|
source = self.write("album/photo.jpg", b"plain")
|
|
self.workspace.run()
|
|
with contextlib.closing(sqlite3.connect(str(self.workspace.database))) as connection:
|
|
row = connection.execute(
|
|
"""SELECT source_relative_path, source_size, source_mtime_ns,
|
|
destination_path, date_source, action
|
|
FROM copied_files"""
|
|
).fetchone()
|
|
self.assertEqual("album/photo.jpg", row[0])
|
|
self.assertEqual(source.stat().st_size, row[1])
|
|
self.assertEqual(source.stat().st_mtime_ns, row[2])
|
|
self.assertEqual(str(self.copied_path("photo.jpg")), row[3])
|
|
self.assertEqual(("mtime", "copied"), row[4:6])
|
|
|
|
def test_initialization_records_without_copying(self):
|
|
self.write("photo.jpg")
|
|
initialized = self.workspace.run(initialize=True)
|
|
regular = self.workspace.run()
|
|
self.assertEqual((1, 1, 0), (initialized.targets, initialized.new, initialized.copied))
|
|
self.assertFalse(self.workspace.destination.exists())
|
|
self.assertEqual(0, regular.new)
|
|
with contextlib.closing(sqlite3.connect(str(self.workspace.database))) as connection:
|
|
row = connection.execute("SELECT date_source, action FROM copied_files").fetchone()
|
|
self.assertEqual(("init", "initialized"), row)
|
|
|
|
def test_unmounted_or_empty_source_stops_before_writing(self):
|
|
self.write("photo.jpg")
|
|
with self.assertRaisesRegex(RuntimeError, "マウントポイント"):
|
|
app.run(self.workspace.config(), mount_checker=lambda unused: False)
|
|
self.assertFalse(self.workspace.database.exists())
|
|
|
|
empty = Workspace(self)
|
|
with self.assertRaisesRegex(RuntimeError, "空です"):
|
|
empty.run()
|
|
self.assertFalse(empty.database.exists())
|
|
|
|
def test_one_copy_failure_does_not_stop_later_files(self):
|
|
self.write("a.jpg", b"a")
|
|
self.write("b.jpg", b"b")
|
|
original = app.atomic_copy
|
|
calls = []
|
|
|
|
def fail_first(source, destination):
|
|
calls.append(source.name)
|
|
if source.name == "a.jpg":
|
|
raise OSError("test failure")
|
|
original(source, destination)
|
|
|
|
summary = app.run(
|
|
self.workspace.config(), mount_checker=lambda unused: True, copy_function=fail_first
|
|
)
|
|
self.assertEqual((1, 1), (summary.failed, summary.copied))
|
|
self.assertEqual(["a.jpg", "b.jpg"], calls)
|
|
self.assertTrue(self.copied_path("b.jpg").exists())
|
|
with contextlib.closing(sqlite3.connect(str(self.workspace.database))) as connection:
|
|
rows = connection.execute("SELECT source_relative_path FROM copied_files ORDER BY source_relative_path").fetchall()
|
|
self.assertEqual([("b.jpg",)], rows)
|
|
|
|
def test_source_files_are_not_changed(self):
|
|
source = self.write("photo.jpg", b"immutable")
|
|
before = (source.read_bytes(), source.stat().st_mtime_ns, source.stat().st_mode)
|
|
self.workspace.run()
|
|
after = (source.read_bytes(), source.stat().st_mtime_ns, source.stat().st_mode)
|
|
self.assertEqual(before, after)
|
|
|
|
def test_dry_run_changes_neither_destination_nor_database_rows(self):
|
|
self.write("photo.jpg")
|
|
summary = self.workspace.run(dry_run=True)
|
|
self.assertEqual((1, 1, 0), (summary.targets, summary.new, summary.copied))
|
|
self.assertFalse(self.workspace.destination.exists())
|
|
self.assertFalse(self.workspace.database.exists())
|
|
self.assertFalse(Path(str(self.workspace.database) + ".lock").exists())
|
|
|
|
def test_dry_run_does_not_change_an_existing_database(self):
|
|
self.write("existing.jpg", b"existing")
|
|
self.workspace.run()
|
|
before = self.workspace.database.read_bytes()
|
|
self.write("new.jpg", b"new")
|
|
summary = self.workspace.run(dry_run=True)
|
|
self.assertEqual(1, summary.new)
|
|
self.assertEqual(before, self.workspace.database.read_bytes())
|
|
self.assertFalse(self.copied_path("new.jpg").exists())
|
|
|
|
def test_init_dry_run_writes_nothing(self):
|
|
self.write("photo.jpg")
|
|
summary = self.workspace.run(initialize=True, dry_run=True)
|
|
self.assertEqual((1, 1, 0), (summary.targets, summary.new, summary.copied))
|
|
self.assertFalse(self.workspace.destination.exists())
|
|
self.assertFalse(self.workspace.database.exists())
|
|
|
|
def test_atomic_copy_leaves_no_temporary_file(self):
|
|
source = self.write("photo.jpg")
|
|
self.workspace.run()
|
|
copied = self.copied_path("photo.jpg")
|
|
directory = copied.parent
|
|
self.assertEqual(["photo.jpg"], sorted(path.name for path in directory.iterdir()))
|
|
self.assertEqual(source.stat().st_mtime_ns, copied.stat().st_mtime_ns)
|
|
|
|
def test_atomic_copy_failure_removes_partial_temporary_file(self):
|
|
source = self.write("photo.jpg")
|
|
destination = self.copied_path("photo.jpg")
|
|
destination.parent.mkdir(parents=True)
|
|
|
|
def write_partial_then_fail(unused_source, temporary):
|
|
Path(temporary).write_bytes(b"partial")
|
|
raise OSError("copy failed")
|
|
|
|
with mock.patch.object(app.shutil, "copy2", side_effect=write_partial_then_fail):
|
|
with self.assertRaisesRegex(OSError, "copy failed"):
|
|
app.atomic_copy(source, destination)
|
|
self.assertEqual([], list(destination.parent.iterdir()))
|
|
|
|
|
|
class LockTests(unittest.TestCase):
|
|
def test_lock_rejects_a_second_holder(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "process.lock"
|
|
with app.process_lock(path):
|
|
with self.assertRaisesRegex(RuntimeError, "実行中"):
|
|
with app.process_lock(path):
|
|
self.fail("二重ロックを取得してはならない")
|
|
|
|
|
|
class MainTests(unittest.TestCase):
|
|
def test_success_prints_all_summary_counts_and_returns_zero(self):
|
|
summary = app.Summary(targets=6, new=5, copied=3, identical=1, numbered=2, failed=0)
|
|
with mock.patch.object(app, "run", return_value=summary):
|
|
with mock.patch("builtins.print") as output:
|
|
self.assertEqual(0, app.main([]))
|
|
output.assert_called_once_with(
|
|
"まとめ: 対象=6 新規=5 コピー=3 同一スキップ=1 連番=2 失敗=0"
|
|
)
|
|
|
|
def test_file_failure_returns_one(self):
|
|
with mock.patch.object(app, "run", return_value=app.Summary(failed=1)):
|
|
with mock.patch("builtins.print"):
|
|
self.assertEqual(1, app.main([]))
|
|
|
|
def test_startup_failure_returns_two(self):
|
|
with mock.patch.object(app, "run", side_effect=RuntimeError("test failure")):
|
|
self.assertEqual(2, app.main([]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|