Migrate filemover_analyzer_test to pytest
This commit is contained in:
parent
27090c8f2f
commit
6dee626cbd
2 changed files with 97 additions and 94 deletions
|
@ -9,8 +9,14 @@ from .analyzer import Analyzer
|
||||||
|
|
||||||
|
|
||||||
class FileMoverAnalyzer(Analyzer):
|
class FileMoverAnalyzer(Analyzer):
|
||||||
"""This analyzer copies a file over from a temporary directory (stor/organize)
|
"""
|
||||||
|
This analyzer copies a file over from a temporary directory (stor/organize)
|
||||||
into the Airtime library (stor/imported).
|
into the Airtime library (stor/imported).
|
||||||
|
|
||||||
|
If you import three copies of the same file, the behaviour is:
|
||||||
|
- The filename is of the first file preserved.
|
||||||
|
- The filename of the second file has the timestamp attached to it.
|
||||||
|
- The filename of the third file has a UUID placed after the timestamp, but ONLY IF it's imported within 1 second of the second file (ie. if the timestamp is the same).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
|
@ -1,132 +1,129 @@
|
||||||
import multiprocessing
|
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
|
import tempfile
|
||||||
import time
|
import time
|
||||||
from pprint import pprint
|
|
||||||
|
|
||||||
import mock
|
import mock
|
||||||
|
import pytest
|
||||||
from airtime_analyzer.filemover_analyzer import FileMoverAnalyzer
|
from airtime_analyzer.filemover_analyzer import FileMoverAnalyzer
|
||||||
from nose.tools import *
|
|
||||||
|
|
||||||
DEFAULT_AUDIO_FILE = u"tests/test_data/44100Hz-16bit-mono.mp3"
|
AUDIO_FILE = "tests/test_data/44100Hz-16bit-mono.mp3"
|
||||||
DEFAULT_IMPORT_DEST = u"Test Artist/Test Album/44100Hz-16bit-mono.mp3"
|
AUDIO_FILENAME = os.path.basename(AUDIO_FILE)
|
||||||
|
|
||||||
|
|
||||||
def setup():
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def teardown():
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
@raises(Exception)
|
|
||||||
def test_dont_use_analyze():
|
def test_dont_use_analyze():
|
||||||
FileMoverAnalyzer.analyze(u"foo", dict())
|
with pytest.raises(Exception):
|
||||||
|
FileMoverAnalyzer.analyze("foo", dict())
|
||||||
|
|
||||||
|
|
||||||
@raises(TypeError)
|
@pytest.mark.parametrize(
|
||||||
def test_move_wrong_string_param1():
|
"params,exception",
|
||||||
FileMoverAnalyzer.move(42, "", "", dict())
|
[
|
||||||
|
((42, "", "", dict()), TypeError),
|
||||||
|
(("", 23, "", dict()), TypeError),
|
||||||
|
(("", "", 5, dict()), TypeError),
|
||||||
|
(("", "", "", 12345), TypeError),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_move_wrong_params(params, exception):
|
||||||
|
with pytest.raises(exception):
|
||||||
|
FileMoverAnalyzer.move(*params)
|
||||||
|
|
||||||
|
|
||||||
@raises(TypeError)
|
@pytest.fixture()
|
||||||
def test_move_wrong_string_param2():
|
def dest_dir():
|
||||||
FileMoverAnalyzer.move(u"", 23, u"", dict())
|
with tempfile.TemporaryDirectory(prefix="dest") as tmpdir:
|
||||||
|
yield tmpdir
|
||||||
|
|
||||||
|
|
||||||
@raises(TypeError)
|
@pytest.fixture()
|
||||||
def test_move_wrong_string_param3():
|
def src_dir():
|
||||||
FileMoverAnalyzer.move("", "", 5, dict())
|
with tempfile.TemporaryDirectory(prefix="src") as tmpdir:
|
||||||
|
shutil.copy(AUDIO_FILE, tmpdir)
|
||||||
|
yield tmpdir
|
||||||
|
|
||||||
|
|
||||||
@raises(TypeError)
|
def test_basic(src_dir, dest_dir):
|
||||||
def test_move_wrong_dict_param():
|
FileMoverAnalyzer.move(
|
||||||
FileMoverAnalyzer.move("", "", "", 12345)
|
os.path.join(src_dir, AUDIO_FILENAME),
|
||||||
|
dest_dir,
|
||||||
|
AUDIO_FILENAME,
|
||||||
|
dict(),
|
||||||
|
)
|
||||||
|
assert os.path.exists(os.path.join(dest_dir, AUDIO_FILENAME))
|
||||||
|
|
||||||
|
|
||||||
@raises(FileNotFoundError)
|
def test_basic_samefile(src_dir):
|
||||||
def test_move_wrong_string_param3():
|
FileMoverAnalyzer.move(
|
||||||
FileMoverAnalyzer.move("", "", "", dict())
|
os.path.join(src_dir, AUDIO_FILENAME),
|
||||||
|
src_dir,
|
||||||
|
AUDIO_FILENAME,
|
||||||
|
dict(),
|
||||||
|
)
|
||||||
|
assert os.path.exists(os.path.join(src_dir, AUDIO_FILENAME))
|
||||||
|
|
||||||
|
|
||||||
def test_basic():
|
def import_and_restore(src_dir, dest_dir) -> dict:
|
||||||
filename = os.path.basename(DEFAULT_AUDIO_FILE)
|
# Import the file
|
||||||
FileMoverAnalyzer.move(DEFAULT_AUDIO_FILE, u".", filename, dict())
|
metadata = FileMoverAnalyzer.move(
|
||||||
# Move the file back
|
os.path.join(src_dir, AUDIO_FILENAME),
|
||||||
shutil.move("./" + filename, DEFAULT_AUDIO_FILE)
|
dest_dir,
|
||||||
assert os.path.exists(DEFAULT_AUDIO_FILE)
|
AUDIO_FILENAME,
|
||||||
|
dict(),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_basic_samefile():
|
|
||||||
filename = os.path.basename(DEFAULT_AUDIO_FILE)
|
|
||||||
FileMoverAnalyzer.move(DEFAULT_AUDIO_FILE, u"tests/test_data", filename, dict())
|
|
||||||
assert os.path.exists(DEFAULT_AUDIO_FILE)
|
|
||||||
|
|
||||||
|
|
||||||
def test_duplicate_file():
|
|
||||||
filename = os.path.basename(DEFAULT_AUDIO_FILE)
|
|
||||||
# Import the file once
|
|
||||||
FileMoverAnalyzer.move(DEFAULT_AUDIO_FILE, u".", filename, dict())
|
|
||||||
# Copy it back to the original location
|
# Copy it back to the original location
|
||||||
shutil.copy("./" + filename, DEFAULT_AUDIO_FILE)
|
shutil.copy(
|
||||||
|
os.path.join(dest_dir, AUDIO_FILENAME),
|
||||||
|
os.path.join(src_dir, AUDIO_FILENAME),
|
||||||
|
)
|
||||||
|
|
||||||
|
return metadata
|
||||||
|
|
||||||
|
|
||||||
|
def test_duplicate_file(src_dir, dest_dir):
|
||||||
|
# Import the file once
|
||||||
|
import_and_restore(src_dir, dest_dir)
|
||||||
|
|
||||||
# Import it again. It shouldn't overwrite the old file and instead create a new
|
# Import it again. It shouldn't overwrite the old file and instead create a new
|
||||||
metadata = dict()
|
metadata = import_and_restore(src_dir, dest_dir)
|
||||||
metadata = FileMoverAnalyzer.move(DEFAULT_AUDIO_FILE, u".", filename, metadata)
|
|
||||||
# Cleanup: move the file (eg. 44100Hz-16bit-mono.mp3) back
|
assert metadata["full_path"] != os.path.join(dest_dir, AUDIO_FILENAME)
|
||||||
shutil.move("./" + filename, DEFAULT_AUDIO_FILE)
|
assert os.path.exists(metadata["full_path"])
|
||||||
# Remove the renamed duplicate, eg. 44100Hz-16bit-mono_03-26-2014-11-58.mp3
|
assert os.path.exists(os.path.join(dest_dir, AUDIO_FILENAME))
|
||||||
os.remove(metadata["full_path"])
|
|
||||||
assert os.path.exists(DEFAULT_AUDIO_FILE)
|
|
||||||
|
|
||||||
|
|
||||||
""" If you import three copies of the same file, the behaviour is:
|
def test_double_duplicate_files(src_dir, dest_dir):
|
||||||
- The filename is of the first file preserved.
|
|
||||||
- The filename of the second file has the timestamp attached to it.
|
|
||||||
- The filename of the third file has a UUID placed after the timestamp, but ONLY IF
|
|
||||||
it's imported within 1 second of the second file (ie. if the timestamp is the same).
|
|
||||||
"""
|
|
||||||
|
|
||||||
|
|
||||||
def test_double_duplicate_files():
|
|
||||||
# Here we use mock to patch out the time.localtime() function so that it
|
# Here we use mock to patch out the time.localtime() function so that it
|
||||||
# always returns the same value. This allows us to consistently simulate this test cases
|
# always returns the same value. This allows us to consistently simulate this test cases
|
||||||
# where the last two of the three files are imported at the same time as the timestamp.
|
# where the last two of the three files are imported at the same time as the timestamp.
|
||||||
with mock.patch("airtime_analyzer.filemover_analyzer.time") as mock_time:
|
with mock.patch("airtime_analyzer.filemover_analyzer.time") as mock_time:
|
||||||
mock_time.localtime.return_value = time.localtime() # date(2010, 10, 8)
|
mock_time.localtime.return_value = time.localtime() # date(2010, 10, 8)
|
||||||
mock_time.side_effect = lambda *args, **kw: time(*args, **kw)
|
mock_time.side_effect = time.time
|
||||||
|
|
||||||
filename = os.path.basename(DEFAULT_AUDIO_FILE)
|
|
||||||
# Import the file once
|
# Import the file once
|
||||||
FileMoverAnalyzer.move(DEFAULT_AUDIO_FILE, u".", filename, dict())
|
import_and_restore(src_dir, dest_dir)
|
||||||
# Copy it back to the original location
|
|
||||||
shutil.copy("./" + filename, DEFAULT_AUDIO_FILE)
|
|
||||||
# Import it again. It shouldn't overwrite the old file and instead create a new
|
# Import it again. It shouldn't overwrite the old file and instead create a new
|
||||||
first_dup_metadata = dict()
|
metadata1 = import_and_restore(src_dir, dest_dir)
|
||||||
first_dup_metadata = FileMoverAnalyzer.move(
|
|
||||||
DEFAULT_AUDIO_FILE, u".", filename, first_dup_metadata
|
|
||||||
)
|
|
||||||
# Copy it back again!
|
|
||||||
shutil.copy("./" + filename, DEFAULT_AUDIO_FILE)
|
|
||||||
# Reimport for the third time, which should have the same timestamp as the second one
|
# Reimport for the third time, which should have the same timestamp as the second one
|
||||||
# thanks to us mocking out time.localtime()
|
# thanks to us mocking out time.localtime()
|
||||||
second_dup_metadata = dict()
|
metadata2 = import_and_restore(src_dir, dest_dir)
|
||||||
second_dup_metadata = FileMoverAnalyzer.move(
|
|
||||||
DEFAULT_AUDIO_FILE, u".", filename, second_dup_metadata
|
assert os.path.exists(metadata1["full_path"])
|
||||||
)
|
# Check if filename is <original>_<date>.<ext>
|
||||||
# Cleanup: move the file (eg. 44100Hz-16bit-mono.mp3) back
|
assert len(metadata1["full_path"].split("_")) == 2
|
||||||
shutil.move("./" + filename, DEFAULT_AUDIO_FILE)
|
assert os.path.exists(metadata2["full_path"])
|
||||||
# Remove the renamed duplicate, eg. 44100Hz-16bit-mono_03-26-2014-11-58.mp3
|
# Check if filename is <original>_<date>_<uuid>.<ext>
|
||||||
os.remove(first_dup_metadata["full_path"])
|
assert len(metadata2["full_path"].split("_")) == 3
|
||||||
os.remove(second_dup_metadata["full_path"])
|
|
||||||
assert os.path.exists(DEFAULT_AUDIO_FILE)
|
|
||||||
|
|
||||||
|
|
||||||
@raises(OSError)
|
def test_bad_permissions_destination_dir(src_dir):
|
||||||
def test_bad_permissions_destination_dir():
|
with pytest.raises(OSError):
|
||||||
filename = os.path.basename(DEFAULT_AUDIO_FILE)
|
# /sys is using sysfs on Linux, which is unwritable
|
||||||
dest_dir = u"/sys/foobar" # /sys is using sysfs on Linux, which is unwritable
|
FileMoverAnalyzer.move(
|
||||||
FileMoverAnalyzer.move(DEFAULT_AUDIO_FILE, dest_dir, filename, dict())
|
os.path.join(src_dir, AUDIO_FILENAME),
|
||||||
# Move the file back
|
"/sys/foobar",
|
||||||
shutil.move(os.path.join(dest_dir, filename), DEFAULT_AUDIO_FILE)
|
AUDIO_FILENAME,
|
||||||
assert os.path.exists(DEFAULT_AUDIO_FILE)
|
dict(),
|
||||||
|
)
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue