100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterator
|
|
|
|
import pyodbc
|
|
import pytest
|
|
|
|
from harlequin_odbc.adapter import HarlequinOdbcConnection
|
|
from harlequin_odbc.catalog import (
|
|
DatabaseCatalogItem,
|
|
SchemaCatalogItem,
|
|
TableCatalogItem,
|
|
ViewCatalogItem,
|
|
)
|
|
|
|
|
|
class FakeCursor:
|
|
def __init__(self) -> None:
|
|
self.executed: list[tuple[str, tuple[object, ...]]] = []
|
|
self.rows: list[tuple[str, ...]] = []
|
|
|
|
def execute(self, sql: str, *params: object) -> FakeCursor:
|
|
self.executed.append((sql, params))
|
|
normalized_sql = sql.lower()
|
|
if "sys_context" in normalized_sql:
|
|
self.rows = [("ORCL",)]
|
|
elif "all_tab_columns" in normalized_sql:
|
|
self.rows = [("EMPLOYEE_ID", "NUMBER(6,0)"), ("FIRST_NAME", "VARCHAR2(20)")]
|
|
elif "all_tables" in normalized_sql and "all_views" in normalized_sql:
|
|
self.rows = [("HR", "EMPLOYEES", "TABLE"), ("HR", "EMPLOYEE_VIEW", "VIEW")]
|
|
else:
|
|
self.rows = []
|
|
return self
|
|
|
|
def fetchval(self) -> str | None:
|
|
if not self.rows:
|
|
return None
|
|
return self.rows[0][0]
|
|
|
|
def tables(self, *_args: object, **_kwargs: object) -> object:
|
|
raise AssertionError("Oracle catalog should not call SQLTables")
|
|
|
|
def columns(self, *_args: object, **_kwargs: object) -> object:
|
|
raise AssertionError("Oracle catalog should not call SQLColumns")
|
|
|
|
def __iter__(self) -> Iterator[tuple[str, ...]]:
|
|
return iter(self.rows)
|
|
|
|
|
|
class FakeConnection:
|
|
def __init__(self) -> None:
|
|
self.cursor_instance = FakeCursor()
|
|
|
|
def cursor(self) -> FakeCursor:
|
|
return self.cursor_instance
|
|
|
|
def getinfo(self, info_type: int) -> str:
|
|
if info_type == pyodbc.SQL_DBMS_NAME:
|
|
return "Oracle"
|
|
if info_type == pyodbc.SQL_DATABASE_NAME:
|
|
return "ORCL"
|
|
return ""
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
|
|
def test_oracle_catalog_uses_data_dictionary(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_connection = FakeConnection()
|
|
|
|
def connect(*_args: object, **_kwargs: object) -> FakeConnection:
|
|
return fake_connection
|
|
|
|
monkeypatch.setattr(pyodbc, "connect", connect)
|
|
connection = HarlequinOdbcConnection(("Driver={Oracle};",))
|
|
|
|
catalog = connection.get_catalog()
|
|
|
|
[database_item] = catalog.items
|
|
assert isinstance(database_item, DatabaseCatalogItem)
|
|
assert database_item.label == "ORCL"
|
|
[schema_item] = database_item.children
|
|
assert isinstance(schema_item, SchemaCatalogItem)
|
|
assert schema_item.label == "HR"
|
|
|
|
[table_item, view_item] = schema_item.children
|
|
assert isinstance(table_item, TableCatalogItem)
|
|
assert table_item.label == "EMPLOYEES"
|
|
assert table_item.query_name == '"HR"."EMPLOYEES"'
|
|
assert isinstance(view_item, ViewCatalogItem)
|
|
assert view_item.label == "EMPLOYEE_VIEW"
|
|
|
|
column_items = table_item.fetch_children()
|
|
assert [(item.label, item.type_label) for item in column_items] == [
|
|
("EMPLOYEE_ID", "NUMBER(6,0)"),
|
|
("FIRST_NAME", "VARCHAR2(20)"),
|
|
]
|
|
assert fake_connection.cursor_instance.executed[-1][1] == ("HR", "EMPLOYEES")
|