604 lines
23 KiB
Python
604 lines
23 KiB
Python
"""Test the functionality in the scan_directories module."""
|
|
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import random
|
|
import shutil
|
|
import tempfile
|
|
import time
|
|
from collections import namedtuple
|
|
|
|
import pytest
|
|
|
|
from openflexure_microscope_server.scan_directories import (
|
|
SCAN_DATA_FILENAME,
|
|
NotEnoughFreeSpaceError,
|
|
ScanDirectory,
|
|
ScanDirectoryManager,
|
|
ScanInfo,
|
|
get_files_in_zip,
|
|
)
|
|
|
|
from .test_scan_data import (
|
|
assert_active_and_historic_data_equivalent,
|
|
fake_active_scan_data,
|
|
)
|
|
from .utilities import assert_unique_of_length
|
|
|
|
# Use our own dir in the root temp dir not a dynamically generated one so we
|
|
# have some control of when it is deleted
|
|
BASE_SCAN_DIR = os.path.join(tempfile.gettempdir(), "scans")
|
|
|
|
|
|
def _clear_scan_dir() -> None:
|
|
"""Delete the scan dir."""
|
|
if os.path.exists(BASE_SCAN_DIR):
|
|
shutil.rmtree(BASE_SCAN_DIR)
|
|
|
|
|
|
def _add_fake_image(scan_dir: ScanDirectory) -> None:
|
|
"""Make a fake image on disk in the scan directory."""
|
|
unique = False
|
|
while not unique:
|
|
x_pos = random.randint(-10000, 10000)
|
|
y_pos = random.randint(-10000, 10000)
|
|
filename = f"image_{x_pos}_{y_pos}.jpg"
|
|
filepath = os.path.join(scan_dir.images_dir, filename)
|
|
unique = not os.path.exists(filepath)
|
|
with open(filepath, "w") as f_obj:
|
|
f_obj.write("fake")
|
|
|
|
|
|
def _add_fake_file(
|
|
scan_dir: ScanDirectory, filename: str, in_im_dir: bool = False
|
|
) -> None:
|
|
"""Make a fake file on disk in the scan directory.
|
|
|
|
:param in_im_dir: Boolean, if set True the fake file is created in the images
|
|
directory of the scan not the root directory.
|
|
"""
|
|
if in_im_dir:
|
|
filepath = os.path.join(scan_dir.images_dir, filename)
|
|
else:
|
|
filepath = os.path.join(scan_dir.dir_path, filename)
|
|
with open(filepath, "w") as f_obj:
|
|
f_obj.write("fake")
|
|
|
|
|
|
def _make_fake_dzi(scan_dir: ScanDirectory, n_layers: int = 8) -> None:
|
|
"""Create a fake DZI in a scan.
|
|
|
|
:param n_layers: The number of layers of tiles. I.e. tile directories numbered
|
|
0...(n_layers-1) will be created. Default 8
|
|
"""
|
|
# Add an a dzi image
|
|
dzi_fname = scan_dir.name + ".dzi"
|
|
dzi_path = os.path.join(scan_dir.images_dir, dzi_fname)
|
|
with open(dzi_path, "w") as f_obj:
|
|
f_obj.write("This should be xml")
|
|
|
|
# and the directory for the dzi tiles
|
|
dzi_tile_dir = scan_dir.name + "_files"
|
|
dzi_tile_dir_path = os.path.join(scan_dir.images_dir, dzi_tile_dir)
|
|
os.makedirs(dzi_tile_dir_path)
|
|
|
|
# this directory then contains a directories numbered from 0-n
|
|
for i in range(n_layers):
|
|
layer_dir_path = os.path.join(dzi_tile_dir_path, str(i))
|
|
os.makedirs(layer_dir_path)
|
|
|
|
# Number of images (in each axis) in this layer. Number of tiles doubles each
|
|
# layer, but the first few layers always have 1 image.
|
|
n_ims = math.ceil(2 ** (i - 4))
|
|
for x_index in range(n_ims):
|
|
for y_index in range(n_ims):
|
|
tile_name = f"{x_index}_{y_index}.jpg"
|
|
tile_path = os.path.join(layer_dir_path, tile_name)
|
|
with open(tile_path, "w") as f_obj:
|
|
f_obj.write("This would normally be jpeg data")
|
|
|
|
|
|
def test_basic_directory_operations():
|
|
"""Test some basic operations.
|
|
|
|
Test some basic operations, including:
|
|
- ScanDirectoryManager creates a scan directory
|
|
- For a fake (dummy) scan, a path can be created and retrieved
|
|
- For a fake (dummy) file, a path can be created and retrieved (for both a .zip and .img file)
|
|
- When a scan directory is created, the directory exists as does the image directory
|
|
- Scan directories can be deleted and scans can be cleared
|
|
"""
|
|
_clear_scan_dir()
|
|
assert not os.path.isdir(BASE_SCAN_DIR)
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
# Check it makes the scan directory
|
|
assert os.path.isdir(BASE_SCAN_DIR)
|
|
|
|
# Considering a fake scan
|
|
scan_name = "fake_scan_0001"
|
|
scan_path = os.path.join(BASE_SCAN_DIR, scan_name)
|
|
scan_im_dir = os.path.join(BASE_SCAN_DIR, scan_name, "images")
|
|
|
|
# It doesn't exist but we can check its paths
|
|
assert not scan_dir_manager.exists(scan_name)
|
|
assert not os.path.isdir(scan_path)
|
|
assert scan_dir_manager.path_for(scan_name) == scan_path
|
|
assert scan_dir_manager.img_dir_for(scan_name) == scan_im_dir
|
|
|
|
# Get the path of a fake file
|
|
fake_file = scan_dir_manager.get_file_path_from(scan_name, "foo.zip")
|
|
assert fake_file == os.path.join(scan_path, "foo.zip")
|
|
# But this is none if we check it exists
|
|
fake_file = scan_dir_manager.get_file_path_from(
|
|
scan_name, "foo.zip", check_exists=True
|
|
)
|
|
assert fake_file is None
|
|
|
|
# Get the path of another fake file
|
|
fake_file = scan_dir_manager.get_file_path_from_img_dir(scan_name, "bar.img")
|
|
assert fake_file == os.path.join(scan_im_dir, "bar.img")
|
|
# But this is none if we check it exists
|
|
fake_file = scan_dir_manager.get_file_path_from_img_dir(
|
|
scan_name, "bar.img", check_exists=True
|
|
)
|
|
assert fake_file is None
|
|
|
|
# Create the dir
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
# This returns a ScanDirectory object
|
|
assert isinstance(scan_dir, ScanDirectory)
|
|
# The directory now exists as does the image directory
|
|
assert scan_dir_manager.exists(scan_name)
|
|
assert os.path.isdir(scan_path)
|
|
assert os.path.isdir(scan_im_dir)
|
|
|
|
# Test cleanup. Delete the created scan and scan directory
|
|
scan_dir_manager.delete_scan(scan_name)
|
|
assert not scan_dir_manager.exists(scan_name)
|
|
assert not os.path.isdir(scan_path)
|
|
|
|
|
|
def test_bad_scan_names():
|
|
"""Check scan names with spaces, or worse BASH commands are not allowed."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake scan")
|
|
assert scan_dir.name == "fake_scan_0001"
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake scan;rm -rf /;")
|
|
assert scan_dir.name == "fake_scan_rm_-rf____0001"
|
|
|
|
|
|
def test_scan_sequence_and_listing(caplog):
|
|
"""Check created scans are added in order and listed correctly."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
|
|
# Create some scan data and mark it as successful to get an end date.
|
|
scan_data = fake_active_scan_data()
|
|
scan_data.set_final_data(result="Success")
|
|
# Make 4 scans
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
scan_dir.save_scan_data(scan_data)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
scan_dir.save_scan_data(scan_data)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
scan_dir.save_scan_data(scan_data)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
scan_dir.save_scan_data(scan_data)
|
|
|
|
# Check they exist and are numbered sequentially
|
|
all_scans = scan_dir_manager.all_scans
|
|
assert_unique_of_length(all_scans, 4)
|
|
assert "fake_scan_0001" in all_scans
|
|
assert "fake_scan_0002" in all_scans
|
|
assert "fake_scan_0003" in all_scans
|
|
assert "fake_scan_0004" in all_scans
|
|
|
|
# Start capturing warnings
|
|
with caplog.at_level(logging.WARNING):
|
|
# Check scan data can be read for all of them
|
|
# (more detailed scan_info tests below)
|
|
all_scan_info = scan_dir_manager.all_scans_info()
|
|
for scan_info in all_scan_info:
|
|
assert isinstance(scan_info, ScanInfo)
|
|
assert scan_info.name.startswith("fake_scan_000")
|
|
assert scan_info.number_of_images == 0
|
|
assert scan_info.duration is not None
|
|
# There are no warnings or errors
|
|
assert len(caplog.records) == 0
|
|
|
|
# Mess up the JSON of the final scan
|
|
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as f:
|
|
f.write("Mock JSON")
|
|
|
|
# Re-read the data with bad data for fake_scan_0004
|
|
all_scan_info = scan_dir_manager.all_scans_info()
|
|
for scan_info in all_scan_info:
|
|
assert isinstance(scan_info, ScanInfo)
|
|
assert scan_info.name.startswith("fake_scan_000")
|
|
assert scan_info.number_of_images == 0
|
|
# Bad data for scan 0004. So duration is None.
|
|
if scan_info.name == "fake_scan_0004":
|
|
assert scan_info.duration is None
|
|
else:
|
|
assert scan_info.duration is not None
|
|
# This should have warned about the bad data
|
|
assert len(caplog.records) == 1
|
|
assert (
|
|
caplog.records[0].message == "Could not load scan data for fake_scan_0004."
|
|
)
|
|
|
|
# Clear the warning logs
|
|
caplog.clear()
|
|
# Re-read the data again claiming fake_scan_0004 is ongoing
|
|
all_scan_info = scan_dir_manager.all_scans_info(ongoing="fake_scan_0004")
|
|
for scan_info in all_scan_info:
|
|
assert isinstance(scan_info, ScanInfo)
|
|
assert scan_info.name.startswith("fake_scan_000")
|
|
assert scan_info.number_of_images == 0
|
|
# scan 0004 is marked as ongoing, so duration is None.
|
|
if scan_info.name == "fake_scan_0004":
|
|
assert scan_info.duration is None
|
|
else:
|
|
assert scan_info.duration is not None
|
|
# This time there is no warning
|
|
assert len(caplog.records) == 0
|
|
|
|
|
|
def test_scan_name_non_sequential():
|
|
"""Check new scan has the correct name if the directories are not sequential."""
|
|
_clear_scan_dir()
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0001"))
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0002"))
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0003"))
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0005"))
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0007"))
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0011"))
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
assert scan_dir.name == "fake_scan_0012"
|
|
|
|
all_scans = scan_dir_manager.all_scans
|
|
assert_unique_of_length(all_scans, 7)
|
|
assert "fake_scan_0001" in all_scans
|
|
assert "fake_scan_0002" in all_scans
|
|
assert "fake_scan_0003" in all_scans
|
|
assert "fake_scan_0005" in all_scans
|
|
assert "fake_scan_0007" in all_scans
|
|
assert "fake_scan_0011" in all_scans
|
|
assert "fake_scan_0012" in all_scans
|
|
|
|
|
|
def test_all_scan_names_taken():
|
|
"""If the next sequential scan name needs more than 4 digits check error is thrown."""
|
|
_clear_scan_dir()
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_9999"))
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
with pytest.raises(FileExistsError):
|
|
scan_dir_manager.new_scan_dir("fake_scan")
|
|
|
|
|
|
def test_no_scan_names_given():
|
|
"""Check correct default scan name is used if empty string is given."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("")
|
|
assert scan_dir.name == "scan_0001"
|
|
|
|
|
|
def test_scan_info():
|
|
"""Test the scan info is correct even using fake scan data."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
|
|
for _i in range(17):
|
|
_add_fake_image(scan_dir)
|
|
info = scan_dir.scan_info()
|
|
now = time.time()
|
|
|
|
assert info.name == "fake_scan_0001"
|
|
# Created and modified in the last 5 seconds
|
|
assert now - 5 < info.created < now
|
|
assert now - 5 < info.modified < now
|
|
assert info.number_of_images == 17
|
|
assert not info.stitch_available
|
|
assert info.dzi is None
|
|
|
|
# Add a fake stitched images and check this is recognised as a stitch not
|
|
# a scan image
|
|
_add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True)
|
|
info = scan_dir.scan_info()
|
|
assert info.number_of_images == 17
|
|
assert info.stitch_available
|
|
|
|
|
|
def test_get_final_stitch():
|
|
"""Check that the final stitch can be retrieved."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
|
|
# Create some scan images files
|
|
for _i in range(17):
|
|
_add_fake_image(scan_dir)
|
|
|
|
# No scans, so None should be returned, from both manager and scan dir
|
|
assert scan_dir_manager.get_final_stitch_path(scan_dir.name) is None
|
|
assert scan_dir.get_final_stitch_name() is None
|
|
|
|
fake_scan_name = "fake_scan_0001_stitched.jpg"
|
|
fake_scan_path = scan_dir_manager.get_file_path_from_img_dir(
|
|
scan_name=scan_dir.name,
|
|
filename="fake_scan_0001_stitched.jpg",
|
|
check_exists=False,
|
|
)
|
|
# Add a fake scan
|
|
_add_fake_file(scan_dir, fake_scan_name, in_im_dir=True)
|
|
# Manager returns full path
|
|
assert scan_dir_manager.get_final_stitch_path(scan_dir.name) == fake_scan_path
|
|
# ScanDirectory object returns just the filename
|
|
assert scan_dir.get_final_stitch_name() == fake_scan_name
|
|
|
|
|
|
def test_get_scan_data_path():
|
|
"""Check that a scan data path behaves as expected."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
scan_name = scan_dir.name
|
|
expected_path = os.path.join(scan_dir.images_dir, SCAN_DATA_FILENAME)
|
|
# Asking the scan manager for the path will return None as there is no file.
|
|
assert scan_dir_manager.get_scan_data_path(scan_name) is None
|
|
# Asking the scan directly will return the location the file should be located
|
|
# this is so it can be created.
|
|
assert scan_dir.scan_data_path == expected_path
|
|
|
|
# Remove the images directory.
|
|
shutil.rmtree(scan_dir.images_dir)
|
|
# When the images directory is deleted, internally `scan_dir.images_dir`
|
|
# will return `None`. Check that `get_scan_data_path` copes with the `None`
|
|
# return from `scan_dir.images_dir`.
|
|
assert scan_dir_manager.get_scan_data_path(scan_name) is None
|
|
|
|
|
|
def test_get_scan_data():
|
|
"""Check that the scan data is returned, or None if doesn't exist."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
scan_name = scan_dir.name
|
|
# Doesn't yet exist
|
|
assert scan_dir_manager.get_scan_data(scan_name) is None
|
|
|
|
fake_active_data = fake_active_scan_data()
|
|
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file:
|
|
json.dump(fake_active_data.model_dump(), json_file)
|
|
|
|
# Should now be able to load this fake data from disk
|
|
fake_historic_data = scan_dir_manager.get_scan_data(scan_name)
|
|
assert_active_and_historic_data_equivalent(fake_active_data, fake_historic_data)
|
|
|
|
# Check None is returned if the data cannot be read.
|
|
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file:
|
|
json_file.write("this is not json")
|
|
assert scan_dir_manager.get_scan_data(scan_name) is None
|
|
|
|
# Check None is returned if the data cannot or is json but cannot be serialised to
|
|
# the data model
|
|
with open(scan_dir.scan_data_path, "w", encoding="utf-8") as json_file:
|
|
json_file.write(json.dumps({"foo": "bar"}))
|
|
assert scan_dir_manager.get_scan_data(scan_name) is None
|
|
|
|
|
|
def test_empty_scan_info():
|
|
"""Test the scan info is correct even if the scan is empty."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
|
|
info = scan_dir.scan_info()
|
|
now = time.time()
|
|
|
|
assert info.name == "fake_scan_0001"
|
|
# Created and modified in the last 5 seconds
|
|
assert now - 5 < info.created < now
|
|
assert now - 5 < info.modified < now
|
|
assert info.number_of_images == 0
|
|
assert not info.stitch_available
|
|
assert info.dzi is None
|
|
|
|
|
|
def test_zipping_scan_data():
|
|
"""Test zipping the scan images with fake image data."""
|
|
# Run twice, once calling the ScanDirectory directly,
|
|
# Once calling the ScanDirectoryManager
|
|
for caller in ["scan_dir", "manager"]:
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
|
|
# Create 21 fake scan images, a fake stitch, and a fake zip
|
|
for _i in range(21):
|
|
_add_fake_image(scan_dir)
|
|
_add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True)
|
|
_add_fake_file(scan_dir, "zipfile.zip")
|
|
|
|
# This fake dzi should have loads of images. The DZI should not be zipped!
|
|
_make_fake_dzi(scan_dir)
|
|
|
|
# zip the directory without setting as the final version. It should only
|
|
# zip the 21 scan images
|
|
if caller == "scan_dir":
|
|
zip_fname = scan_dir.zip_files()
|
|
else:
|
|
zip_fname = scan_dir_manager.zip_scan("fake_scan_0001")
|
|
zip_files = get_files_in_zip(zip_fname)
|
|
assert_unique_of_length(zip_files, 21)
|
|
|
|
# Zip again with final version on and there should be 22 images
|
|
if caller == "scan_dir":
|
|
scan_dir.zip_files(final_version=True)
|
|
else:
|
|
scan_dir_manager.zip_scan("fake_scan_0001", final_version=True)
|
|
zip_files = get_files_in_zip(zip_fname)
|
|
assert_unique_of_length(get_files_in_zip(zip_fname), 22)
|
|
# Check the zips are not in the zip
|
|
for file in zip_files:
|
|
assert not file.endswith(".zip")
|
|
assert not file.endswith(".dzi")
|
|
|
|
|
|
def test_saving_scan_data_error():
|
|
"""Test that saving scan data if there is no images directory raises FileNotFoundError."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
|
|
# Remove the images directory.
|
|
shutil.rmtree(scan_dir.images_dir)
|
|
# Should raise FileNotFoundError.
|
|
with pytest.raises(FileNotFoundError):
|
|
scan_dir.save_scan_data(fake_active_scan_data())
|
|
|
|
|
|
def test_all_files():
|
|
"""Test all_files returns the path, and respects skipped directories."""
|
|
_clear_scan_dir()
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir = scan_dir_manager.new_scan_dir("fake_scan")
|
|
# This fake dzi should have loads of images. The DZI should not be zipped!
|
|
_make_fake_dzi(scan_dir)
|
|
all_files = scan_dir.all_files()
|
|
|
|
# As standard for 8 layers there are 89 jpegs and 1 dzi file.
|
|
assert_unique_of_length(all_files, 90)
|
|
dzi_file = None
|
|
# Check all files exist
|
|
for file in all_files:
|
|
assert os.path.exists(os.path.join(scan_dir.dir_path, file))
|
|
if file.endswith(".dzi"):
|
|
# There is only 1 dzi file, so this should be None
|
|
assert dzi_file is None
|
|
dzi_file = file
|
|
# Once loop is complete there should be a dzi file
|
|
assert dzi_file is not None
|
|
# Get the dzi tile directory name
|
|
dzi_dir = os.path.basename(dzi_file[:-4] + "_files")
|
|
|
|
# Get all files skipping the dzi tile directory
|
|
all_files = scan_dir.all_files(skip_dirs=dzi_dir)
|
|
# There is now only one file
|
|
assert len(all_files) == 1
|
|
# It is the DZI file
|
|
assert all_files[0] == dzi_file
|
|
|
|
|
|
def test_creating_scan_dir_for_missing_scan():
|
|
"""Check creating ScanDirectory object for a dir that doesn't exist fails."""
|
|
_clear_scan_dir()
|
|
with pytest.raises(FileNotFoundError):
|
|
ScanDirectory("not_real_0001", BASE_SCAN_DIR)
|
|
|
|
|
|
def test_none_returned_for_missing_images_dir():
|
|
"""None should be returned for image dir path if images dir does not exist.
|
|
|
|
By default images directories are created at the same time the scan directory
|
|
is created. However, edge cases such as problems in deletions, microscopes
|
|
with older scans on, etc can cause and empty scan directory, so it is handled
|
|
explicitly.
|
|
"""
|
|
_clear_scan_dir()
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0001"))
|
|
scan_dir = ScanDirectory("fake_scan_0001", BASE_SCAN_DIR)
|
|
assert scan_dir.images_dir is None
|
|
# Also check that get scan files returns and empty list
|
|
assert scan_dir.get_scan_files() == []
|
|
|
|
|
|
def test_extracting_files():
|
|
"""Test the private _find_files method of ScanDirectories.
|
|
|
|
Add files to directory and check expected returns.
|
|
"""
|
|
_clear_scan_dir()
|
|
os.makedirs(os.path.join(BASE_SCAN_DIR, "fake_scan_0001", "images"))
|
|
scan_dir = ScanDirectory("fake_scan_0001", BASE_SCAN_DIR)
|
|
|
|
# Starting all lists should be empty
|
|
scan_files = scan_dir.get_scan_files()
|
|
assert scan_dir._extract_scan_images(scan_files) == []
|
|
assert scan_dir._extract_final_stitches(scan_files) == []
|
|
assert scan_dir._extract_dzi_files(scan_files) == []
|
|
|
|
# Add a number of images
|
|
for _i in range(2321):
|
|
_add_fake_image(scan_dir)
|
|
|
|
scan_files = scan_dir.get_scan_files()
|
|
assert_unique_of_length(scan_dir._extract_scan_images(scan_files), 2321)
|
|
assert_unique_of_length(scan_dir._extract_final_stitches(scan_files), 0)
|
|
assert_unique_of_length(scan_dir._extract_dzi_files(scan_files), 0)
|
|
|
|
# Add and a stitched image
|
|
_add_fake_file(scan_dir, "fake_scan_0001_stitched.jpg", in_im_dir=True)
|
|
|
|
scan_files = scan_dir.get_scan_files()
|
|
assert_unique_of_length(scan_dir._extract_scan_images(scan_files), 2321)
|
|
assert_unique_of_length(scan_dir._extract_final_stitches(scan_files), 1)
|
|
assert_unique_of_length(scan_dir._extract_dzi_files(scan_files), 0)
|
|
|
|
_make_fake_dzi(scan_dir)
|
|
|
|
# check totals are still correct after adding a dzi with lots of tiles.
|
|
scan_files = scan_dir.get_scan_files()
|
|
scan_images = scan_dir._extract_scan_images(scan_files)
|
|
stitches = scan_dir._extract_final_stitches(scan_files)
|
|
dzi_files = scan_dir._extract_dzi_files(scan_files)
|
|
assert_unique_of_length(scan_images, 2321)
|
|
assert_unique_of_length(stitches, 1)
|
|
assert_unique_of_length(dzi_files, 1)
|
|
|
|
# And check the names are as expected
|
|
assert stitches[0] == "fake_scan_0001_stitched.jpg"
|
|
assert dzi_files[0] == "fake_scan_0001.dzi"
|
|
|
|
|
|
DiskUsage = namedtuple("DiskUsage", ["total", "used", "free"])
|
|
|
|
|
|
@pytest.mark.parametrize("free_space", [500_000_001, 700_000_000, 5_000_000_000])
|
|
def test_disk_not_full(mocker, free_space):
|
|
"""Check no error thrown if disk has over 500MB of space."""
|
|
total_space = 16_000_000_000
|
|
# Mock the disk_usage
|
|
mocker.patch(
|
|
"shutil.disk_usage",
|
|
return_value=DiskUsage(
|
|
total=total_space, used=total_space - free_space, free=free_space
|
|
),
|
|
)
|
|
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
scan_dir_manager.check_free_disk_space()
|
|
|
|
|
|
@pytest.mark.parametrize("free_space", [100_000_000, 499_999_999])
|
|
def test_disk_full(mocker, free_space):
|
|
"""Check error thrown if disk has under 500MB of space."""
|
|
total_space = 16_000_000_000
|
|
# Mock the disk_usage
|
|
mocker.patch(
|
|
"shutil.disk_usage",
|
|
return_value=DiskUsage(
|
|
total=total_space, used=total_space - free_space, free=free_space
|
|
),
|
|
)
|
|
|
|
scan_dir_manager = ScanDirectoryManager(BASE_SCAN_DIR)
|
|
with pytest.raises(NotEnoughFreeSpaceError):
|
|
scan_dir_manager.check_free_disk_space()
|