tests/_delta_helpers.py
def write_import_csv(tmpdir, importer_name, csv_imei_change_type_tuples, delta_import=True):
    """Helper function to create a csv input file.

    Takes in input the importer name to get metadata for csv header and a dict to map change_type to the imei_norm.
    i.e. csv_imei_change_type_tuples = [('52345678901234', 'remove'), ('22345678901234', 'add')];
         importer_name = pairing_list
    """
    valid_csv_import_data_file_name = 'sample_import_list.csv'
    valid_csv_import_data_file = str(tmpdir.join(valid_csv_import_data_file_name))
    with open(valid_csv_import_data_file, 'w') as csvfile:
        # generate metadata dict for the current importer
        importer_metadata = importer_to_fields_dict[importer_name]
        csv_imei_field_name = importer_metadata['csv_imei_field_name']
        extra_field_names = importer_metadata['extra_field_names']
        # add a default value for extra field if there is any

        # generate csv header (header is a list of fieldnames) i.e. [imei, imsi, change_type]
        csv_field_name_list = [csv_imei_field_name]
        csv_field_name_list.extend(extra_field_names)
        if delta_import:
            csv_field_name_list.append('change_type')
        # generate a list of csv rows to be written
        rows_to_write_list = []
        for imei_norm, change_type in csv_imei_change_type_tuples:
            row_dict = {csv_imei_field_name: imei_norm}
            extra_field_to_values = {'make': '',
                                     'model': '',
                                     'status': '',
                                     'imsi': '11111111111111',
                                     'msisdn': '2322211122112',
                                     'reporting_date': '20160420',
                                     'model_number': '',
                                     'brand_name': '',
                                     'device_type': '',
                                     'radio_interface': '',
                                     'device_id': '123'
                                     }
            for e in extra_field_names:
                row_dict.update({e: extra_field_to_values[e]})
            # in full_import mode change_type is Null
            if change_type:
                assert delta_import
                row_dict.update({'change_type': change_type})
            rows_to_write_list.append(row_dict)
            print(rows_to_write_list)
        # write into csv
        writer = csv.DictWriter(csvfile, fieldnames=csv_field_name_list)
        writer.writeheader()
        for d in rows_to_write_list:
            writer.writerow(d)

    valid_zip_import_data_file_path = str(tmpdir.join('sample_import_list.zip'))
    with zipfile.ZipFile(valid_zip_import_data_file_path, 'w') as valid_csv_import_data_file_zfile:
        valid_csv_import_data_file_zfile.write(valid_csv_import_data_file,
                                               valid_csv_import_data_file_name)
    return valid_zip_import_data_file_path
Helper function to create a csv input file.

    Takes in input the importer name to get metadata for csv header and a dict to map change_type to the imei_norm.
    i.e. csv_imei_change_type_tuples = [('52345678901234', 'remove'), ('22345678901234', 'add')];
         importer_name = pairing_list
    
