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
|
from __future__ import annotations
import sys
from typing import Awaitable, Callable
import pytest
from textual.message import Message
from harlequin import Harlequin
@pytest.mark.asyncio
async def test_run_query_bar(
app_all_adapters_small_db: Harlequin,
app_snapshot: Callable[..., Awaitable[bool]],
wait_for_workers: Callable[[Harlequin], Awaitable[None]],
transaction_button_visible: Callable[[Harlequin], bool],
) -> None:
app = app_all_adapters_small_db
snap_results: list[bool] = []
messages: list[Message] = []
async with app.run_test(size=(120, 36), message_hook=messages.append) as pilot:
await wait_for_workers(app)
while app.editor is None:
await pilot.pause()
# initialization
bar = app.run_query_bar
assert bar.limit_checkbox.value is False
assert bar.limit_input.value == "500"
assert bar.limit_value is None
# query without any limit by clicking the button;
# dataset has 857 records
assert app.editor is not None
app.editor.text = "select * from drivers"
await pilot.click("#run_query")
await wait_for_workers(app)
await pilot.pause()
await wait_for_workers(app)
await pilot.pause()
await wait_for_workers(app)
await pilot.pause()
await pilot.wait_for_scheduled_animations()
table = app.results_viewer.get_visible_table()
assert table
assert table.row_count == table.source_row_count == 857
snap_results.append(await app_snapshot(app, "No limit"))
# apply a limit by clicking the limit checkbox
await pilot.click(bar.limit_checkbox.__class__)
assert bar.limit_checkbox.value is True
assert bar.limit_value == 500
await pilot.click("#run_query")
await wait_for_workers(app)
await pilot.pause()
await wait_for_workers(app)
await pilot.pause()
await wait_for_workers(app)
await pilot.pause()
await pilot.wait_for_scheduled_animations()
table = app.results_viewer.get_visible_table()
assert table
assert table.row_count == table.source_row_count == 500
snap_results.append(await app_snapshot(app, "Limit 500"))
# type an invalid limit, checkbox should be unchecked
# and a tooltip should appear on hover
await pilot.click(bar.limit_input.__class__)
await pilot.press("a")
assert bar.limit_input.value == "a500"
assert bar.limit_value is None
assert bar.limit_checkbox.value is False
assert bar.limit_input.tooltip is not None
await pilot.pause()
await pilot.wait_for_scheduled_animations()
snap_results.append(await app_snapshot(app, "Invalid limit"))
# type a valid limit
await pilot.press("backspace")
await pilot.press("delete")
await pilot.press("1")
assert bar.limit_input.value == "100"
assert bar.limit_value == 100
assert bar.limit_checkbox.value is True
assert bar.limit_input.tooltip is None
# run the query with a smaller limit
await pilot.click("#run_query")
await wait_for_workers(app)
await pilot.pause()
await wait_for_workers(app)
await pilot.pause()
await wait_for_workers(app)
await pilot.pause()
await pilot.wait_for_scheduled_animations()
table = app.results_viewer.get_visible_table()
assert table
assert table.row_count == table.source_row_count == 100
snap_results.append(await app_snapshot(app, "Limit 100"))
if not transaction_button_visible(app):
assert all(snap_results)
@pytest.mark.py12
@pytest.mark.skipif(
sys.version_info < (3, 12), reason="SQLite in Python < 3.12 won't show txn button"
)
@pytest.mark.asyncio
async def test_transaction_button(
app_small_sqlite: Harlequin,
app_snapshot: Callable[..., Awaitable[bool]],
wait_for_workers: Callable[[Harlequin], Awaitable[None]],
) -> None:
app = app_small_sqlite
snap_results: list[bool] = []
async with app.run_test(size=(120, 36)) as pilot:
await wait_for_workers(app)
while app.editor is None or app.connection is None:
await pilot.pause()
assert app.connection.transaction_mode
assert app.connection.transaction_mode.label == "Auto"
snap_results.append(await app_snapshot(app, "Initialize with Tx: Auto"))
await pilot.click("#transaction_button")
await pilot.pause(0.3)
await pilot.wait_for_animation()
assert app.connection.transaction_mode
assert app.connection.transaction_mode.label == "Manual"
assert app.connection.transaction_mode.commit is not None
assert app.connection.transaction_mode.rollback is not None
snap_results.append(await app_snapshot(app, "After click with Tx: Manual"))
assert all(snap_results)
|
