forked from catherinedevlin/ipython-sql
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_parse.py
More file actions
180 lines (123 loc) · 4.94 KB
/
Copy pathtest_parse.py
File metadata and controls
180 lines (123 loc) · 4.94 KB
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
import json
import os
from pathlib import Path
from six.moves import configparser
from sql.parse import connection_from_dsn_section, parse, without_sql_comment
try:
from traitlets.config.configurable import Configurable
except ImportError:
from IPython.config.configurable import Configurable
empty_config = Configurable()
default_connect_args = {"options": "-csearch_path=test"}
def test_parse_no_sql():
assert parse("will:longliveliz@localhost/shakes", empty_config) == {
"connection": "will:longliveliz@localhost/shakes",
"sql": "",
"result_var": None,
}
def test_parse_with_sql():
assert parse(
"postgresql://will:longliveliz@localhost/shakes SELECT * FROM work",
empty_config,
) == {
"connection": "postgresql://will:longliveliz@localhost/shakes",
"sql": "SELECT * FROM work",
"result_var": None,
}
def test_parse_sql_only():
assert parse("SELECT * FROM work", empty_config) == {
"connection": "",
"sql": "SELECT * FROM work",
"result_var": None,
}
def test_parse_postgresql_socket_connection():
assert parse("postgresql:///shakes SELECT * FROM work", empty_config) == {
"connection": "postgresql:///shakes",
"sql": "SELECT * FROM work",
"result_var": None,
}
def test_expand_environment_variables_in_connection():
os.environ["DATABASE_URL"] = "postgresql:///shakes"
assert parse("$DATABASE_URL SELECT * FROM work", empty_config) == {
"connection": "postgresql:///shakes",
"sql": "SELECT * FROM work",
"result_var": None,
}
def test_parse_shovel_operator():
assert parse("dest << SELECT * FROM work", empty_config) == {
"connection": "",
"sql": "SELECT * FROM work",
"result_var": "dest",
}
def test_parse_connect_plus_shovel():
assert parse("sqlite:// dest << SELECT * FROM work", empty_config) == {
"connection": "sqlite://",
"sql": "SELECT * FROM work",
"result_var": None,
}
def test_parse_shovel_operator():
assert parse("dest << SELECT * FROM work", empty_config) == {
"connection": "",
"sql": "SELECT * FROM work",
"result_var": "dest",
}
def test_parse_connect_plus_shovel():
assert parse("sqlite:// dest << SELECT * FROM work", empty_config) == {
"connection": "sqlite://",
"sql": "SELECT * FROM work",
"result_var": "dest",
}
class DummyConfig:
dsn_filename = Path("src/tests/test_dsn_config.ini")
def test_connection_from_dsn_section():
result = connection_from_dsn_section(section="DB_CONFIG_1", config=DummyConfig())
result = connection_from_dsn_section(section="DB_CONFIG_2", config=DummyConfig())
class Bunch:
def __init__(self, **kwds):
self.__dict__.update(kwds)
class ParserStub:
opstrs = [
[],
["-l", "--connections"],
["-x", "--close"],
["-c", "--creator"],
["-s", "--section"],
["-p", "--persist"],
["--append"],
["-a", "--connection_arguments"],
["-f", "--file"],
]
_actions = [Bunch(option_strings=o) for o in opstrs]
parser_stub = ParserStub()
def test_without_sql_comment_plain():
line = "SELECT * FROM author"
assert without_sql_comment(parser=parser_stub, line=line) == line
def test_without_sql_comment_with_arg():
line = "--file moo.txt --persist SELECT * FROM author"
assert without_sql_comment(parser=parser_stub, line=line) == line
def test_without_sql_comment_with_comment():
line = "SELECT * FROM author -- uff da"
expected = "SELECT * FROM author"
assert without_sql_comment(parser=parser_stub, line=line) == expected
def test_without_sql_comment_with_arg_and_comment():
line = "--file moo.txt --persist SELECT * FROM author -- uff da"
expected = "--file moo.txt --persist SELECT * FROM author"
assert without_sql_comment(parser=parser_stub, line=line) == expected
def test_without_sql_comment_unspaced_comment():
line = "SELECT * FROM author --uff da"
expected = "SELECT * FROM author"
assert without_sql_comment(parser=parser_stub, line=line) == expected
def test_without_sql_comment_dashes_in_string():
line = "SELECT '--very --confusing' FROM author -- uff da"
expected = "SELECT '--very --confusing' FROM author"
assert without_sql_comment(parser=parser_stub, line=line) == expected
def test_without_sql_comment_with_arg_and_leading_comment():
line = "--file moo.txt --persist --comment, not arg"
expected = "--file moo.txt --persist"
assert without_sql_comment(parser=parser_stub, line=line) == expected
def test_without_sql_persist():
line = "--persist my_table --uff da"
expected = "--persist my_table"
assert without_sql_comment(parser=parser_stub, line=line) == expected