Helper function to create a csv input file.
Takes in input the importer name to get metadata for csv header and a dict to map change_type to the imei_norm.
Helper function to create a csv input file . Takes in input the importer name to get metadata for csv header and a dict to map change_type to the imei_norm .
def write_import_csv(tmpdir, importer_name, csv_imei_change_type_tuples, delta_import=True):
    valid_csv_import_data_file_name = 'sample_import_list.csv'
    valid_csv_import_data_file = str(tmpdir.join(valid_csv_import_data_file_name))
    with open(valid_csv_import_data_file, 'w') as csvfile:
        importer_metadata = importer_to_fields_dict[importer_name]
        csv_imei_field_name = importer_metadata['csv_imei_field_name']
        extra_field_names = importer_metadata['extra_field_names']
        csv_field_name_list = [csv_imei_field_name]
        csv_field_name_list.extend(extra_field_names)
        if delta_import:
            csv_field_name_list.append('change_type')
        rows_to_write_list = []
        for imei_norm, change_type in csv_imei_change_type_tuples:
            row_dict = {csv_imei_field_name: imei_norm}
            extra_field_to_values = {'make': '',
                                     'model': '',
                                     'status': '',
                                     'imsi': '11111111111111',
                                     'msisdn': '2322211122112',
                                     'reporting_date': '20160420',
                                     'model_number': '',
                                     'brand_name': '',
                                     'device_type': '',
                                     'radio_interface': '',
                                     'device_id': '123'
                                     }
            for e in extra_field_names:
                row_dict.update({e: extra_field_to_values[e]})
            if change_type:
                assert delta_import
                row_dict.update({'change_type': change_type})
            rows_to_write_list.append(row_dict)
            print(rows_to_write_list)
        writer = csv.DictWriter(csvfile, fieldnames=csv_field_name_list)
        writer.writeheader()
        for d in rows_to_write_list:
            writer.writerow(d)
    valid_zip_import_data_file_path = str(tmpdir.join('sample_import_list.zip'))
    with zipfile.ZipFile(valid_zip_import_data_file_path, 'w') as valid_csv_import_data_file_zfile:
        valid_csv_import_data_file_zfile.write(valid_csv_import_data_file,
                                               valid_csv_import_data_file_name)
    return valid_zip_import_data_file_path
def write_import_csv ( tmpdir , importer_name , csv_imei_change_type_tuples , delta_import = True ) : valid_csv_import_data_file_name = 'sample_import_list.csv' valid_csv_import_data_file = str ( tmpdir . join ( valid_csv_import_data_file_name ) ) with open ( valid_csv_import_data_file , 'w' ) as csvfile : importer_metadata = importer_to_fields_dict [ importer_name ] csv_imei_field_name = importer_metadata [ 'csv_imei_field_name' ] extra_field_names = importer_metadata [ 'extra_field_names' ] csv_field_name_list = [ csv_imei_field_name ] csv_field_name_list . extend ( extra_field_names ) if delta_import : csv_field_name_list . append ( 'change_type' ) rows_to_write_list = [ ] for imei_norm , change_type in csv_imei_change_type_tuples : row_dict = { csv_imei_field_name : imei_norm } extra_field_to_values = { 'make' : '' , 'model' : '' , 'status' : '' , 'imsi' : '11111111111111' , 'msisdn' : '2322211122112' , 'reporting_date' : '20160420' , 'model_number' : '' , 'brand_name' : '' , 'device_type' : '' , 'radio_interface' : '' , 'device_id' : '123' } for e in extra_field_names : row_dict . update ( { e : extra_field_to_values [ e ] } ) if change_type : assert delta_import row_dict . update ( { 'change_type' : change_type } ) rows_to_write_list . append ( row_dict ) print ( rows_to_write_list ) writer = csv . DictWriter ( csvfile , fieldnames = csv_field_name_list ) writer . writeheader ( ) for d in rows_to_write_list : writer . writerow ( d ) valid_zip_import_data_file_path = str ( tmpdir . join ( 'sample_import_list.zip' ) ) with zipfile . ZipFile ( valid_zip_import_data_file_path , 'w' ) as valid_csv_import_data_file_zfile : valid_csv_import_data_file_zfile . write ( valid_csv_import_data_file , valid_csv_import_data_file_name ) return valid_zip_import_data_file_path
Helper function to create a csv input file.
Helper function to create a csv input file .
"""Helper function to create a csv input file.

    Takes in input the importer name to get metadata for csv header and a dict to map change_type to the imei_norm.
    i.e. csv_imei_change_type_tuples = [('52345678901234', 'remove'), ('22345678901234', 'add')];
         importer_name = pairing_list
    """ # generate metadata dict for the current importer # add a default value for extra field if there is any # generate csv header (header is a list of fieldnames) i.e. [imei, imsi, change_type] # generate a list of csv rows to be written # in full_import mode change_type is Null # write into csv
tmpdir
importer_name
csv_imei_change_type_tuples
delta_import