tests/listgen.py
def _cli_listgen_helper(db_conn, tmpdir, sub_temp_dir, mocked_config, date=None, base_run_id=None,  # noqa: C901
                        no_full_list=None, no_clean_up=None, unzip_files=True, combine_deltas=True,
                        disable_sanity_checks=True, conditions=None):
    """Helper function for CLI list-gen."""
    options_list = []
    if date:
        options_list.extend(['--curr-date', date])
    if base_run_id:
        options_list.extend(['--base', base_run_id])
    if no_full_list:
        options_list.extend(['--no-full-lists'])
    if no_clean_up:
        options_list.extend(['--no-cleanup'])
    if conditions:
        options_list.extend(['--conditions', conditions])
    options_list.extend(['--disable-sanity-checks'])
    output_dir = str(tmpdir.mkdir(sub_temp_dir))
    options_list.append(output_dir)
    runner = CliRunner()
    result = runner.invoke(dirbs_listgen_cli, options_list, obj={'APP_CONFIG': mocked_config},
                           catch_exceptions=False)
    assert result.exit_code == 0
    job_record_list = query_for_command_runs(db_conn, 'dirbs-listgen')
    db_conn.commit()
    assert job_record_list
    run_id = [x.run_id for x in job_record_list][0]

    # If requested, auto unzip all the files as well as many test rely on the .csv files being there
    if unzip_files:
        listgen_path = find_subdirectory_in_dir('listgen*', output_dir)
        for zip_path in glob.glob(os.path.join(listgen_path, '*.zip')):
            with zipfile.ZipFile(zip_path, 'r') as zf:
                zf.extractall(path=listgen_path)

        # If requested, combine all delta files into a single file containing an extra change_type column to match
        # the previous spec that the tests were based off
        if combine_deltas:
            for delta_csv_path in glob.glob(os.path.join(listgen_path, '*delta*.csv')):
                fn = os.path.basename(delta_csv_path)
                file_type = re.sub(r'^\d+_\d+_([a-z]+)_.*$', r'\1', fn)
                date_str = re.sub(r'^(.*)_{0}.*$'.format(file_type), r'\1', fn)
                run_id_range = re.sub(r'^.*_delta_([-0-9]+_[0-9]+)_.*$', r'\1', fn)
                change_type = re.sub(r'^.*{0}_(.*)\.csv$'.format(run_id_range), r'\1', fn)

                if file_type == 'blacklist':
                    combined_fn = os.path.join(listgen_path,
                                               '{0}_blacklist_delta_{1}.csv'.format(date_str, run_id_range))
                else:
                    operator_id = re.sub(r'^.*_{0}_(.*)_delta.*csv$'.format(file_type), r'\1', fn)
                    combined_fn = os.path.join(listgen_path,
                                               '{0}_{1}_{2}_delta_{3}.csv'.format(date_str,
                                                                                  file_type,
                                                                                  operator_id,
                                                                                  run_id_range))

                write_header = True if not os.path.exists(combined_fn) else False
                with open(delta_csv_path, 'r') as input_file, open(combined_fn, 'a') as output_file:
                    input_lines = input_file.read().splitlines()
                    if write_header:
                        output_file.write(input_lines[0] + ',change_type\n')
                    for input_line in input_lines[1:]:
                        output_file.write(input_line + ',{0}\n'.format(change_type))

                os.remove(delta_csv_path)

    return run_id, output_dir
Helper function for CLI list-gen.
Helper function for CLI list-gen.
Helper function for CLI list - gen .
def _cli_listgen_helper(db_conn, tmpdir, sub_temp_dir, mocked_config, date=None, base_run_id=None,  
                        no_full_list=None, no_clean_up=None, unzip_files=True, combine_deltas=True,
                        disable_sanity_checks=True, conditions=None):
    options_list = []
    if date:
        options_list.extend(['--curr-date', date])
    if base_run_id:
        options_list.extend(['--base', base_run_id])
    if no_full_list:
        options_list.extend(['--no-full-lists'])
    if no_clean_up:
        options_list.extend(['--no-cleanup'])
    if conditions:
        options_list.extend(['--conditions', conditions])
    options_list.extend(['--disable-sanity-checks'])
    output_dir = str(tmpdir.mkdir(sub_temp_dir))
    options_list.append(output_dir)
    runner = CliRunner()
    result = runner.invoke(dirbs_listgen_cli, options_list, obj={'APP_CONFIG': mocked_config},
                           catch_exceptions=False)
    assert result.exit_code == 0
    job_record_list = query_for_command_runs(db_conn, 'dirbs-listgen')
    db_conn.commit()
    assert job_record_list
    run_id = [x.run_id for x in job_record_list][0]
    if unzip_files:
        listgen_path = find_subdirectory_in_dir('listgen*', output_dir)
        for zip_path in glob.glob(os.path.join(listgen_path, '*.zip')):
            with zipfile.ZipFile(zip_path, 'r') as zf:
                zf.extractall(path=listgen_path)
        if combine_deltas:
            for delta_csv_path in glob.glob(os.path.join(listgen_path, '*delta*.csv')):
                fn = os.path.basename(delta_csv_path)
                file_type = re.sub(r'^\d+_\d+_([a-z]+)_.*$', r'\1', fn)
                date_str = re.sub(r'^(.*)_{0}.*$'.format(file_type), r'\1', fn)
                run_id_range = re.sub(r'^.*_delta_([-0-9]+_[0-9]+)_.*$', r'\1', fn)
                change_type = re.sub(r'^.*{0}_(.*)\.csv$'.format(run_id_range), r'\1', fn)
                if file_type == 'blacklist':
                    combined_fn = os.path.join(listgen_path,
                                               '{0}_blacklist_delta_{1}.csv'.format(date_str, run_id_range))
                else:
                    operator_id = re.sub(r'^.*_{0}_(.*)_delta.*csv$'.format(file_type), r'\1', fn)
                    combined_fn = os.path.join(listgen_path,
                                               '{0}_{1}_{2}_delta_{3}.csv'.format(date_str,
                                                                                  file_type,
                                                                                  operator_id,
                                                                                  run_id_range))
                write_header = True if not os.path.exists(combined_fn) else False
                with open(delta_csv_path, 'r') as input_file, open(combined_fn, 'a') as output_file:
                    input_lines = input_file.read().splitlines()
                    if write_header:
                        output_file.write(input_lines[0] + ',change_type\n')
                    for input_line in input_lines[1:]:
                        output_file.write(input_line + ',{0}\n'.format(change_type))
                os.remove(delta_csv_path)
    return run_id, output_dir
