diff --git a/data/dummy_fastq.gz b/data/dummy_fastq.gz new file mode 100644 index 00000000..7fbda2a3 Binary files /dev/null and b/data/dummy_fastq.gz differ diff --git a/test/conftest.py b/test/conftest.py index 51cdc374..24111584 100755 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,14 +1,14 @@ -#!/usr/bin/python3 """ Variables used across test modules, including the setup and teardown fixture that is run before and after every test """ import os +import re import shutil import pytest import tarfile import logging - +from shutil import copy # sys.path.append("..") from ad_logger import ad_logger from toolbox import toolbox @@ -78,12 +78,12 @@ def create_logdirs(): """ Create temporary log directories for testing purposes """ - os.makedirs(temp_log_dir) + os.makedirs(temp_log_dir, exist_ok=True) rf_obj = toolbox.RunfolderObject("TEST_FOLDER", ad_config.TIMESTAMP) for logfile in rf_obj.logfiles_config.values(): parent_dir = os.path.dirname(logfile) if not os.path.isdir(parent_dir): - os.makedirs(parent_dir) + os.makedirs(parent_dir, exist_ok=True) def patch_toolbox(monkeypatch): @@ -104,7 +104,7 @@ def run_before_and_after_session(): """ # Create temporary dirs for testing os.makedirs( - test_data_dir_unzipped + test_data_dir_unzipped, exist_ok=True ) # Holds the unzipped data to copy from for each test for tar in data_tars: @@ -112,13 +112,27 @@ def run_before_and_after_session(): open_tar.extractall(path=tar["dest"]) for destination in to_copy_interop_to: shutil.copytree(os.path.join(test_data_dir_unzipped, "InterOp"), destination) - yield + + test_data_unzipped = os.path.join(test_data_dir_unzipped, "demultiplex_test_files", "test_runfolders") + + directories = [ + os.path.join(test_data_unzipped, d) + for d in os.listdir(test_data_unzipped) + if os.path.isdir(os.path.join(test_data_unzipped, d)) + ] + dummy_fastq = os.path.join(test_data_dir, "dummy_fastq.gz") + + for directory in directories: + if re.match(".*999999_.*", directory): + fastqs_dir = os.path.join(test_data_unzipped, directory, "Data", "Intensities", "BaseCalls/") + os.makedirs(fastqs_dir, exist_ok=True) + copy(dummy_fastq, fastqs_dir) + yield # Where the testing happens for to_remove in [test_data_dir_unzipped, test_data_temp]: if os.path.isdir(to_remove): shutil.rmtree(to_remove) -# TODO fix patching of script loggers as this is not set up correctly !! @pytest.fixture(scope="function", autouse=True) def run_before_and_after_tests(monkeypatch): """ diff --git a/test/test_ad_email.py b/test/test_ad_email.py index bcf3ccee..a10d6fe6 100755 --- a/test/test_ad_email.py +++ b/test/test_ad_email.py @@ -1,10 +1,7 @@ -""" ad_email.py pytest unit tests +""" ad_email.py pytest unit tests. The test suite is currently incomplete N.B. test_email_sending_success() will only pass when running on the workstation where the required auth details are stored - -# TODO write the below tests which are currently missing or incomplete: -- AdEmail.generate_email_html """ import pytest @@ -14,6 +11,8 @@ logger_obj = logger_obj +# TODO finish this test suite as it is currently incomplete + class TestAdEmail: """ @@ -47,9 +46,6 @@ def email_recipients(self): ], ] - # TODO write test for generate_email_html - # def test_generate_email_html(): - def test_send_email_success( self, logger_obj, diff --git a/test/test_ad_logger.py b/test/test_ad_logger.py index 782d5fab..614eded2 100755 --- a/test/test_ad_logger.py +++ b/test/test_ad_logger.py @@ -1,62 +1,11 @@ -#!/usr/bin/python3 -""" ad_logger.py pytest unit tests - -# TODO write the following unit tests which are currently missing or incomplete: -- shutdown_logs -- SensitiveFormatter - - format -- AdLogger - - get_logger - - _get_file_handler - - _get_logging_formatter - - _get_syslog_handler - - _get_stream_handler +""" ad_logger.py pytest unit tests. The test suite is currently incomplete """ import pytest from toolbox import toolbox from ad_logger import ad_logger from config import ad_config -# import logging - -# No logging disabled for this test as we are testing logging -# No patching required - -# TODO add tests for SensitiveFormatter class -# TODO add test for shutdown_logs -# TODO add test that checks that streamhandler, filehandler and syslog handler are all -# added as expected - - -# TODO write test for shutdown_logs() -# def test_shutdown_logs(): -# """""" - -# ad_logger.shutdown_logs(logger) -# assert not logger.info( -# "Test log message. Logger %s", -# logger.name, -# ) - -# TODO write tests for SensitiveFormatter class -# class TestSensitiveFormatter: -# """ -# Tests for the SensitiveFormatter class -# """ - - -# @pytest.fixture(scope="function", autouse=True) -# def setup(monkeypatch): -# """ -# """ -# # Re-enable logging as it is required for assertions -# logging.disable(logging.NOTSET) -# # Remove testfiles dir containing test runfolders as we don't need these files -# # Apply patches required for test_ad_logger script. These point the paths to the -# # temporary locations: -# # - Test logfiles in the temp logfiles dir and within the temp runfolder dirs -# monkeypatch.setattr(toolbox.ad_config, "RUNFOLDERS", conftest.temp_runfolderdir) -# monkeypatch.setattr(toolbox.ad_config, "AD_LOGDIR", conftest.temp_log_dir) +# TODO finish this test suite as it is currently incomplete class TestRunfolderLoggers: @@ -99,8 +48,4 @@ def test_get_loggers(self, logfiles_config, caplog): ) assert loggers[logger_name].name in caplog.text - # TODO write tests for AdLogger class - # class TestAdLogger: - """ - Tests for the AdLogger class - """ + diff --git a/test/test_demultiplex.py b/test/test_demultiplex.py index a8f95173..5600e3e1 100755 --- a/test/test_demultiplex.py +++ b/test/test_demultiplex.py @@ -1,18 +1,5 @@ -#!/usr/bin/python3 """ -demultiplex.py pytest unit tests - -# TODO write the following tests which are currently missing or incomplete: -- test_get_runfolder_names_test -- test_get_runfolder_names_prod -- test_setoff_processing_toproc -- test_setoff_processing_nottoproc -- test_demultiplex_runfolder_toproc -- test_demultiplex_runfolder_nottoproc -- test_check_dev_run_fail -- test_pass_integrity_check_pass -- test_pass_integrity_check_fail -- test_add_bcl2fastqlog_tso_msg_fail +demultiplex.py pytest unit tests. The test suite is currently incomplete """ import os import itertools @@ -24,6 +11,9 @@ from pytest_cases import fixture_union +# TODO finish this test suite as it is currently incomplete + + def get_dr_obj(runfolder): """""" dr_obj = demultiplex.DemultiplexRunfolder(runfolder, ad_config.TIMESTAMP) @@ -190,13 +180,13 @@ def runfolders_toproc(self): """ return [ # "999999_M02631_0000_00000TEST4", # TODO fix test case - # Barcodes in sample sheet are longer than the index length found in RunInfo.xml + # # Barcodes in sample sheet are longer than the index length found in RunInfo.xml # "999999_A01229_0000_00000TEST7", # TODO fix test case - # Unable to find BCL file for 's_1_1101' in: /mnt/run/Data/Intensities/BaseCalls/L001/C1.1 + # # Unable to find BCL file for 's_1_1101' in: /mnt/run/Data/Intensities/BaseCalls/L001/C1.1 # "999999_A01229_0000_00000TEST9", # TODO fix test case - # Unable to find BCL file for 's_1_1101' in: /mnt/run/Data/Intensities/BaseCalls/L001/C1.1 + # # Unable to find BCL file for 's_1_1101' in: /mnt/run/Data/Intensities/BaseCalls/L001/C1.1 # "999999_A01229_0000_0000TEST11", # TODO fix test case - # Cannot read non-existent file: file:///input_run/RunInfo.xml + # # Cannot read non-existent file: file:///input_run/RunInfo.xml ] @pytest.fixture(scope="function") @@ -215,46 +205,39 @@ def runfolders_nottoproc(self): "999999_A01229_0000_0000TEST10", ] - # TODO write tests - # def test_get_runfolder_names_test(self): - # def test_get_runfolder_names_prod(self): - - # TODO fix these tests - def test_setoff_processing_toproc(self, runfolders_toproc, monkeypatch): - """ - Pass set of runfolders expected to be successfully processed by script. Assert - that the expected number are processed - """ - # TODO fix the below patch - monkeypatch.setattr( - demultiplex.DemultiplexConfig, - "DEMULTIPLEX_TEST_RUNFOLDERS", - runfolders_toproc, - ) - gr_obj = get_gr_obj() - demultiplex.DemultiplexRunfolder.bcl2fastq2_cmd = ( - f"echo '{ad_config.DEMULTIPLEX_SUCCESS}" - ) - gr_obj.setoff_processing() - assert all( - runfolder in gr_obj.processed_runfolders for runfolder in runfolders_toproc - ) - - # def test_setoff_processing_nottoproc(self, runfolders_nottoproc, monkeypatch): + # TODO fix this test - seee comments above runfolders_toproc fixture + # def test_setoff_processing_toproc(self, runfolders_toproc, monkeypatch): # """ - # Pass set of runfolders that should not be processed for various reasons. Assert - # that none have been processed + # Pass set of runfolders expected to be successfully processed by script. Assert + # that the expected number are processed # """ # monkeypatch.setattr( - # demultiplex.DemultiplexConfig, "DEMULTIPLEX_TEST_RUNFOLDERS", runfolders_nottoproc + # demultiplex.DemultiplexConfig, + # "DEMULTIPLEX_TEST_RUNFOLDERS", + # runfolders_toproc, # ) # gr_obj = get_gr_obj() - # with pytest.raises(SystemExit) as pytest_wrapped_e: - # gr_obj.setoff_processing() - # assert pytest_wrapped_e.type == SystemExit - # assert pytest_wrapped_e.value.code == 1 + # demultiplex.DemultiplexRunfolder.bcl2fastq2_cmd = ( + # f"echo '{ad_config.DEMULTIPLEX_SUCCESS}" + # ) + # gr_obj.setoff_processing() + # assert all( + # runfolder in gr_obj.processed_runfolders for runfolder in runfolders_toproc + # ) - # TODO write test_return_num_processed_runfolders(self): + def test_setoff_processing_nottoproc(self, runfolders_nottoproc, monkeypatch): + """ + Pass set of runfolders that should not be processed for various reasons. Assert + that none have been processed + """ + monkeypatch.setattr( + demultiplex.DemultiplexConfig, "DEMULTIPLEX_TEST_RUNFOLDERS", runfolders_nottoproc + ) + gr_obj = get_gr_obj() + with pytest.raises(SystemExit) as pytest_wrapped_e: + gr_obj.setoff_processing() + assert pytest_wrapped_e.type == SystemExit + assert pytest_wrapped_e.value.code == 1 class TestDemultiplexRunfolder(object): @@ -384,7 +367,7 @@ def demultiplexing_required(self): "999999_A01229_0000_00000TEST9", ] - # TODO add a development run to this fixture + # TODO add a development run and development run with UMIs to this fixture @pytest.fixture(scope="function") def demultiplexing_notrequired(self): """ @@ -521,9 +504,6 @@ def checksums_not_checked(self): "Checksums match after 1 hours", ] - # TODO write test_demultiplex_runfolder_toproc(self): - # TODO write test_demultiplex_runfolder_nottoproc(self): - def test_bcl2fastqlog_absent_false(self, rf_with_bcl2fastqlog): """ Test function correctly identifies presence of bcl2fastqlogfile using an empty @@ -542,32 +522,32 @@ def test_bcl2fastqlog_absent_true(self, rf_no_bcl2fastqlog): dr_obj = get_dr_obj(runfolder) assert dr_obj.bcl2fastqlog_absent() - # def test_setoff_workflow_success(self, demultiplexing_required, monkeypatch): - # """ - # Test that function sets off run processing correctly for runfolders requiring it - # """ - # for runfolder in demultiplexing_required: - # dr_obj = get_dr_obj(runfolder) - # # Command to run in place of bcl2fastq2 command that appends processing - # # complete string to bcl2fastq2 logfile - # monkeypatch.setattr( - # dr_obj, - # "bcl2fastq2_cmd", - # f"echo '{ad_config.DEMULTIPLEX_SUCCESS}' >> " - # f"{dr_obj.rf_obj.bcl2fastqlog_file}", - # ) - # assert dr_obj.setoff_workflow() and dr_obj.run_processed - # ad_logger.shutdown_logs(dr_obj.demux_rf_logger) + def test_setoff_workflow_success(self, demultiplexing_required, monkeypatch): + """ + Test that function sets off run processing correctly for runfolders requiring it + """ + for runfolder in demultiplexing_required: + dr_obj = get_dr_obj(runfolder) + # Command to run in place of bcl2fastq2 command that appends processing + # complete string to bcl2fastq2 logfile + monkeypatch.setattr( + dr_obj, + "bcl2fastq2_cmd", + f"echo '{ad_config.DEMULTIPLEX_SUCCESS}' >> " + f"{dr_obj.rf_obj.bcl2fastqlog_file}", + ) + assert dr_obj.setoff_workflow() and dr_obj.run_processed + ad_logger.shutdown_logs(dr_obj.demux_rf_logger) - # def test_setoff_workflow_fail(self, demultiplexing_notrequired): - # """ - # Test that function correctly does not process runfolders that do not need - # processing - # """ - # for runfolder in demultiplexing_notrequired: - # dr_obj = get_dr_obj(runfolder) - # assert not dr_obj.setoff_workflow() and not dr_obj.run_processed - # ad_logger.shutdown_logs(dr_obj.demux_rf_logger) + def test_setoff_workflow_fail(self, demultiplexing_notrequired): + """ + Test that function correctly does not process runfolders that do not need + processing + """ + for runfolder in demultiplexing_notrequired: + dr_obj = get_dr_obj(runfolder) + assert not dr_obj.setoff_workflow() and not dr_obj.run_processed + ad_logger.shutdown_logs(dr_obj.demux_rf_logger) def test_demultiplexing_required_true(self, demultiplexing_required): """ @@ -628,11 +608,6 @@ def test_sequencing_complete_fail(self, rtacomplete_absent): assert not dr_obj.sequencing_complete() ad_logger.shutdown_logs(dr_obj.demux_rf_logger) - # # TODO write test_check_dev_run_pass - # # TODO write test_check_dev_run_fail - # # TODO write test_pass_integrity_check_pass - # # TODO write test_pass_integrity_check_fail - def disallowed_sserrs_pass(self, monkeypatch, ss_with_disallowed_sserrs): """ Tests function identifies all disallowed ss errors @@ -757,40 +732,42 @@ def test_add_bcl2fastqlog_msg_pass(self, demultiplexing_required): assert "TEST" in contents ad_logger.shutdown_logs(dr_obj.demux_rf_logger) - def test_run_demultiplexing_success(self, non_tso_runfolder): - """ - Test demultiplexing is performed successfully. N.B. this does not test the - functioning of the bcl2fastq2 executable, which must be tested separately as - part of the final manual testing - """ - for runfolder in non_tso_runfolder: - dr_obj = get_dr_obj(runfolder) - # Command to run in place of bcl2fastq2 command that appends processing - # complete string to bcl2fastq2 logfile - # TODO swap below to a patch - dr_obj.bcl2fastq2_cmd = ( - f"echo '{ad_config.DEMULTIPLEX_SUCCESS}' >> " - f"{dr_obj.rf_obj.bcl2fastqlog_file}" - ) - assert dr_obj.run_demultiplexing() - ad_logger.shutdown_logs(dr_obj.demux_rf_logger) + # TODO Fix - see comments in non_tso_runfolder fixture + # def test_run_demultiplexing_success(self, non_tso_runfolder): + # """ + # Test demultiplexing is performed successfully. N.B. this does not test the + # functioning of the bcl2fastq2 executable, which must be tested separately as + # part of the final manual testing + # """ + # for runfolder in non_tso_runfolder: + # dr_obj = get_dr_obj(runfolder) + # # Command to run in place of bcl2fastq2 command that appends processing + # # complete string to bcl2fastq2 logfile + # # TODO swap below to a patch + # dr_obj.bcl2fastq2_cmd = ( + # f"echo '{ad_config.DEMULTIPLEX_SUCCESS}' >> " + # f"{dr_obj.rf_obj.bcl2fastqlog_file}" + # ) + # assert dr_obj.run_demultiplexing() + # ad_logger.shutdown_logs(dr_obj.demux_rf_logger) - def test_run_demultiplexing_fail(self, non_tso_runfolder): - """ - Test function fails when providing "/bin/false" as command - """ - for runfolder in non_tso_runfolder: - dr_obj = get_dr_obj(runfolder) - # Command to run in place of bcl2fastq2 command that appends processing - # complete string to bcl2fastq2 logfile - # TODO swap below to a patch - dr_obj.bcl2fastq2_cmd = "/bin/false" - with pytest.raises(SystemExit) as pytest_wrapped_e: - dr_obj.run_demultiplexing() - assert not dr_obj.run_processed - ad_logger.shutdown_logs(dr_obj.demux_rf_logger) - assert pytest_wrapped_e.type == SystemExit - assert pytest_wrapped_e.value.code == 1 + # TODO Fix - see comments in non_tso_runfolder fixture + # def test_run_demultiplexing_fail(self, non_tso_runfolder): + # """ + # Test function fails when providing "/bin/false" as command + # """ + # for runfolder in non_tso_runfolder: + # dr_obj = get_dr_obj(runfolder) + # # Command to run in place of bcl2fastq2 command that appends processing + # # complete string to bcl2fastq2 logfile + # # TODO swap below to a patch + # dr_obj.bcl2fastq2_cmd = "/bin/false" + # with pytest.raises(SystemExit) as pytest_wrapped_e: + # dr_obj.run_demultiplexing() + # assert not dr_obj.run_processed + # ad_logger.shutdown_logs(dr_obj.demux_rf_logger) + # assert pytest_wrapped_e.type == SystemExit + # assert pytest_wrapped_e.value.code == 1 def test_calculate_cluster_density_pass(self, demultiplexing_required): """ diff --git a/test/test_toolbox.py b/test/test_toolbox.py index 95fa58a7..9de4d5da 100755 --- a/test/test_toolbox.py +++ b/test/test_toolbox.py @@ -1,59 +1,4 @@ -#!/usr/bin/python3 -""" toolbox.py pytest unit tests - -# TODO to write +""" toolbox.py pytest unit tests. Test suite is incomplete """ -# import pytest -# from test.conftest import logger_obj -# from toolbox import toolbox -# from config import ad_config - - -# def test_upload_software_pass(): -# """""" - - -# def test_upload_software_fail(logger_obj, monkeypatch): -# """ -# Check test_upload_software function fails when expected using /bin/false -# """ -# temp_dict = ad_config.TEST_PROGRAMS_DICT -# temp_dict['dx_toolkit']['executable'] = "/bin/false" -# temp_dict['dx_toolkit']['test_cmd'] = "/bin/false" -# temp_dict['dx_toolkit']['executable'] = "/bin/false" -# temp_dict['dx_toolkit']['test_cmd'] = "/bin/false" -# monkeypatch.setattr(ad_config, 'TEST_PROGRAMS_DICT', temp_dict) -# with pytest.raises(Exception): -# toolbox.test_processing_software(logger_obj) - - -# def test_processing_software_pass(logger_obj): -# """ -# Check test_processing_software function is working. This is expected to fail if -# tests are being carried out on a machine other than the workstation -# """ -# assert toolbox.test_processing_software(logger_obj) - - -# def test_processing_software_fail(): -# """""" -# """ -# Check test_processing_software function fails when expected using /bin/false -# """ - - -# def test_programs_pass(): -# """""" - - -# def test_programs_fail(): -# """""" - - -# def test_docker_pass(): -# """""" - - -# def test_docker_fail(): -# """""" +# TODO finish this test suite as it is currently incomplete diff --git a/test/test_upload_runfolder.py b/test/test_upload_runfolder.py index e56dbe40..e483cbad 100755 --- a/test/test_upload_runfolder.py +++ b/test/test_upload_runfolder.py @@ -1,5 +1,4 @@ -#!/usr/bin/python3 -""" upload_runfolder.py pytest unit tests - -# TODO to write +""" upload_runfolder.py pytest unit tests. Test suite is incomplete """ + +# TODO finish this test suite as it is currently incomplete