tests/_delta_helpers.py
def row_count_stats_common(postgres, db_conn, tmpdir, mocked_config, logger, importer_name, historic_tbl_name):
    """Test Depot not available yet. Verify output stats for CLI import command."""
    # Part 1) populate import_table and verify before import row count
    # Part 2) import file containing one duplicate (same row) and verify that staging_row_count value includes
    # duplicates and import_table_new_row_count doesn't.
    imei_norm_one = '12345678901243'
    imei_norm_two = '22345678901243'
    imei_norm_three = '32345678901234'
    imei_norm_four = '42345678901243'
    imei_norm_five = '52345678901243'
    imei_norm_pk = 'imei_norm'
    if importer_name == 'golden_list':
        imei_norm_one = imeis_md5_hashing_uuid(imei_norm_one, convert_to_uuid=True)
        imei_norm_two = imeis_md5_hashing_uuid(imei_norm_two, convert_to_uuid=True)
        imei_norm_three = imeis_md5_hashing_uuid(imei_norm_three, convert_to_uuid=True)
        imei_norm_four = imeis_md5_hashing_uuid(imei_norm_four, convert_to_uuid=True)
        imei_norm_five = imeis_md5_hashing_uuid(imei_norm_five, convert_to_uuid=True)
        imei_norm_pk = 'hashed_imei_norm'
    # populate historic_tbl
    imei_norm_to_insert_list = [imei_norm_one, imei_norm_three]
    # write into csv
    csv_imei_change_type_tuples = [('12345678901243', 'remove'),
                                   ('22345678901243', 'add'),
                                   ('22345678901243', 'add'),
                                   ('42345678901243', 'add'),
                                   ('52345678901243', 'add')]
    # Test part 1)
    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, '--delta', 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 == 5
    imei_norm_set = {getattr(r, imei_norm_pk) for r in res}
    assert imei_norm_set == {imei_norm_one, imei_norm_two, imei_norm_three, imei_norm_four, imei_norm_five}
    # Test  Part 1)
    assert 'Rows in table prior to import: 2' in logger_stream_contents(logger)
    # Test  Part 2) - self.staging_row_count
    assert 'Rows supplied in delta input file: 5' in logger_stream_contents(logger)
    # Test  Part 2) - import_table_new_row_count=rows_before + rows_inserted - rows_deleted
    assert 'Rows in table after import: 4 (3 new, 0 updated, 1 removed)' in logger_stream_contents(logger)
Test Depot not available yet. Verify output stats for CLI import command.
Test Depot not available yet. Verify output stats for CLI import command.
Test Depot not available yet . Verify output stats for CLI import command .
def row_count_stats_common(postgres, db_conn, tmpdir, mocked_config, logger, importer_name, historic_tbl_name):
    imei_norm_one = '12345678901243'
    imei_norm_two = '22345678901243'
    imei_norm_three = '32345678901234'
    imei_norm_four = '42345678901243'
    imei_norm_five = '52345678901243'
    imei_norm_pk = 'imei_norm'
    if importer_name == 'golden_list':
        imei_norm_one = imeis_md5_hashing_uuid(imei_norm_one, convert_to_uuid=True)
        imei_norm_two = imeis_md5_hashing_uuid(imei_norm_two, convert_to_uuid=True)
        imei_norm_three = imeis_md5_hashing_uuid(imei_norm_three, convert_to_uuid=True)
        imei_norm_four = imeis_md5_hashing_uuid(imei_norm_four, convert_to_uuid=True)
        imei_norm_five = imeis_md5_hashing_uuid(imei_norm_five, convert_to_uuid=True)
        imei_norm_pk = 'hashed_imei_norm'
    imei_norm_to_insert_list = [imei_norm_one, imei_norm_three]
    csv_imei_change_type_tuples = [('12345678901243', 'remove'),
                                   ('22345678901243', 'add'),
                                   ('22345678901243', 'add'),
                                   ('42345678901243', 'add'),
                                   ('52345678901243', '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, '--delta', 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 == 5
    imei_norm_set = {getattr(r, imei_norm_pk) for r in res}
    assert imei_norm_set == {imei_norm_one, imei_norm_two, imei_norm_three, imei_norm_four, imei_norm_five}
    assert 'Rows in table prior to import: 2' in logger_stream_contents(logger)
    assert 'Rows supplied in delta input file: 5' in logger_stream_contents(logger)
    assert 'Rows in table after import: 4 (3 new, 0 updated, 1 removed)' in logger_stream_contents(logger)
def row_count_stats_common ( postgres , db_conn , tmpdir , mocked_config , logger , importer_name , historic_tbl_name ) : imei_norm_one = '12345678901243' imei_norm_two = '22345678901243' imei_norm_three = '32345678901234' imei_norm_four = '42345678901243' imei_norm_five = '52345678901243' imei_norm_pk = 'imei_norm' if importer_name == 'golden_list' : imei_norm_one = imeis_md5_hashing_uuid ( imei_norm_one , convert_to_uuid = True ) imei_norm_two = imeis_md5_hashing_uuid ( imei_norm_two , convert_to_uuid = True ) imei_norm_three = imeis_md5_hashing_uuid ( imei_norm_three , convert_to_uuid = True ) imei_norm_four = imeis_md5_hashing_uuid ( imei_norm_four , convert_to_uuid = True ) imei_norm_five = imeis_md5_hashing_uuid ( imei_norm_five , convert_to_uuid = True ) imei_norm_pk = 'hashed_imei_norm' imei_norm_to_insert_list = [ imei_norm_one , imei_norm_three ] csv_imei_change_type_tuples = [ ( '12345678901243' , 'remove' ) , ( '22345678901243' , 'add' ) , ( '22345678901243' , 'add' ) , ( '42345678901243' , 'add' ) , ( '52345678901243' , '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 , '--delta' , 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 == 5 imei_norm_set = { getattr ( r , imei_norm_pk ) for r in res } assert imei_norm_set == { imei_norm_one , imei_norm_two , imei_norm_three , imei_norm_four , imei_norm_five } assert 'Rows in table prior to import: 2' in logger_stream_contents ( logger ) assert 'Rows supplied in delta input file: 5' in logger_stream_contents ( logger ) assert 'Rows in table after import: 4 (3 new, 0 updated, 1 removed)' in logger_stream_contents ( logger )
Test Depot not available yet.
Test Depot not available yet .
"""Test Depot not available yet. Verify output stats for CLI import command.""" # Part 1) populate import_table and verify before import row count # Part 2) import file containing one duplicate (same row) and verify that staging_row_count value includes # duplicates and import_table_new_row_count doesn't. # populate historic_tbl # write into csv # Test part 1) # Test  Part 1) # Test  Part 2) - self.staging_row_count # Test  Part 2) - import_table_new_row_count=rows_before + rows_inserted - rows_deleted
postgres
db_conn
tmpdir
mocked_config
logger
importer_name
historic_tbl_name