tests/_fixtures.py
def db_conn(postgres, mocked_config, request):
    """Fixture to inject a DB connection into a fixture. Cleans up to make sure DB is clean after each test."""
    # Create db connection
    current_db_user = mocked_config.db_config.user
    mocked_config.db_config.user = request.param
    conn = create_db_connection(mocked_config.db_config)
    yield conn
    # Close connection and create new one as the db role might have changed for security tests
    conn.close()
    mocked_config.db_config.user = current_db_user
    conn = create_db_connection(mocked_config.db_config)
    with conn.cursor() as table_cursor, conn.cursor() as truncate_cursor:
        table_cursor.execute('SELECT tablename FROM pg_tables WHERE schemaname = current_schema() '
                             "AND tablename != \'schema_metadata\' AND tablename != \'radio_access_technology_map\'")
        for tblname in table_cursor:
            truncate_cursor.execute(sql.SQL('TRUNCATE {0} CASCADE').format(sql.Identifier(tblname[0])))

        table_cursor.execute('SELECT sequence_name FROM information_schema.sequences '
                             'WHERE sequence_schema = current_schema()')
        for seqname in table_cursor:
            truncate_cursor.execute(sql.SQL('ALTER SEQUENCE {0} RESTART WITH 1').format(sql.Identifier(seqname[0])))

        table_cursor.execute('SELECT matviewname FROM pg_matviews WHERE schemaname = current_schema()')
        for matviewname in table_cursor:
            truncate_cursor.execute(sql.SQL('REFRESH MATERIALIZED VIEW CONCURRENTLY {0}')
                                    .format(sql.Identifier(matviewname[0])))
    # Commit truncations
    conn.commit()
    conn.close()
Fixture to inject a DB connection into a fixture. Cleans up to make sure DB is clean after each test.
Fixture to inject a DB connection into a fixture. Cleans up to make sure DB is clean after each test.
Fixture to inject a DB connection into a fixture . Cleans up to make sure DB is clean after each test .
def db_conn(postgres, mocked_config, request):
    current_db_user = mocked_config.db_config.user
    mocked_config.db_config.user = request.param
    conn = create_db_connection(mocked_config.db_config)
    yield conn
    conn.close()
    mocked_config.db_config.user = current_db_user
    conn = create_db_connection(mocked_config.db_config)
    with conn.cursor() as table_cursor, conn.cursor() as truncate_cursor:
        table_cursor.execute('SELECT tablename FROM pg_tables WHERE schemaname = current_schema() '
                             "AND tablename != \'schema_metadata\' AND tablename != \'radio_access_technology_map\'")
        for tblname in table_cursor:
            truncate_cursor.execute(sql.SQL('TRUNCATE {0} CASCADE').format(sql.Identifier(tblname[0])))
        table_cursor.execute('SELECT sequence_name FROM information_schema.sequences '
                             'WHERE sequence_schema = current_schema()')
        for seqname in table_cursor:
            truncate_cursor.execute(sql.SQL('ALTER SEQUENCE {0} RESTART WITH 1').format(sql.Identifier(seqname[0])))
        table_cursor.execute('SELECT matviewname FROM pg_matviews WHERE schemaname = current_schema()')
        for matviewname in table_cursor:
            truncate_cursor.execute(sql.SQL('REFRESH MATERIALIZED VIEW CONCURRENTLY {0}')
                                    .format(sql.Identifier(matviewname[0])))
    conn.commit()
    conn.close()
def db_conn ( postgres , mocked_config , request ) : current_db_user = mocked_config . db_config . user mocked_config . db_config . user = request . param conn = create_db_connection ( mocked_config . db_config ) yield conn conn . close ( ) mocked_config . db_config . user = current_db_user conn = create_db_connection ( mocked_config . db_config ) with conn . cursor ( ) as table_cursor , conn . cursor ( ) as truncate_cursor : table_cursor . execute ( 'SELECT tablename FROM pg_tables WHERE schemaname = current_schema() ' "AND tablename != \'schema_metadata\' AND tablename != \'radio_access_technology_map\'" ) for tblname in table_cursor : truncate_cursor . execute ( sql . SQL ( 'TRUNCATE {0} CASCADE' ) . format ( sql . Identifier ( tblname [ 0 ] ) ) ) table_cursor . execute ( 'SELECT sequence_name FROM information_schema.sequences ' 'WHERE sequence_schema = current_schema()' ) for seqname in table_cursor : truncate_cursor . execute ( sql . SQL ( 'ALTER SEQUENCE {0} RESTART WITH 1' ) . format ( sql . Identifier ( seqname [ 0 ] ) ) ) table_cursor . execute ( 'SELECT matviewname FROM pg_matviews WHERE schemaname = current_schema()' ) for matviewname in table_cursor : truncate_cursor . execute ( sql . SQL ( 'REFRESH MATERIALIZED VIEW CONCURRENTLY {0}' ) . format ( sql . Identifier ( matviewname [ 0 ] ) ) ) conn . commit ( ) conn . close ( )
Fixture to inject a DB connection into a fixture.
Fixture to inject a DB connection into a fixture .
"""Fixture to inject a DB connection into a fixture. Cleans up to make sure DB is clean after each test.""" # Create db connection # Close connection and create new one as the db role might have changed for security tests # Commit truncations
postgres
mocked_config
request