def _cli_listgen_helper ( db_conn , tmpdir , sub_temp_dir , mocked_config , date = None , base_run_id = None , no_full_list = None , no_clean_up = None , unzip_files = True , combine_deltas = True , disable_sanity_checks = True , conditions = None ) : options_list = [ ] if date : options_list . extend ( [ '--curr-date' , date ] ) if base_run_id : options_list . extend ( [ '--base' , base_run_id ] ) if no_full_list : options_list . extend ( [ '--no-full-lists' ] ) if no_clean_up : options_list . extend ( [ '--no-cleanup' ] ) if conditions : options_list . extend ( [ '--conditions' , conditions ] ) options_list . extend ( [ '--disable-sanity-checks' ] ) output_dir = str ( tmpdir . mkdir ( sub_temp_dir ) ) options_list . append ( output_dir ) runner = CliRunner ( ) result = runner . invoke ( dirbs_listgen_cli , options_list , obj = { 'APP_CONFIG' : mocked_config } , catch_exceptions = False ) assert result . exit_code == 0 job_record_list = query_for_command_runs ( db_conn , 'dirbs-listgen' ) db_conn . commit ( ) assert job_record_list run_id = [ x . run_id for x in job_record_list ] [ 0 ] if unzip_files : listgen_path = find_subdirectory_in_dir ( 'listgen*' , output_dir ) for zip_path in glob . glob ( os . path . join ( listgen_path , '*.zip' ) ) : with zipfile . ZipFile ( zip_path , 'r' ) as zf : zf . extractall ( path = listgen_path ) if combine_deltas : for delta_csv_path in glob . glob ( os . path . join ( listgen_path , '*delta*.csv' ) ) : fn = os . path . basename ( delta_csv_path ) file_type = re . sub ( r'^\d+_\d+_([a-z]+)_.*$' , r'\1' , fn ) date_str = re . sub ( r'^(.*)_{0}.*$' . format ( file_type ) , r'\1' , fn ) run_id_range = re . sub ( r'^.*_delta_([-0-9]+_[0-9]+)_.*$' , r'\1' , fn ) change_type = re . sub ( r'^.*{0}_(.*)\.csv$' . format ( run_id_range ) , r'\1' , fn ) if file_type == 'blacklist' : combined_fn = os . path . join ( listgen_path , '{0}_blacklist_delta_{1}.csv' . format ( date_str , run_id_range ) ) else : operator_id = re . sub ( r'^.*_{0}_(.*)_delta.*csv$' . format ( file_type ) , r'\1' , fn ) combined_fn = os . path . join ( listgen_path , '{0}_{1}_{2}_delta_{3}.csv' . format ( date_str , file_type , operator_id , run_id_range ) ) write_header = True if not os . path . exists ( combined_fn ) else False with open ( delta_csv_path , 'r' ) as input_file , open ( combined_fn , 'a' ) as output_file : input_lines = input_file . read ( ) . splitlines ( ) if write_header : output_file . write ( input_lines [ 0 ] + ',change_type\n' ) for input_line in input_lines [ 1 : ] : output_file . write ( input_line + ',{0}\n' . format ( change_type ) ) os . remove ( delta_csv_path ) return run_id , output_dir
Helper function for CLI list-gen.
Helper function for CLI list - gen .
# noqa: C901 """Helper function for CLI list-gen.""" # If requested, auto unzip all the files as well as many test rely on the .csv files being there # If requested, combine all delta files into a single file containing an extra change_type column to match # the previous spec that the tests were based off
db_conn
tmpdir
sub_temp_dir
mocked_config
date
base_run_id
no_full_list
no_clean_up
unzip_files
combine_deltas
disable_sanity_checks
conditions