Pular para o conteúdo principal

Testando o driver ODBC da Databricks

Este artigo descreve como testar o código que usa o driverDatabricks ODBC.

Para testar o código que usa o driver Databricks ODBC junto com DSNs ou conexão sem DSN strings, o senhor pode usar estruturas de teste populares para linguagens de programação compatíveis com ODBC. Por exemplo, o exemplo de código Python a seguir usa pyodbc, pytest e unittest.mock para automatizar e testar o driver ODBC da Databricks usando um DSN. Este código de exemplo é baseado no código de exemplo em Connect Python and pyodbc to Databricks.

O arquivo de código de exemplo a seguir, denominado helpers.py, contém várias funções que automatizam o Databricks Driver usando um DSN:

  • A função connect_to_dsn usa um DSN para abrir uma conexão por meio de um recurso Databricks compute .
  • A função get_cursor_from_connection usa a conexão para obter um cursor , o que permite buscar operações nos dados por meio do recurso compute.
  • A função select_from_nyctaxi_trips usa o cursor para selecionar o número especificado de linhas de dados da tabela trips no esquema nyctaxi do catálogo samples.
  • A função print_rows imprime o conteúdo das linhas de dados na tela.
Python
# helpers.py

from pyodbc import connect, Connection, Cursor

def connect_to_dsn(
connstring: str,
autocommit: bool
) -> Connection:

connection = connect(
connstring,
autocommit = autocommit
)

return connection

def get_cursor_from_connection(
connection: Connection
) -> Cursor:

cursor = connection.cursor()
return cursor

def select_from_nyctaxi_trips(
cursor: Cursor,
num_rows: int
) -> Cursor:

select_cursor = cursor.execute(f"SELECT * FROM samples.nyctaxi.trips LIMIT {num_rows}")
return select_cursor

def print_rows(cursor: Cursor):
for row in cursor.fetchall():
print(row)

O exemplo de arquivo de código a seguir chamado main.py file chama as funções no arquivo helpers.py:

Python
# main.py

from helpers import *

connection = connect_to_dsn(
connstring = "DSN=<your-dsn-name>",
autocommit = True
)

cursor = get_cursor_from_connection(
connection = connection)

select_cursor = select_from_nyctaxi_trips(
cursor = cursor,
num_rows = 2
)

print_rows(
cursor = select_cursor
)

O exemplo de arquivo de código a seguir chamado test_helpers.py usa pytest para testar as funções no arquivo helpers.py. Em vez de usar o tempo e o custo do recurso compute real para chamar as funções no arquivo helpers.py, o código de exemplo a seguir usa unittest.mock para simular essas chamadas. Essas chamadas simuladas geralmente são concluídas em apenas alguns segundos, o que aumenta a confiança na qualidade do código e não altera o estado da conta ou do espaço de trabalho existente no site Databricks.

Python
# test_helpers.py

from pyodbc import SQL_DBMS_NAME
from helpers import *
from unittest.mock import patch
import datetime

@patch("helpers.connect_to_dsn")
def test_connect_to_dsn(mock_connection):
mock_connection.return_value.getinfo.return_value = "Spark SQL"

mock_connection = connect_to_dsn(
connstring = "DSN=<your-dsn-name>",
autocommit = True
)

assert mock_connection.getinfo(SQL_DBMS_NAME) == "Spark SQL"

@patch('helpers.get_cursor_from_connection')
def test_get_cursor_from_connection(mock_connection):
mock_cursor = mock_connection.return_value.cursor
mock_cursor.return_value.rowcount = -1

mock_connection = connect_to_dsn(
connstring = "DSN=<your-dsn-name>",
autocommit = True
)

mock_cursor = get_cursor_from_connection(
connection = mock_connection
)

assert mock_cursor.rowcount == -1

@patch('helpers.select_from_nyctaxi_trips')
def test_select_from_nyctaxi_trips(mock_connection):
mock_cursor = mock_connection.return_value.cursor
mock_get_cursor = mock_cursor.return_value.execute
mock_select_cursor = mock_get_cursor.return_value.arraysize = 1

mock_connection = connect_to_dsn(
connstring = "DSN=<your-dsn-name>",
autocommit = True
)

mock_get_cursor = get_cursor_from_connection(
connection = mock_connection
)

mock_select_cursor = select_from_nyctaxi_trips(
cursor = mock_get_cursor,
num_rows = 2
)

assert mock_select_cursor.arraysize == 1

@patch('helpers.print_rows')
def test_print_rows(mock_connection, capsys):
mock_cursor = mock_connection.return_value.cursor
mock_get_cursor = mock_cursor.return_value.execute
mock_select_cursor = mock_get_cursor.return_value.fetchall.return_value = [
(datetime.datetime(2016, 2, 14, 16, 52, 13), datetime.datetime(2016, 2, 14, 17, 16, 4), 4.94, 19.0, 10282, 10171),
(datetime.datetime(2016, 2, 4, 18, 44, 19), datetime.datetime(2016, 2, 4, 18, 46), 0.28, 3.5, 10110, 10110)
]

mock_connection = connect_to_dsn(
connstring = "DSN=<your-dsn-name>",
autocommit = True
)

mock_get_cursor = get_cursor_from_connection(
connection = mock_connection
)

mock_select_cursor = select_from_nyctaxi_trips(
cursor = mock_get_cursor,
num_rows = 2
)

print_rows(
cursor = mock_select_cursor
)

captured = capsys.readouterr()
assert captured.out == "(datetime.datetime(2016, 2, 14, 16, 52, 13), datetime.datetime(2016, 2, 14, 17, 16, 4), 4.94, 19.0, 10282, 10171)\n" \
"(datetime.datetime(2016, 2, 4, 18, 44, 19), datetime.datetime(2016, 2, 4, 18, 46), 0.28, 3.5, 10110, 10110)\n"

Como a função select_from_nyctaxi_trips contém uma instrução SELECT e, portanto, não altera o estado da tabela trips, a simulação não é absolutamente necessária neste exemplo. No entanto, o mocking permite que o senhor execute rapidamente seus testes sem esperar que uma conexão real seja feita com o recurso compute. Além disso, o mocking permite que o senhor execute testes simulados várias vezes para funções que possam alterar o estado de uma tabela, como INSERT INTO, UPDATE e DELETE FROM.