from test_blob_download import * from test_upload_block_blob import * from test_upload_page_blob import * from test_blobfs_upload_sharedkey import * from test_blobfs_download_sharedkey import * from test_blob_piping import * from test_blob_sync import * from test_service_to_service_copy import * from test_google_cloud_storage_copy import * import glob, os import configparser import platform import sys import unittest import re def parse_config_file_set_env(): config = configparser.RawConfigParser() files_read = config.read('../test_suite_config.ini') if len(files_read) != 1: raise Exception("Failed to find/open test_suite_config.ini file") # get the platform type since config file has property for platform respectively. os_type = platform.system() os_type = os_type.upper() # check if the config are defined for current os type. platform_list = config.sections() try: platform_list.index(os_type) except: raise Exception("not able to find the config defined for ostype " + os_type) # set all the environment variables # TEST_DIRECTORY_PATH is the location where test_data folder will be created and test files will be created further. # set the environment variable TEST_DIRECTORY_PATH os.environ['TEST_DIRECTORY_PATH'] = config[os_type]['TEST_DIRECTORY_PATH'] # AZCOPY_EXECUTABLE_PATH is the location of the azcopy executable # azcopy executable will be copied to test data folder. # set the environment variables os.environ['AZCOPY_EXECUTABLE_PATH'] = config[os_type]['AZCOPY_EXECUTABLE_PATH'] # TEST_SUITE_EXECUTABLE_LOCATION is the location of the test suite executable # test suite executable will be copied to test data folder. # set the environment variable TEST_SUITE_EXECUTABLE_LOCATION os.environ['TEST_SUITE_EXECUTABLE_LOCATION'] = config[os_type]['TEST_SUITE_EXECUTABLE_LOCATION'] # CONTAINER_SAS_URL is the shared access signature of the container # where test data will be uploaded to and downloaded from. os.environ['CONTAINER_SAS_URL'] = config['CREDENTIALS']['CONTAINER_SAS_URL'] # container whose storage account has been configured properly for the interactive testing user. os.environ['CONTAINER_OAUTH_URL'] = config['CREDENTIALS']['CONTAINER_OAUTH_URL'] # container which should be same to CONTAINER_OAUTH_URL, while with SAS for validation purpose. os.environ['CONTAINER_OAUTH_VALIDATE_SAS_URL'] = config['CREDENTIALS']['CONTAINER_OAUTH_VALIDATE_SAS_URL'] # share_sas_url is the URL with SAS of the share where test data will be uploaded to and downloaded from. os.environ['SHARE_SAS_URL'] = config['CREDENTIALS']['SHARE_SAS_URL'] # container sas of the premium storage account. os.environ['PREMIUM_CONTAINER_SAS_URL'] = config['CREDENTIALS']['PREMIUM_CONTAINER_SAS_URL'] # set the account name for blob fs service operation os.environ['ACCOUNT_NAME'] = config['CREDENTIALS']['ACCOUNT_NAME'] # set the account key for blob fs service operation os.environ['ACCOUNT_KEY'] = config['CREDENTIALS']['ACCOUNT_KEY'] # set the filesystem url in the environment os.environ['FILESYSTEM_URL'] = config['CREDENTIALS']['FILESYSTEM_URL'] os.environ['FILESYSTEM_SAS_URL'] = config['CREDENTIALS']['FILESYSTEM_SAS_URL'] # set env var for service-2-service copy source blob account os.environ['S2S_SRC_BLOB_ACCOUNT_SAS_URL'] = config['CREDENTIALS']['S2S_SRC_BLOB_ACCOUNT_SAS_URL'] # set env var for service-2-service copy destination blob account os.environ['S2S_DST_BLOB_ACCOUNT_SAS_URL'] = config['CREDENTIALS']['S2S_DST_BLOB_ACCOUNT_SAS_URL'] # set env var for service-2-service copy source file account os.environ['S2S_SRC_FILE_ACCOUNT_SAS_URL'] = config['CREDENTIALS']['S2S_SRC_FILE_ACCOUNT_SAS_URL'] # set env var for service-2-service copy source s3 and gcp os.environ['S2S_SRC_S3_SERVICE_URL'] = config['CREDENTIALS']['S2S_SRC_S3_SERVICE_URL'] os.environ['S2S_SRC_GCP_SERVICE_URL'] = config['CREDENTIALS']['S2S_SRC_GCP_SERVICE_URL'] os.environ['AWS_ACCESS_KEY_ID'] = config['CREDENTIALS']['AWS_ACCESS_KEY_ID'] os.environ['AWS_SECRET_ACCESS_KEY'] = config['CREDENTIALS']['AWS_SECRET_ACCESS_KEY'] os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = config['CREDENTIALS']['GOOGLE_APPLICATION_CREDENTIALS'] os.environ['GOOGLE_CLOUD_PROJECT'] = config['CREDENTIALS']['GOOGLE_CLOUD_PROJECT'] os.environ['OAUTH_AAD_ENDPOINT'] = config['CREDENTIALS']['OAUTH_AAD_ENDPOINT'] os.environ['S3_TESTS_OFF'] = config['CREDENTIALS']['S3_TESTS_OFF'] os.environ['GCP_TESTS_OFF'] = config['CREDENTIALS']['GCP_TESTS_OFF'] def check_env_not_exist(key): if os.environ.get(key, '-1') == '-1': print('Environment variable: ' + key + ' not set.') return True return False def get_env_logged(key): value = os.environ.get(key) if value is None: print(key + " = None") else: print(key + " = " + re.sub("(?i)(?Psig[ \t]*[:=][ \t]*)(?P[^& ,;\t\n\r]+)", "sig=REDACTED", value)) return value def init(): # Check the environment variables. # If they are not set, then parse the config file and set # environment variables. If any of the env variable is not set # test_config_file is parsed and env variables are reset. if check_env_not_exist('TEST_DIRECTORY_PATH') or check_env_not_exist('AZCOPY_EXECUTABLE_PATH') or \ check_env_not_exist('TEST_SUITE_EXECUTABLE_LOCATION') or check_env_not_exist('CONTAINER_SAS_URL') or \ check_env_not_exist('CONTAINER_OAUTH_URL') or check_env_not_exist('CONTAINER_OAUTH_VALIDATE_SAS_URL') or \ check_env_not_exist('SHARE_SAS_URL') or check_env_not_exist('PREMIUM_CONTAINER_SAS_URL') or \ check_env_not_exist('FILESYSTEM_URL') or check_env_not_exist('FILESYSTEM_SAS_URL') or \ check_env_not_exist('ACCOUNT_NAME') or check_env_not_exist('ACCOUNT_KEY') or \ check_env_not_exist('S2S_SRC_BLOB_ACCOUNT_SAS_URL') or check_env_not_exist('S2S_DST_BLOB_ACCOUNT_SAS_URL') \ or check_env_not_exist('S2S_SRC_FILE_ACCOUNT_SAS_URL') or check_env_not_exist('S2S_SRC_S3_SERVICE_URL') or check_env_not_exist('S2S_SRC_GCP_SERVICE_URL'): parse_config_file_set_env() # Get the environment variables value # test_dir_path is the location where test_data folder will be created and test files will be created further. test_dir_path = get_env_logged('TEST_DIRECTORY_PATH') # azcopy_exec_location is the location of the azcopy executable # azcopy executable will be copied to test data folder. azcopy_exec_location = get_env_logged('AZCOPY_EXECUTABLE_PATH') # test_suite_exec_location is the location of the test suite executable # test suite executable will be copied to test data folder. test_suite_exec_location = get_env_logged('TEST_SUITE_EXECUTABLE_LOCATION') # container_sas is the shared access signature of the container # where test data will be uploaded to and downloaded from. container_sas = get_env_logged('CONTAINER_SAS_URL') # container_oauth is container for oauth testing. container_oauth = get_env_logged('CONTAINER_OAUTH_URL') # container_oauth_validate is the URL with SAS for oauth validation. container_oauth_validate = get_env_logged('CONTAINER_OAUTH_VALIDATE_SAS_URL') # share_sas_url is the URL with SAS of the share where test data will be uploaded to and downloaded from. share_sas_url = get_env_logged('SHARE_SAS_URL') # container sas of the premium storage account. premium_container_sas = get_env_logged('PREMIUM_CONTAINER_SAS_URL') # get the filesystem url filesystem_url = get_env_logged('FILESYSTEM_URL') filesystem_sas_url = get_env_logged('FILESYSTEM_SAS_URL') # get the s2s copy src URLs s2s_src_blob_account_url = get_env_logged('S2S_SRC_BLOB_ACCOUNT_SAS_URL') s2s_src_file_account_url = get_env_logged('S2S_SRC_FILE_ACCOUNT_SAS_URL') s2s_src_s3_service_url = get_env_logged('S2S_SRC_S3_SERVICE_URL') s2s_src_gcp_service_url = get_env_logged('S2S_SRC_GCP_SERVICE_URL') # get the s2s copy dest account URLs s2s_dst_blob_account_url = get_env_logged('S2S_DST_BLOB_ACCOUNT_SAS_URL') get_env_logged("ACCOUNT_NAME") # do NOT log ACCOUNT_KEY # don't log, it will just get redacted by DevOps logging system: get_env_logged("AWS_ACCESS_KEY_ID") # do NOT log AWS_SECRET_ACCESS_KEY get_env_logged("OAUTH_AAD_ENDPOINT") # don't log, it will just get redacted by DevOps logging system: get_env_logged("OAUTH_TENANT_ID") get_env_logged("S3_TESTS_OFF") get_env_logged("GCP_TESTS_OFF") # deleting the log files. cleanup() if not util.initialize_test_suite(test_dir_path, container_sas, container_oauth, container_oauth_validate, share_sas_url, premium_container_sas, filesystem_url, filesystem_sas_url, s2s_src_blob_account_url, s2s_src_file_account_url, s2s_src_s3_service_url, s2s_src_gcp_service_url, s2s_dst_blob_account_url, azcopy_exec_location, test_suite_exec_location): print("failed to initialize the test suite with given user input") return else: test_dir_path += "\\test_data" def cleanup(): # delete the log files for f in glob.glob('*.log'): try: os.remove(f) except OSError: pass def main(): print("Smoke tests starting...") init() test_class_to_run = [BlobPipingTests, Blob_Sync_User_Scenario, Block_Upload_User_Scenarios, Blob_Download_User_Scenario, PageBlob_Upload_User_Scenarios, BlobFs_Upload_ShareKey_User_Scenarios, BlobFs_Download_SharedKey_User_Scenarios, Service_2_Service_Copy_User_Scenario, Google_Cloud_Storage_Copy_User_Scenario] suites_list = [] loader = unittest.TestLoader() for test_class in test_class_to_run: suite = loader.loadTestsFromTestCase(test_class) suites_list.append(suite) big_suite = unittest.TestSuite(suites_list) runner = unittest.TextTestRunner(verbosity=2) results = runner.run(big_suite) if results.wasSuccessful() is False: sys.exit(1) # suite = unittest.TestLoader().loadTestsFromTestCase(Block_Upload_User_Scenarios) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(Blob_Download_User_Scenario) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(PageBlob_Upload_User_Scenarios) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(BlobFs_Upload_OAuth_User_Scenarios) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(BlobFs_Download_OAuth_User_Scenarios) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(Azcopy_Operation_User_Scenario) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(FileShare_Download_User_Scenario) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(FileShare_Upload_User_Scenario) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(BlobFs_Upload_ShareKey_User_Scenarios) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(BlobFs_Download_SharedKey_User_Scenarios) # unittest.TextTestRunner(verbosity=2).run(suite) # # suite = unittest.TestLoader().loadTestsFromTestCase(Service_2_Service_Copy_User_Scenario) # unittest.TextTestRunner(verbosity=2).run(suite) cleanup() if __name__ == '__main__': main()