Freigeben über

Testen des Databricks ODBC-Treibers (Simba)

Auf dieser Seite wird beschrieben, wie Sie Code testen, der den ODBC-Treiber "Databricks" verwendet.

Verwenden Sie ein beliebiges Testframework für ODBC-kompatible Sprachen. In den folgenden Beispielen wird pyodbc, pytest und unittest.mock verwendet, um ODBC-Treiberverbindungen zu testen. Dieser Code basiert auf dem Beispiel in Connect Python und Pyodbc mit Azure Databricks.

Hilfsfunktionen

Die helpers.py Datei enthält Hilfsfunktionen für das Arbeiten mit ODBC-Verbindungen:

  • connect_to_dsn: Öffnet eine Verbindung mit einer Azure Databricks-Computeressource.
  • get_cursor_from_connection: Ruft einen Cursor zum Ausführen von Abfragen ab.
  • select_from_nyctaxi_trips: Fragt aus samples.nyctaxi.trips die angegebene Anzahl von Zeilen ab.
  • print_rows: Druckt den Inhalt des Resultsets in der Konsole.
# 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)

Main-Klasse

Die main.py Datei ruft die Hilfsfunktionen auf, um Daten zu verbinden und abzufragen:

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

Komponententests mit Mocking

Die test_helpers.py Datei verwendet pytest und unittest.mock, um die select_from_nyctaxi_trips Funktion zu testen. Mocking simuliert Datenbankverbindungen ohne Verwendung tatsächlicher Rechenressourcen, sodass Tests in Sekunden ausgeführt werden, ohne dass sich dies auf Ihre Azure Databricks-Arbeitsbereiche auswirkt.

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

Da select_from_nyctaxi_trips nur eine SELECT Anweisung ausführt, ist Mocking hier nicht unbedingt erforderlich. Das Mocking ist jedoch besonders hilfreich beim Testen von Funktionen, die Daten ändern (INSERT INTO, UPDATE, DELETE FROM), da Sie Tests wiederholt ausführen können, ohne den Tabellenzustand zu beeinträchtigen.

  • Last updated on