1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158
|
from __future__ import annotations
import sys
from datetime import date, datetime
import pytest
from harlequin.adapter import HarlequinAdapter, HarlequinConnection, HarlequinCursor
from harlequin.catalog import Catalog, CatalogItem
from harlequin.exception import HarlequinConnectionError, HarlequinQueryError
from harlequin_postgres.adapter import (
HarlequinPostgresAdapter,
HarlequinPostgresConnection,
)
from textual_fastdatatable.backend import create_backend
if sys.version_info < (3, 10):
from importlib_metadata import entry_points
else:
from importlib.metadata import entry_points
TEST_DB_CONN = "postgresql://postgres:for-testing@localhost:5432"
def test_plugin_discovery() -> None:
PLUGIN_NAME = "postgres"
eps = entry_points(group="harlequin.adapter")
assert eps[PLUGIN_NAME]
adapter_cls = eps[PLUGIN_NAME].load()
assert issubclass(adapter_cls, HarlequinAdapter)
assert adapter_cls == HarlequinPostgresAdapter
def test_connect() -> None:
conn = HarlequinPostgresAdapter(conn_str=(TEST_DB_CONN,)).connect()
assert isinstance(conn, HarlequinConnection)
def test_init_extra_kwargs() -> None:
assert HarlequinPostgresAdapter(
conn_str=(TEST_DB_CONN,), foo=1, bar="baz"
).connect()
@pytest.mark.parametrize(
"conn_str",
[
("foo",),
("host=foo",),
("postgresql://admin:pass@foo:5432/db",),
],
)
def test_connect_raises_connection_error(conn_str: tuple[str]) -> None:
with pytest.raises(HarlequinConnectionError):
_ = HarlequinPostgresAdapter(conn_str=conn_str, connect_timeout=0.1).connect()
@pytest.mark.parametrize(
"conn_str,options,expected",
[
(("",), {}, "localhost:5432/postgres"),
(("host=foo",), {}, "foo:5432/postgres"),
(("postgresql://foo",), {}, "foo:5432/postgres"),
(("postgresql://foo",), {"port": 5431}, "foo:5431/postgres"),
(("postgresql://foo/mydb",), {"port": 5431}, "foo:5431/mydb"),
(("postgresql://admin:pass@foo/mydb",), {"port": 5431}, "foo:5431/mydb"),
(("postgresql://admin:pass@foo:5431/mydb",), {}, "foo:5431/mydb"),
],
)
def test_connection_id(
conn_str: tuple[str], options: dict[str, int | float | str | None], expected: str
) -> None:
adapter = HarlequinPostgresAdapter(
conn_str=conn_str,
**options, # type: ignore[arg-type]
)
assert adapter.connection_id == expected
def test_get_catalog(connection: HarlequinPostgresConnection) -> None:
catalog = connection.get_catalog()
assert isinstance(catalog, Catalog)
assert catalog.items
assert isinstance(catalog.items[0], CatalogItem)
def test_get_completions(connection: HarlequinPostgresConnection) -> None:
completions = connection.get_completions()
test_labels = ["atomic", "greatest", "point_right", "autovacuum"]
filtered = list(filter(lambda x: x.label in test_labels, completions))
assert len(filtered) == 4
value_filtered = list(filter(lambda x: x.value in test_labels, completions))
assert len(value_filtered) == 4
def test_execute_ddl(connection: HarlequinPostgresConnection) -> None:
cur = connection.execute("create table foo (a int)")
assert cur is None
def test_execute_select(connection: HarlequinPostgresConnection) -> None:
cur = connection.execute("select 1 as a")
assert isinstance(cur, HarlequinCursor)
assert cur.columns() == [("a", "#")]
data = cur.fetchall()
backend = create_backend(data)
assert backend.column_count == 1
assert backend.row_count == 1
def test_execute_select_dupe_cols(connection: HarlequinPostgresConnection) -> None:
cur = connection.execute("select 1 as a, 2 as a, 3 as a")
assert isinstance(cur, HarlequinCursor)
assert len(cur.columns()) == 3
data = cur.fetchall()
backend = create_backend(data)
assert backend.column_count == 3
assert backend.row_count == 1
def test_set_limit(connection: HarlequinPostgresConnection) -> None:
cur = connection.execute("select 1 as a union all select 2 union all select 3")
assert isinstance(cur, HarlequinCursor)
cur = cur.set_limit(2)
assert isinstance(cur, HarlequinCursor)
data = cur.fetchall()
backend = create_backend(data)
assert backend.column_count == 1
assert backend.row_count == 2
def test_execute_raises_query_error(connection: HarlequinPostgresConnection) -> None:
with pytest.raises(HarlequinQueryError):
_ = connection.execute("sel;")
def test_inf_timestamps(connection: HarlequinPostgresConnection) -> None:
cur = connection.execute(
"""select
'infinity'::date,
'infinity'::timestamp,
'infinity'::timestamptz,
'-infinity'::date,
'-infinity'::timestamp,
'-infinity'::timestamptz
"""
)
assert cur is not None
data = cur.fetchall()
assert data == [
(
date.max,
datetime.max,
datetime.max,
date.min,
datetime.min,
datetime.min,
)
]
|