Add dozo tests
|
@ -20,6 +20,12 @@ EXIFTOOL_STAYOPEN_EOF_LEN = len(EXIFTOOL_STAYOPEN_EOF)
|
||||||
EXIFTOOL_PROCESSES = []
|
EXIFTOOL_PROCESSES = []
|
||||||
|
|
||||||
|
|
||||||
|
def exiftool_is_running():
|
||||||
|
ps = subprocess.run(["ps"], capture_output=True)
|
||||||
|
stdout = ps.stdout.decode("utf-8")
|
||||||
|
return "exiftool" in stdout
|
||||||
|
|
||||||
|
|
||||||
@atexit.register
|
@atexit.register
|
||||||
def terminate_exiftool():
|
def terminate_exiftool():
|
||||||
"""Terminate any running ExifTool subprocesses; call this to cleanup when done using ExifTool """
|
"""Terminate any running ExifTool subprocesses; call this to cleanup when done using ExifTool """
|
||||||
|
|
After Width: | Height: | Size: 2.1 MiB |
After Width: | Height: | Size: 2.8 MiB |
After Width: | Height: | Size: 2.9 MiB |
After Width: | Height: | Size: 1.9 MiB |
After Width: | Height: | Size: 2.3 MiB |
After Width: | Height: | Size: 545 KiB |
After Width: | Height: | Size: 532 KiB |
After Width: | Height: | Size: 578 KiB |
After Width: | Height: | Size: 1.5 MiB |
After Width: | Height: | Size: 1.2 MiB |
After Width: | Height: | Size: 1.4 MiB |
After Width: | Height: | Size: 504 KiB |
After Width: | Height: | Size: 500 KiB |
After Width: | Height: | Size: 157 KiB |
After Width: | Height: | Size: 399 KiB |
After Width: | Height: | Size: 123 KiB |
After Width: | Height: | Size: 429 KiB |
After Width: | Height: | Size: 349 KiB |
After Width: | Height: | Size: 353 KiB |
After Width: | Height: | Size: 224 KiB |
After Width: | Height: | Size: 335 KiB |
After Width: | Height: | Size: 545 KiB |
After Width: | Height: | Size: 314 KiB |
After Width: | Height: | Size: 118 KiB |
After Width: | Height: | Size: 277 KiB |
After Width: | Height: | Size: 259 KiB |
After Width: | Height: | Size: 266 KiB |
After Width: | Height: | Size: 266 KiB |
After Width: | Height: | Size: 265 KiB |
After Width: | Height: | Size: 263 KiB |
After Width: | Height: | Size: 4.3 MiB |
After Width: | Height: | Size: 457 KiB |
After Width: | Height: | Size: 2.1 MiB |
After Width: | Height: | Size: 5.0 MiB |
After Width: | Height: | Size: 1.6 MiB |
After Width: | Height: | Size: 1.4 MiB |
After Width: | Height: | Size: 550 KiB |
After Width: | Height: | Size: 453 KiB |
After Width: | Height: | Size: 524 KiB |
|
@ -0,0 +1,13 @@
|
||||||
|
""" pytest test configuration """
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from dozo.exiftool import _ExifToolProc
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def reset_singletons():
|
||||||
|
""" Need to clean up any ExifTool singletons between tests """
|
||||||
|
_ExifToolProc.instance = None
|
||||||
|
|
||||||
|
|
|
@ -0,0 +1,75 @@
|
||||||
|
from configparser import RawConfigParser
|
||||||
|
from pathlib import Path
|
||||||
|
import pytest
|
||||||
|
import shutil
|
||||||
|
import tempfile
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
from dozo import config
|
||||||
|
|
||||||
|
# Helpers
|
||||||
|
import random
|
||||||
|
import string
|
||||||
|
|
||||||
|
def random_char(y):
|
||||||
|
return ''.join(random.choice(string.printable) for x in range(y))
|
||||||
|
|
||||||
|
def write_random_file(file_path):
|
||||||
|
with open(file_path, 'w') as conf_file:
|
||||||
|
conf_file.write(random_char(20))
|
||||||
|
|
||||||
|
class TestConfig:
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def conf_path(self):
|
||||||
|
tmp_path = tempfile.mkdtemp(prefix='dozo-')
|
||||||
|
yield Path(tmp_path, "dozo.conf")
|
||||||
|
shutil.rmtree(tmp_path)
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def conf(self, conf_path):
|
||||||
|
return config.load_config(conf_path)
|
||||||
|
|
||||||
|
def test_write(self, conf_path):
|
||||||
|
conf = RawConfigParser()
|
||||||
|
conf['Path'] = {
|
||||||
|
'day_begins': '4',
|
||||||
|
'dirs_path':'%u{%Y-%m}/{city}|{city}-{%Y}/{folders[:1]}/{folder}',
|
||||||
|
'name':'{%Y-%m-%b-%H-%M-%S}-{basename}.%l{ext}'
|
||||||
|
}
|
||||||
|
conf['Geolocation'] = {
|
||||||
|
'geocoder': 'Nominatium'
|
||||||
|
}
|
||||||
|
|
||||||
|
config.write(conf_path, conf)
|
||||||
|
assert conf_path.is_file()
|
||||||
|
|
||||||
|
def test_load_config(self, conf):
|
||||||
|
"""
|
||||||
|
Read files from config and return variables
|
||||||
|
"""
|
||||||
|
# test valid config file
|
||||||
|
assert conf['Path']['dirs_path'] == '%u{%Y-%m}/{city}|{city}-{%Y}/{folders[:1]}/{folder}'
|
||||||
|
assert conf['Path']['name'] == '{%Y-%m-%b-%H-%M-%S}-{basename}.%l{ext}'
|
||||||
|
assert conf['Path']['day_begins'] == '4'
|
||||||
|
assert conf['Geolocation']['geocoder'] == 'Nominatium'
|
||||||
|
|
||||||
|
def test_load_config_no_exist(self):
|
||||||
|
# test file not exist
|
||||||
|
conf = config.load_config('filename')
|
||||||
|
assert conf == {}
|
||||||
|
|
||||||
|
def test_load_config_invalid(self, conf_path):
|
||||||
|
# test invalid config
|
||||||
|
write_random_file(conf_path)
|
||||||
|
with pytest.raises(Exception) as e:
|
||||||
|
config.load_config(conf_path)
|
||||||
|
assert e.typename == 'MissingSectionHeaderError'
|
||||||
|
|
||||||
|
def test_get_path_definition(self, conf):
|
||||||
|
"""
|
||||||
|
Get path definition from config
|
||||||
|
"""
|
||||||
|
path = config.get_path_definition(conf)
|
||||||
|
assert path == '%u{%Y-%m}/{city}|{city}-{%Y}/{folders[:1]}/{folder}/{%Y-%m-%b-%H-%M-%S}-{basename}.%l{ext}'
|
||||||
|
|
|
@ -0,0 +1,19 @@
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
CONTENT = "content"
|
||||||
|
|
||||||
|
class TestDozo:
|
||||||
|
@pytest.mark.skip()
|
||||||
|
def test__sort(self):
|
||||||
|
assert 0
|
||||||
|
|
||||||
|
def test_needsfiles(tmpdir):
|
||||||
|
assert tmpdir
|
||||||
|
|
||||||
|
def test_create_file(tmp_path):
|
||||||
|
d = tmp_path / "sub"
|
||||||
|
d.mkdir()
|
||||||
|
p = d / "hello.txt"
|
||||||
|
p.write_text(CONTENT)
|
||||||
|
assert p.read_text() == CONTENT
|
||||||
|
assert len(list(tmp_path.iterdir())) == 1
|
|
@ -1,6 +1,5 @@
|
||||||
import json
|
import json
|
||||||
import pytest
|
import pytest
|
||||||
import subprocess
|
|
||||||
|
|
||||||
import dozo.exiftool
|
import dozo.exiftool
|
||||||
from dozo.exiftool import get_exiftool_path
|
from dozo.exiftool import get_exiftool_path
|
||||||
|
@ -178,15 +177,11 @@ def test_exiftool_terminate():
|
||||||
""" Test that exiftool process is terminated when exiftool.terminate() is called """
|
""" Test that exiftool process is terminated when exiftool.terminate() is called """
|
||||||
exif1 = dozo.exiftool.ExifTool(TEST_FILE_ONE_KEYWORD)
|
exif1 = dozo.exiftool.ExifTool(TEST_FILE_ONE_KEYWORD)
|
||||||
|
|
||||||
ps = subprocess.run(["ps"], capture_output=True)
|
assert dozo.exiftool.exiftool_is_running()
|
||||||
stdout = ps.stdout.decode("utf-8")
|
|
||||||
assert "exiftool" in stdout
|
|
||||||
|
|
||||||
dozo.exiftool.terminate_exiftool()
|
dozo.exiftool.terminate_exiftool()
|
||||||
|
|
||||||
ps = subprocess.run(["ps"], capture_output=True)
|
assert not dozo.exiftool.exiftool_is_running()
|
||||||
stdout = ps.stdout.decode("utf-8")
|
|
||||||
assert "exiftool" not in stdout
|
|
||||||
|
|
||||||
# verify we can create a new instance after termination
|
# verify we can create a new instance after termination
|
||||||
exif2 = dozo.exiftool.ExifTool(TEST_FILE_ONE_KEYWORD)
|
exif2 = dozo.exiftool.ExifTool(TEST_FILE_ONE_KEYWORD)
|
||||||
|
|
|
@ -0,0 +1,205 @@
|
||||||
|
# TODO to be removed later
|
||||||
|
from datetime import datetime
|
||||||
|
import os
|
||||||
|
import pytest
|
||||||
|
from pathlib import Path
|
||||||
|
import re
|
||||||
|
import shutil
|
||||||
|
from sys import platform
|
||||||
|
import tempfile
|
||||||
|
from time import sleep
|
||||||
|
|
||||||
|
from dozo import constants
|
||||||
|
from dozo.database import Db
|
||||||
|
from dozo.filesystem import FileSystem
|
||||||
|
from dozo.media.media import Media
|
||||||
|
from dozo.exiftool import ExifToolCaching, exiftool_is_running, terminate_exiftool
|
||||||
|
|
||||||
|
|
||||||
|
DOZO_PATH = Path(__file__).parent.parent
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skip()
|
||||||
|
class TestDb:
|
||||||
|
pass
|
||||||
|
|
||||||
|
class TestFilesystem:
|
||||||
|
def setup_class(cls):
|
||||||
|
cls.SRCPATH = tempfile.mkdtemp(prefix='dozo-src')
|
||||||
|
filenames = ['photo.png', 'plain.jpg', 'text.txt', 'withoutextension',
|
||||||
|
'no-exif.jpg']
|
||||||
|
cls.FILE_PATHS = set()
|
||||||
|
for filename in filenames:
|
||||||
|
source_path = Path(cls.SRCPATH, filename)
|
||||||
|
file_path = Path(DOZO_PATH, 'samples', filename)
|
||||||
|
shutil.copyfile(file_path, source_path)
|
||||||
|
cls.FILE_PATHS.add(source_path)
|
||||||
|
cls.path_format = constants.default_path + '/' + constants.default_name
|
||||||
|
|
||||||
|
def teardown_class(self):
|
||||||
|
terminate_exiftool()
|
||||||
|
assert not exiftool_is_running()
|
||||||
|
|
||||||
|
def test_get_part(self):
|
||||||
|
"""
|
||||||
|
Test all parts
|
||||||
|
"""
|
||||||
|
# Item to search for:
|
||||||
|
filesystem = FileSystem()
|
||||||
|
|
||||||
|
items = filesystem.get_items()
|
||||||
|
masks = [
|
||||||
|
'{album}',
|
||||||
|
'{basename}',
|
||||||
|
'{camera_make}',
|
||||||
|
'{camera_model}',
|
||||||
|
'{city}',
|
||||||
|
'{"custom"}',
|
||||||
|
'{country}',
|
||||||
|
'{ext}',
|
||||||
|
'{folder}',
|
||||||
|
'{folders[1:3]}',
|
||||||
|
'{location}',
|
||||||
|
'{name}',
|
||||||
|
'{original_name}',
|
||||||
|
'{state}',
|
||||||
|
'{title}',
|
||||||
|
'{%Y-%m-%d}',
|
||||||
|
'{%Y-%m-%d_%H-%M-%S}',
|
||||||
|
'{%Y-%m-%b}'
|
||||||
|
]
|
||||||
|
|
||||||
|
media = Media()
|
||||||
|
exif_tags = {
|
||||||
|
'album': media.album_keys,
|
||||||
|
'camera_make': media.camera_make_keys,
|
||||||
|
'camera_model': media.camera_model_keys,
|
||||||
|
# 'date_original': media.date_original,
|
||||||
|
# 'date_created': media.date_created,
|
||||||
|
# 'date_modified': media.date_modified,
|
||||||
|
'latitude': media.latitude_keys,
|
||||||
|
'longitude': media.longitude_keys,
|
||||||
|
'original_name': [media.original_name_key],
|
||||||
|
'title': media.title_keys
|
||||||
|
}
|
||||||
|
|
||||||
|
subdirs = Path('a', 'b', 'c', 'd')
|
||||||
|
|
||||||
|
for file_path in self.FILE_PATHS:
|
||||||
|
media = Media(str(file_path))
|
||||||
|
exif_data = ExifToolCaching(str(file_path)).asdict()
|
||||||
|
metadata = media.get_metadata()
|
||||||
|
|
||||||
|
for item, regex in items.items():
|
||||||
|
for mask in masks:
|
||||||
|
matched = re.search(regex, mask)
|
||||||
|
if matched:
|
||||||
|
part = filesystem.get_part(item, mask[1:-1], metadata,
|
||||||
|
{}, subdirs)
|
||||||
|
# check if part is correct
|
||||||
|
assert isinstance(part, str)
|
||||||
|
expected_part = ''
|
||||||
|
if item == 'basename':
|
||||||
|
expected_part = file_path.stem
|
||||||
|
elif item == 'date':
|
||||||
|
assert datetime.strptime(part, mask[1:-1])
|
||||||
|
expected_part = part
|
||||||
|
elif item == 'folder':
|
||||||
|
expected_part = subdirs.name
|
||||||
|
elif item == 'folders':
|
||||||
|
if platform == "win32":
|
||||||
|
assert '\\' in part
|
||||||
|
else:
|
||||||
|
assert '/' in part
|
||||||
|
expected_part = part
|
||||||
|
elif item == 'ext':
|
||||||
|
expected_part = file_path.suffix[1:]
|
||||||
|
if item == 'name':
|
||||||
|
expected_part = file_path.stem
|
||||||
|
for i, rx in filesystem.match_date_from_string(expected_part):
|
||||||
|
expected_part = re.sub(rx, '', expected_part)
|
||||||
|
elif item == 'custom':
|
||||||
|
expected_part = mask[2:-2]
|
||||||
|
elif item in ('city', 'country', 'location', 'state'):
|
||||||
|
expected_part = part
|
||||||
|
elif item in exif_tags.keys():
|
||||||
|
f = False
|
||||||
|
for key in exif_tags[item]:
|
||||||
|
if key in exif_data:
|
||||||
|
f = True
|
||||||
|
expected_part = exif_data[key]
|
||||||
|
break
|
||||||
|
if f == False:
|
||||||
|
expected_part = ''
|
||||||
|
|
||||||
|
assert part == expected_part
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_date_taken(self):
|
||||||
|
filesystem = FileSystem()
|
||||||
|
for file_path in self.FILE_PATHS:
|
||||||
|
exif_data = ExifToolCaching(str(file_path)).asdict()
|
||||||
|
media = Media(str(file_path))
|
||||||
|
metadata = media.get_metadata()
|
||||||
|
date_taken = filesystem.get_date_taken(metadata)
|
||||||
|
|
||||||
|
dates = {}
|
||||||
|
for key, date in ('original', media.date_original), ('created',
|
||||||
|
media.date_created), ('modified', media.date_modified):
|
||||||
|
dates[key] = media.get_date_attribute(date)
|
||||||
|
|
||||||
|
if media.original_name_key in exif_data:
|
||||||
|
date_filename = filesystem.get_date_from_string(
|
||||||
|
exif_data[media.original_name_key])
|
||||||
|
else:
|
||||||
|
date_filename = filesystem.get_date_from_string(file_path.name)
|
||||||
|
|
||||||
|
if dates['original']:
|
||||||
|
assert date_taken == dates['original']
|
||||||
|
elif date_filename:
|
||||||
|
assert date_taken == date_filename
|
||||||
|
elif dates['created']:
|
||||||
|
assert date_taken == dates['created']
|
||||||
|
elif dates['modified']:
|
||||||
|
assert date_taken == dates['modified']
|
||||||
|
|
||||||
|
def test_sort_files(self, tmp_path):
|
||||||
|
db = Db(tmp_path)
|
||||||
|
filesystem = FileSystem(path_format=self.path_format)
|
||||||
|
|
||||||
|
summary, has_errors = filesystem.sort_files([self.SRCPATH], tmp_path, db)
|
||||||
|
|
||||||
|
# Summary is created and there is no errors
|
||||||
|
assert summary, summary
|
||||||
|
assert not has_errors, has_errors
|
||||||
|
|
||||||
|
# TODO check if path follow path_format
|
||||||
|
|
||||||
|
# TODO make another class?
|
||||||
|
def test_sort_file(self, tmp_path):
|
||||||
|
|
||||||
|
for mode in 'copy', 'move':
|
||||||
|
filesystem = FileSystem(path_format=self.path_format, mode=mode)
|
||||||
|
# copy mode
|
||||||
|
src_path = Path(self.SRCPATH, 'photo.png')
|
||||||
|
dest_path = Path(tmp_path,'photo_copy.png')
|
||||||
|
src_checksum = filesystem.checksum(src_path)
|
||||||
|
result_copy = filesystem.sort_file(src_path, dest_path)
|
||||||
|
assert result_copy
|
||||||
|
# Ensure files remain the same
|
||||||
|
assert filesystem.checkcomp(dest_path, src_checksum)
|
||||||
|
|
||||||
|
if mode == 'copy':
|
||||||
|
assert src_path.exists()
|
||||||
|
else:
|
||||||
|
assert not src_path.exists()
|
||||||
|
|
||||||
|
# TODO check for conflicts
|
||||||
|
|
||||||
|
|
||||||
|
# TODO check date
|
||||||
|
|
||||||
|
# filesystem.sort_files
|
||||||
|
#- Sort similar images into a directory
|
||||||
|
# filesystem.sort_similar
|
||||||
|
|
|
@ -0,0 +1,44 @@
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pathlib import Path
|
||||||
|
import shutil
|
||||||
|
import tempfile
|
||||||
|
|
||||||
|
from dozo import constants
|
||||||
|
from dozo.media.media import Media
|
||||||
|
from dozo.media.audio import Audio
|
||||||
|
from dozo.media.photo import Photo
|
||||||
|
from dozo.media.video import Video
|
||||||
|
from dozo.exiftool import ExifToolCaching
|
||||||
|
|
||||||
|
DOZO_PATH = Path(__file__).parent.parent
|
||||||
|
|
||||||
|
class TestMetadata:
|
||||||
|
|
||||||
|
def setup_class(cls):
|
||||||
|
cls.SRCPATH = tempfile.mkdtemp(prefix='dozo-src')
|
||||||
|
filenames = ['invalid.jpg', 'photo.png', 'plain.jpg', 'text.txt', 'withoutextension']
|
||||||
|
cls.file_paths = set()
|
||||||
|
for filename in filenames:
|
||||||
|
source_path = Path(cls.SRCPATH, filename)
|
||||||
|
file_path = Path(DOZO_PATH, 'samples', filename)
|
||||||
|
shutil.copyfile(file_path, source_path)
|
||||||
|
cls.file_paths.add(source_path)
|
||||||
|
cls.path_format = constants.default_path + '/' + constants.default_name
|
||||||
|
|
||||||
|
def test_get_exiftool_attribute(self, tmp_path):
|
||||||
|
for file_path in self.file_paths:
|
||||||
|
exif_data = ExifToolCaching(str(file_path)).asdict()
|
||||||
|
ignore_tags = ('File:FileModifyDate', 'File:FileAccessDate')
|
||||||
|
exif_data_filtered = {}
|
||||||
|
for key in exif_data:
|
||||||
|
if key not in ignore_tags:
|
||||||
|
exif_data_filtered[key] = exif_data[key]
|
||||||
|
media = Media(str(file_path), ignore_tags)
|
||||||
|
exif = media.get_exiftool_attributes()
|
||||||
|
# Ensure returned value is a dictionary
|
||||||
|
assert isinstance(exif, dict)
|
||||||
|
for tag in ignore_tags:
|
||||||
|
assert tag not in exif
|
||||||
|
assert exif == exif_data_filtered
|
||||||
|
|