Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_commit_on_closed_connection_raises(dsn, configuration):
connection = connect(dsn, **get_credentials(configuration))
connection.close()
with pytest.raises(InterfaceError):
connection.commit()
def test_fetchmany_with_bad_global_arraysize_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'SELECT MULTIPLE INTEGERS') as query:
cursor.execute(query)
cursor.arraysize = -1
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchmany()
cursor.arraysize = 0
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchmany()
def test_column_with_non_contiguous_data_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
two_dimensional = array([[1, 2, 3], [4, 5, 6]], dtype='int64')
one_dimensional = two_dimensional[:, 1]
assert one_dimensional.flags.c_contiguous == False
columns = [one_dimensional]
with pytest.raises(turbodbc.InterfaceError):
cursor.executemanycolumns("INSERT INTO {} VALUES (?)".format(table_name), columns)
def test_closed_cursor_raises_when_used(dsn, configuration):
connection = connect(dsn, **get_credentials(configuration))
cursor = connection.cursor()
cursor.close()
with pytest.raises(InterfaceError):
cursor.execute("SELECT 42")
with pytest.raises(InterfaceError):
cursor.executemany("SELECT 42")
with pytest.raises(InterfaceError):
cursor.executemanycolumns("SELECT 42", [])
with pytest.raises(InterfaceError):
cursor.fetchone()
with pytest.raises(InterfaceError):
cursor.fetchmany()
with pytest.raises(InterfaceError):
cursor.fetchall()
def test_pep343_with_statement(dsn, configuration):
with connect(dsn, **get_credentials(configuration)) as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT 42")
# cursor should be closed
with pytest.raises(InterfaceError):
cursor.execute("SELECT 42")
def test_fetchmany_with_bad_global_arraysize_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'SELECT MULTIPLE INTEGERS') as query:
cursor.execute(query)
cursor.arraysize = -1
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchmany()
cursor.arraysize = 0
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchmany()
def test_column_of_unsupported_type_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with query_fixture(cursor, configuration, 'INSERT INTEGER') as table_name:
columns = ["this is not a NumPy MaskedArray"]
with pytest.raises(turbodbc.InterfaceError):
cursor.executemanycolumns("INSERT INTO {} VALUES (?)".format(table_name), columns)
def test_numpy_without_result_set_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchallnumpy()
def test_arrow_without_result_set_raises(dsn, configuration):
with open_cursor(configuration) as cursor:
with pytest.raises(turbodbc.InterfaceError):
cursor.fetchallarrow()
def test_cursor_on_closed_connection_raises(dsn, configuration):
connection = connect(dsn, **get_credentials(configuration))
connection.close()
with pytest.raises(InterfaceError):
connection.cursor()