# Testing the CLI import filecmp import os import pathlib import shutil import tarfile from unittest.mock import patch from delivery.context import DeliveryContext from delivery.delivery import Delivery, main def test_local_rawdata_no_tar(tmpdir_factory): """ Test that a directory in the source is copied to a directory in the destination in the manner expected. :return: """ temp_directory = str(tmpdir_factory.mktemp("test_basic_rawdata_no_tar")) test_data_path = "../../../../shared/workspaces/test/test_data/spool/724126739/" eb_name = "17A-109.sb33151331.eb33786546.57892.65940042824" main(["-r", "-l", temp_directory, test_data_path]) # compare the source and destination compare_dirs = filecmp.dircmp(temp_directory + "/" + eb_name, (test_data_path + eb_name)) # did the comparison report they are the same assert len(compare_dirs.left_only) == 0 assert len(compare_dirs.right_only) == 0 assert len(compare_dirs.funny_files) == 0 def test_local_rawdata_with_tar(tmpdir_factory): temp_directory = str(tmpdir_factory.mktemp("test_basic_rawdata_with_tar")) test_data_path = "../../../../shared/workspaces/test/test_data/spool/724126739/" main(["-r", "-t", "-l", temp_directory, test_data_path]) eb_name = "17A-109.sb33151331.eb33786546.57892.65940042824" tar_path = temp_directory + "/17A-109.sb33151331.eb33786546.57892.65940042824.tar" # does a tar exist where we think assert os.path.exists(tar_path) # do we only have it and the SHA1SUMS assert len(os.listdir(temp_directory)) == 2 # is it actually a tar assert tarfile.is_tarfile(tar_path) # lets unpack it shutil.unpack_archive(tar_path, temp_directory + "/extracted") # did it output what we expect assert os.path.exists(temp_directory + "/extracted/" + eb_name) # compare the extracted results with the source compare_dirs = filecmp.dircmp( temp_directory + "/extracted/" + eb_name, (test_data_path + eb_name) ) # is the source and extracted the same assert len(compare_dirs.left_only) == 0 assert len(compare_dirs.right_only) == 0 assert len(compare_dirs.funny_files) == 0 # @pytest.mark.skip(reason="Test needs more dev time") def test_web_rawdata_no_tar(tmpdir_factory): """ Test that a directory in the source is copied to a directory in the destination in the manner expected. """ temp_directory = pathlib.Path(tmpdir_factory.mktemp("test_web_rawdata_no_tar")) test_data_path = "../../../../shared/workspaces/test/test_data/spool/724126739/" eb_name = "17A-109.sb33151331.eb33786546.57892.65940042824" test_context = DeliveryContext.parse_commandline(["-r", test_data_path]) with patch("delivery.destinations.sharedweb.CapoConfig.settings") as mocked_capo_settings: mocked_capo_settings.return_value.nraoDownloadDirectory = str(temp_directory) mocked_capo_settings.return_value.nraoDownloadUrl = "http://testing" assert str(temp_directory) == mocked_capo_settings().nraoDownloadDirectory destination_url = Delivery().deliver(test_context) # determine the destination by looking at the URL actual_delivery_dir = temp_directory / destination_url.lstrip("http://testing") # compare the source and destination compare_dirs = filecmp.dircmp(actual_delivery_dir / eb_name, f"{test_data_path}{eb_name}") # did the comparison report they are the same assert len(compare_dirs.left_only) == 0 assert len(compare_dirs.right_only) == 0 assert len(compare_dirs.funny_files) == 0 def test_web_rawdata_with_tar(tmpdir_factory): temp_directory = pathlib.Path(tmpdir_factory.mktemp("test_web_rawdata_with_tar")) test_data_path = "../../../../shared/workspaces/test/test_data/spool/724126739/" test_context = DeliveryContext.parse_commandline(["-r", "-t", test_data_path]) with patch("delivery.destinations.sharedweb.CapoConfig.settings") as mocked_capo_settings: mocked_capo_settings.return_value.nraoDownloadDirectory = temp_directory mocked_capo_settings.return_value.nraoDownloadUrl = "http://testing" assert temp_directory == mocked_capo_settings().nraoDownloadDirectory destination_url = Delivery().deliver(test_context) eb_name = "17A-109.sb33151331.eb33786546.57892.65940042824" # determine the destination by looking at the URL actual_delivery_dir = temp_directory / destination_url.lstrip("http://testing") # does a tar exist where we think tar_path = actual_delivery_dir / (eb_name + ".tar") assert tar_path.exists() # is it the only thing there (did cleanup work) assert len(list(actual_delivery_dir.iterdir())) == 3 # is it actually a tar assert tarfile.is_tarfile(tar_path) # lets unpack it shutil.unpack_archive(tar_path, temp_directory / "extracted") # did it output what we expect assert (temp_directory / "extracted" / eb_name).exists() # compare the extracted results with the source compare_dirs = filecmp.dircmp( temp_directory / "extracted" / eb_name, (test_data_path + eb_name) ) # is the source and extracted the same assert len(compare_dirs.left_only) == 0 assert len(compare_dirs.right_only) == 0 assert len(compare_dirs.funny_files) == 0