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
|
import pytest
from dbutils import (create_db, db_connection, setup_db, teardown_db, TEST_DB_NAME)
from pgspecial.main import PGSpecial
@pytest.yield_fixture(scope='module')
def connection():
create_db(TEST_DB_NAME)
connection = db_connection(TEST_DB_NAME)
setup_db(connection)
yield connection
teardown_db(connection)
connection.close()
@pytest.fixture
def cursor(connection):
with connection.cursor() as cur:
return cur
@pytest.fixture
def executor(connection):
cur = connection.cursor()
pgspecial = PGSpecial()
def query_runner(sql):
results = []
for title, rows, headers, status in pgspecial.execute(cur=cur, sql=sql):
if rows:
results.extend((title, list(rows), headers, status))
else:
results.extend((title, None, headers, status))
return results
return query_runner
|
