diff --git a/packs/tests/conftest.py b/packs/tests/conftest.py new file mode 100644 index 0000000..6fc7dae --- /dev/null +++ b/packs/tests/conftest.py @@ -0,0 +1,14 @@ +import os +import pytest + +@pytest.fixture(scope="session") +def MULE_dir(): + return str(os.environ['MULE_DIR']) + +@pytest.fixture(scope="session") +def data_dir(MULE_dir): + return MULE_dir + '/packs/tests/data/' + +@pytest.fixture(scope="session") +def wd2_3ch_bin(data_dir): + return data_dir + 'three_channels_WD2.bin' \ No newline at end of file diff --git a/packs/tests/processing_test.py b/packs/tests/processing_test.py index 4366160..57af0db 100644 --- a/packs/tests/processing_test.py +++ b/packs/tests/processing_test.py @@ -1,4 +1,3 @@ -import os import sys import numpy as np @@ -10,6 +9,7 @@ from pytest import mark from pytest import raises from pytest import warns +from pytest import fixture from packs.proc.processing_utils import read_defaults_WD2 from packs.proc.processing_utils import process_header @@ -36,17 +36,14 @@ def test_rwf_type_has_correct_shape(samples): assert x['rwf'].shape[0] == samples -def test_header_components_read_as_expected(): - - MULE_dir = str(os.environ['MULE_DIR']) - file = MULE_dir + '/packs/tests/data/three_channels_WD2.bin' +def test_header_components_read_as_expected(wd2_3ch_bin): evt_num = 0 tstamp = 1998268 smpls = 1000 smpl_prd = 8 - with open(file, 'rb') as f: + with open(wd2_3ch_bin, 'rb') as f: event_number, timestamp, samples, sampling_period = read_defaults_WD2(f, sys.byteorder) assert event_number == evt_num @@ -55,24 +52,21 @@ def test_header_components_read_as_expected(): assert sampling_period == smpl_prd -def test_header_processed_correctly(): - - MULE_dir = str(os.environ['MULE_DIR']) - file = MULE_dir + '/packs/tests/data/three_channels_WD2.bin' +def test_header_processed_correctly(wd2_3ch_bin): smpls = 1000 smpl_prd = 8 channels = 3 wdtype = generate_wfdtype(channels, smpls) # 3 channels in this case - result = process_header(file) + result = process_header(wd2_3ch_bin) assert result[0] == wdtype assert result[1] == smpls assert result[2] == smpl_prd assert result[3] == channels -def test_header_works_when_data_malformed(): +def test_header_works_when_data_malformed(data_dir): # this test would normally cause a memory error as the data # provided is singular channel, and `process_header()` tests # for single channel behaviour by analysing it as multi-channel @@ -82,32 +76,25 @@ def test_header_works_when_data_malformed(): # This has been fixed quickly in process_header, but should be # optimised in a different fashion. - MULE_dir = str(os.environ['MULE_DIR']) - file = MULE_dir + '/packs/tests/data/malformed_data.bin' + file = data_dir + 'malformed_data.bin' with warns(UserWarning): process_header(file) @mark.parametrize("function, error", [(process_header, NameError), (read_defaults_WD2, ValueError)]) -def test_endian_error_when_reading(function, error): - - MULE_dir = str(os.environ['MULE_DIR']) - file = MULE_dir + '/packs/tests/data/three_channels_WD2.bin' - +def test_endian_error_when_reading(function, error, wd2_3ch_bin): byte_order = 'Big' # this will raise a ValueError with raises(error): - with open(file, 'rb') as f: + with open(wd2_3ch_bin, 'rb') as f: holder = function(f, byte_order) -def test_invalid_file_for_reading(): - - MULE_dir = str(os.environ['MULE_DIR']) - file = MULE_dir + '/packs/tests/data/false_data.npy' +def test_invalid_file_for_reading(data_dir): + file = data_dir + 'false_data.npy' x = read_binary(file, types.generate_wfdtype(1, 1000)) @@ -116,14 +103,10 @@ def test_invalid_file_for_reading(): assert len(x) == 0 -def test_formatting_works(): - - MULE_dir = str(os.environ['MULE_DIR']) - - file_path = MULE_dir + '/packs/tests/data/three_channels_WD2.bin' +def test_formatting_works(data_dir, wd2_3ch_bin): # collect relevant data from output - check_file = MULE_dir + '/packs/tests/data/three_channels_WD2.h5' + check_file = data_dir + 'three_channels_WD2.h5' check_rwf = load_rwf_info(check_file, 1000) check_evt_info = load_evt_info(check_file) @@ -133,7 +116,7 @@ def test_formatting_works(): wdtype = types.generate_wfdtype(channels, samples) - with open(file_path, 'rb') as file: + with open(wd2_3ch_bin, 'rb') as file: # read in data data = read_binary(file, wdtype) @@ -147,21 +130,19 @@ def test_formatting_works(): assert evt_info.equals(check_evt_info) -def test_ensure_new_path_created(): +def test_ensure_new_path_created(data_dir): - MULE_dir = str(os.environ['MULE_DIR']) - data_path = MULE_dir + '/packs/tests/data/three_channels_WD2.h5' - new_data_path = MULE_dir + '/packs/tests/data/three_channels_WD21.h5' + data_path = data_dir + 'three_channels_WD2.h5' + new_data_path = data_dir + 'three_channels_WD21.h5' found_path = check_save_path(data_path, overwrite = False) assert found_path == new_data_path -def test_runtime_error_when_too_many_save_files(): +def test_runtime_error_when_too_many_save_files(data_dir): - MULE_dir = str(os.environ['MULE_DIR']) - relevant_dir = MULE_dir + '/packs/tests/data/repetitive_data/' + relevant_dir = data_dir + 'repetitive_data/' # generate 101 empty files with open(relevant_dir + f'test_.txt', 'w'): pass @@ -173,16 +154,13 @@ def test_runtime_error_when_too_many_save_files(): @mark.parametrize("config, inpt, output, comparison", [("process_WD2_1channel.conf", "one_channel_WD2.bin", "one_channel_tmp.h5", "one_channel_WD2.h5"), ("process_WD2_3channel.conf", "three_channels_WD2.bin", "three_channels_tmp.h5", "three_channels_WD2.h5")]) -def test_decode_produces_expected_output(config, inpt, output, comparison): - - MULE_dir = str(os.environ['MULE_DIR']) - data_dir = "/packs/tests/data/" +def test_decode_produces_expected_output(config, inpt, output, comparison, MULE_dir, data_dir): # ensure path is correct - file_path = MULE_dir + data_dir + inpt - save_path = MULE_dir + data_dir + output - comparison_path = MULE_dir + data_dir + comparison - config_path = MULE_dir + data_dir + "configs/" + config + file_path = data_dir + inpt + save_path = data_dir + output + comparison_path = data_dir + comparison + config_path = data_dir + "configs/" + config # collect samples from header _, samples, _, _ = process_header(file_path) diff --git a/packs/tests/setup_test.py b/packs/tests/setup_test.py index 26ff4f2..b352c1c 100644 --- a/packs/tests/setup_test.py +++ b/packs/tests/setup_test.py @@ -1,22 +1,21 @@ -import os import subprocess import configparser from pytest import mark from pytest import raises +from pytest import fixture from packs.core.io import read_config_file @mark.parametrize("pack", ["acq", "proc", "tests"]) -def test_executable_runs_successfully(pack): +def test_executable_runs_successfully(pack, MULE_dir): ''' This test is made to check if the current executable method for `bin/mule` to work as intended, accessing the relevant files when run. ''' - bin_dir = str(os.environ['MULE_DIR']) - # config will need to be improved - run_pack = ["python3", bin_dir + "/bin/mule", str(pack), "test_config"] + # config will need to be improved + run_pack = ["python3", MULE_dir + "/bin/mule", str(pack), "test_config"] # ensure output is successful (no errors) @@ -26,19 +25,17 @@ def test_executable_runs_successfully(pack): assert subprocess.run(run_pack).returncode == 0 -def test_incorrect_pack_returns_error(): - bin_dir = str(os.environ['MULE_DIR']) +def test_incorrect_pack_returns_error(MULE_dir): # give an incorrect pack - run_pack = ["python3", bin_dir + "/bin/mule", "donkey", "config"] + run_pack = ["python3", MULE_dir + "/bin/mule", "donkey", "config"] with raises(subprocess.CalledProcessError): subprocess.run(run_pack, check = True) -def test_config_read_correctly(): +def test_config_read_correctly(data_dir): - MULE_dir = str(os.environ['MULE_DIR']) - file_path = MULE_dir + '/packs/tests/data/configs/test_config.conf' + file_path = data_dir + 'configs/test_config.conf' expected_dict = {'test_1': 'a string', 'test_2': 6.03, 'test_3': 5, 'test_4': True} @@ -50,10 +47,9 @@ def test_config_read_correctly(): @mark.parametrize("config, error", [('malformed_header.conf', configparser.MissingSectionHeaderError), ('empty_entry.conf', SyntaxError), ('incorrect_format.conf', configparser.ParsingError)]) -def test_malformed_config(config, error): +def test_malformed_config(config, error, data_dir): # provides expected output when config file is malformed - MULE_dir = str(os.environ['MULE_DIR']) - file_path = MULE_dir + '/packs/tests/data/configs/' + config + file_path = data_dir + 'configs/' + config with raises(error): x = read_config_file(file_path) @@ -63,10 +59,9 @@ def test_malformed_config(config, error): ('single_multi_chan.conf', RuntimeError)]) # these will change to value errors when other # packs are implemented -def test_processing_catches(config, error): +def test_processing_catches(config, error, MULE_dir, data_dir): - MULE_dir = str(os.environ['MULE_DIR']) - config_path = MULE_dir + "/packs/tests/data/configs/" + config + config_path = data_dir + "configs/" + config run_pack = ["python3", MULE_dir + "/bin/mule", "proc", config_path]