Databricks ODBC Sürücüsünü Test Edin

Bu sayfada Databricks ODBC Sürücüsünü kullanan kodu test etme adımları açıklanmaktadır.

ODBC uyumlu diller için herhangi bir test çerçevesi kullanın. Aşağıdaki örneklerde ODBC sürücü bağlantılarını test etmek için pyodbc, pytest ve unittest.mock kullanılır. Bu kod, Python ve pyodbc'yi Azure Databricks'e bağlama sayfasındaki örneği temel alır.

Yardımcı işlevler

Dosya ODBC helpers.py bağlantılarıyla çalışmaya yönelik yardımcı program işlevlerini içerir:

  • connect_to_dsn: Azure Databricks işlem kaynağına bir bağlantı açar.
  • get_cursor_from_connection: Sorguları yürütmek için bir imleç alır.
  • select_from_nyctaxi_trips: içinden samples.nyctaxi.tripsbelirtilen satır sayısını sorgular.
  • print_rows: Sonuç kümesinin içeriğini konsola yazdırır.
# 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 sınıf

Dosya, main.py bağlanmak ve verileri sorgulamak için yardımcı işlevleri çağırır:

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

Birim testleri taklit (mocking) ile

Dosya, test_helpers.py işlevini test etmek için pytest ve unittest.mock kullanır. Mocking, gerçek hesaplama kaynaklarını kullanmadan veritabanı bağlantılarını simüle eder, böylece testler Azure Databricks çalışma alanlarınızı etkilemeden saniyeler içinde çalışır.

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

Yalnızca bir select_from_nyctaxi_trips deyimi çalıştırdığı için SELECT, burada mocking kesinlikle gerekli değildir. Ancak, tablo durumunu etkilemeden testleri tekrar tekrar çalıştırabileceğinizden, sahte işlem özellikle verileri (INSERT INTO, UPDATE, DELETE FROM) değiştiren işlevleri test ederken yararlıdır.

  • Last updated on