tests/_helpers.py
def expect_success(importer, row_count, conn, logger, log_message=None):
    """Helper function to test if data has been imported succesfully."""
    with conn, conn.cursor() as cursor:
        importer.import_data()
        cursor.execute(sql.SQL('SELECT COUNT(*) FROM {0}').format(sql.Identifier(importer._import_relation_name)))
        result = cursor.fetchone()
        assert result[0] == row_count

    if log_message is not None:
        assert (log_message in logger_stream_contents(logger))
Helper function to test if data has been imported succesfully.
Helper function to test if data has been imported succesfully.
Helper function to test if data has been imported succesfully .
def expect_success(importer, row_count, conn, logger, log_message=None):
    with conn, conn.cursor() as cursor:
        importer.import_data()
        cursor.execute(sql.SQL('SELECT COUNT(*) FROM {0}').format(sql.Identifier(importer._import_relation_name)))
        result = cursor.fetchone()
        assert result[0] == row_count
    if log_message is not None:
        assert (log_message in logger_stream_contents(logger))
def expect_success ( importer , row_count , conn , logger , log_message = None ) : with conn , conn . cursor ( ) as cursor : importer . import_data ( ) cursor . execute ( sql . SQL ( 'SELECT COUNT(*) FROM {0}' ) . format ( sql . Identifier ( importer . _import_relation_name ) ) ) result = cursor . fetchone ( ) assert result [ 0 ] == row_count if log_message is not None : assert ( log_message in logger_stream_contents ( logger ) )
Helper function to test if data has been imported succesfully.
Helper function to test if data has been imported succesfully .
"""Helper function to test if data has been imported succesfully."""
importer
row_count
conn
logger
log_message