tests/_delta_helpers.py
def delta_list_import_common(db_conn, mocked_config, tmpdir, importer_name, historic_tbl_name, logger):
    """Common code to test delta list import.

    1)  Verify import failure without using delta option because delta file pre-validation fails due to change_type col
    2)  Verify import succeeds using delta option
    2a) Verify that all record added have end_date None
    2b) Verify that record with imei_norm '52345678901234' has been removed and end_date set to not None.
    """
    imei_one = '12345678901234'
    imei_two = '22345678901234'
    imei_five = '52345678901234'
    imei_norm_pk_name = 'imei_norm'
    if importer_name == 'golden_list':
        imei_one = imeis_md5_hashing_uuid(imei_one, convert_to_uuid=True)
        imei_two = imeis_md5_hashing_uuid(imei_two, convert_to_uuid=True)
        imei_five = imeis_md5_hashing_uuid(imei_five, convert_to_uuid=True)
        imei_norm_pk_name = 'hashed_imei_norm'
    # populate historic_tbl
    imei_norm_to_insert_list = [imei_five]
    # write into csv
    csv_imei_change_type_tuples = [('52345678901234', 'remove'), ('12345678901234', 'add'),
                                   ('22345678901234', 'add')]
    # populate table
    historic_table_insert_params_from_dict(db_conn, importer_name, imei_norm_to_insert_list)
    # write csv
    valid_zip_import_data_file_path = write_import_csv(tmpdir, importer_name, csv_imei_change_type_tuples)
    runner = CliRunner()
    # Test part 1) Verify import failure without delta option
    result = runner.invoke(dirbs_import_cli, [importer_name, valid_zip_import_data_file_path],
                           obj={'APP_CONFIG': mocked_config})
    assert result.exit_code == 1
    assert "Pre-validation failed: b\'Error:   " \
           'Metadata header, cannot find the column headers - change_type' in logger_stream_contents(logger)

    # Test part 2) Verify import success using delta option
    result = runner.invoke(dirbs_import_cli, [importer_name, '--delta', '--disable-delta-adds-check',
                                              valid_zip_import_data_file_path], obj={'APP_CONFIG': mocked_config})
    assert result.exit_code == 0
    count, res = fetch_tbl_rows(historic_tbl_name, db_conn)
    assert count == 3
    expected_output = {imei_one,
                       imei_two,
                       imei_five}
    imei_norm_set = {getattr(r, imei_norm_pk_name) for r in res}
    assert imei_norm_set == expected_output
    # Test part 2a
    assert all([(r.end_date is None) for r in res if getattr(r, imei_norm_pk_name) not in imei_five])
    # Test part 2b
    assert all([(r.end_date is not None) for r in res if getattr(r, imei_norm_pk_name) in imei_five])
Common code to test delta list import.

    1)  Verify import failure without using delta option because delta file pre-validation fails due to change_type col
    2)  Verify import succeeds using delta option
    2a) Verify that all record added have end_date None
    2b) Verify that record with imei_norm '52345678901234' has been removed and end_date set to not None.
    
