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,
)
]
|
