Testowanie sterownika ODBC Databricks (Simba)

Na tej stronie opisano sposób testowania kodu korzystającego ze sterownika ODBC usługi Databricks.

Użyj dowolnej platformy testowej dla języków zgodnych z ODBC. W poniższych przykładach użyto narzędzia pyodbc, pytest i unittest.mock do testowania połączeń sterowników ODBC. Ten kod jest oparty na przykładzie w temacie Connect Python and pyodbc to Azure Databricks (Łączenie języka Python i pyodbc z usługą Azure Databricks).

Funkcje pomocnicze

Plik helpers.py zawiera funkcje narzędzi do pracy z połączeniami ODBC:

• connect_to_dsn: otwiera połączenie z zasobem obliczeniowym usługi Azure Databricks.
• get_cursor_from_connection: uzyskuje kursor na potrzeby wykonywania zapytań.
• select_from_nyctaxi_trips: wykonuje zapytanie o określoną liczbę wierszy z samples.nyctaxi.trips.
• print_rows: wyświetla zawartość zestawu wyników w konsoli.

# 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)

Klasa główna

Plik main.py wywołuje funkcje pomocnika w celu nawiązania połączenia z danymi i wykonywania zapytań o nie:

# 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
)

Testy jednostkowe z wyśmiewaniem

Plik test_helpers.py używa pliku pytest i unittest.mock do testowania select_from_nyctaxi_trips funkcji. Pozorowanie symuluje połączenia z bazą danych bez używania rzeczywistych zasobów obliczeniowych, więc testy są uruchamiane w ciągu kilku sekund bez wpływu na obszary robocze usługi Azure Databricks.

# 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"

Ponieważ select_from_nyctaxi_trips uruchamia tylko instrukcję SELECT, pozorowanie nie jest tutaj koniecznie potrzebne. Jednak pozorowanie jest szczególnie przydatne podczas testowania funkcji modyfikujących dane (INSERT INTO, UPDATE, DELETE FROM), ponieważ można wielokrotnie uruchamiać testy bez wpływu na stan tabeli.