Common code to test delta list import.
1)  Verify import failure without using delta option because delta file pre-validation fails due to change_type col
2)  Verify import succeeds using delta option
2a) Verify that all record added have end_date None
2b) Verify that record with imei_norm '52345678901234' has been removed and end_date set to not None.
Common code to test delta list import . 1 ) Verify import failure without using delta option because delta file pre - validation fails due to change_type col 2 ) Verify import succeeds using delta option 2a ) Verify that all record added have end_date None 2b ) Verify that record with imei_norm ' 52345678901234 ' has been removed and end_date set to not None .
def delta_list_import_common(db_conn, mocked_config, tmpdir, importer_name, historic_tbl_name, logger):
    imei_one = '12345678901234'
    imei_two = '22345678901234'
    imei_five = '52345678901234'
    imei_norm_pk_name = 'imei_norm'
    if importer_name == 'golden_list':
        imei_one = imeis_md5_hashing_uuid(imei_one, convert_to_uuid=True)
        imei_two = imeis_md5_hashing_uuid(imei_two, convert_to_uuid=True)
        imei_five = imeis_md5_hashing_uuid(imei_five, convert_to_uuid=True)
        imei_norm_pk_name = 'hashed_imei_norm'
    imei_norm_to_insert_list = [imei_five]
    csv_imei_change_type_tuples = [('52345678901234', 'remove'), ('12345678901234', 'add'),
                                   ('22345678901234', 'add')]
    historic_table_insert_params_from_dict(db_conn, importer_name, imei_norm_to_insert_list)
    valid_zip_import_data_file_path = write_import_csv(tmpdir, importer_name, csv_imei_change_type_tuples)
    runner = CliRunner()
    result = runner.invoke(dirbs_import_cli, [importer_name, valid_zip_import_data_file_path],
                           obj={'APP_CONFIG': mocked_config})
    assert result.exit_code == 1
    assert "Pre-validation failed: b\'Error:   " \
           'Metadata header, cannot find the column headers - change_type' in logger_stream_contents(logger)
    result = runner.invoke(dirbs_import_cli, [importer_name, '--delta', '--disable-delta-adds-check',
                                              valid_zip_import_data_file_path], obj={'APP_CONFIG': mocked_config})
    assert result.exit_code == 0
    count, res = fetch_tbl_rows(historic_tbl_name, db_conn)
    assert count == 3
    expected_output = {imei_one,
                       imei_two,
                       imei_five}
    imei_norm_set = {getattr(r, imei_norm_pk_name) for r in res}
    assert imei_norm_set == expected_output
    assert all([(r.end_date is None) for r in res if getattr(r, imei_norm_pk_name) not in imei_five])
    assert all([(r.end_date is not None) for r in res if getattr(r, imei_norm_pk_name) in imei_five])
def delta_list_import_common ( db_conn , mocked_config , tmpdir , importer_name , historic_tbl_name , logger ) : imei_one = '12345678901234' imei_two = '22345678901234' imei_five = '52345678901234' imei_norm_pk_name = 'imei_norm' if importer_name == 'golden_list' : imei_one = imeis_md5_hashing_uuid ( imei_one , convert_to_uuid = True ) imei_two = imeis_md5_hashing_uuid ( imei_two , convert_to_uuid = True ) imei_five = imeis_md5_hashing_uuid ( imei_five , convert_to_uuid = True ) imei_norm_pk_name = 'hashed_imei_norm' imei_norm_to_insert_list = [ imei_five ] csv_imei_change_type_tuples = [ ( '52345678901234' , 'remove' ) , ( '12345678901234' , 'add' ) , ( '22345678901234' , 'add' ) ] historic_table_insert_params_from_dict ( db_conn , importer_name , imei_norm_to_insert_list ) valid_zip_import_data_file_path = write_import_csv ( tmpdir , importer_name , csv_imei_change_type_tuples ) runner = CliRunner ( ) result = runner . invoke ( dirbs_import_cli , [ importer_name , valid_zip_import_data_file_path ] , obj = { 'APP_CONFIG' : mocked_config } ) assert result . exit_code == 1 assert "Pre-validation failed: b\'Error:   " 'Metadata header, cannot find the column headers - change_type' in logger_stream_contents ( logger ) result = runner . invoke ( dirbs_import_cli , [ importer_name , '--delta' , '--disable-delta-adds-check' , valid_zip_import_data_file_path ] , obj = { 'APP_CONFIG' : mocked_config } ) assert result . exit_code == 0 count , res = fetch_tbl_rows ( historic_tbl_name , db_conn ) assert count == 3 expected_output = { imei_one , imei_two , imei_five } imei_norm_set = { getattr ( r , imei_norm_pk_name ) for r in res } assert imei_norm_set == expected_output assert all ( [ ( r . end_date is None ) for r in res if getattr ( r , imei_norm_pk_name ) not in imei_five ] ) assert all ( [ ( r . end_date is not None ) for r in res if getattr ( r , imei_norm_pk_name ) in imei_five ] )
Common code to test delta list import.
Common code to test delta list import .
"""Common code to test delta list import.

    1)  Verify import failure without using delta option because delta file pre-validation fails due to change_type col
    2)  Verify import succeeds using delta option
    2a) Verify that all record added have end_date None
    2b) Verify that record with imei_norm '52345678901234' has been removed and end_date set to not None.
    """ # populate historic_tbl # write into csv # populate table # write csv # Test part 1) Verify import failure without delta option # Test part 2) Verify import success using delta option # Test part 2a # Test part 2b
db_conn
mocked_config
tmpdir
importer_name
historic_tbl_name
logger