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
|
# coding=utf-8
from pgcli.packages.formatter.sqlformatter import escape_for_sql_statement
from cli_helpers.tabular_output import TabularOutputFormatter
from pgcli.packages.formatter.sqlformatter import adapter, register_new_formatter
def test_escape_for_sql_statement_bytes():
bts = b"837124ab3e8dc0f"
escaped_bytes = escape_for_sql_statement(bts)
assert escaped_bytes == "X'383337313234616233653864633066'"
def test_escape_for_sql_statement_number():
num = 2981
escaped_bytes = escape_for_sql_statement(num)
assert escaped_bytes == "'2981'"
def test_escape_for_sql_statement_str():
example_str = "example str"
escaped_bytes = escape_for_sql_statement(example_str)
assert escaped_bytes == "'example str'"
def test_output_sql_insert():
global formatter
formatter = TabularOutputFormatter
register_new_formatter(formatter)
data = [
[
1,
"Jackson",
"jackson_test@gmail.com",
"132454789",
None,
"2022-09-09 19:44:32.712343+08",
"2022-09-09 19:44:32.712343+08",
]
]
header = ["id", "name", "email", "phone", "description", "created_at", "updated_at"]
table_format = "sql-insert"
kwargs = {
"column_types": [int, str, str, str, str, str, str],
"sep_title": "RECORD {n}",
"sep_character": "-",
"sep_length": (1, 25),
"missing_value": "<null>",
"integer_format": "",
"float_format": "",
"disable_numparse": True,
"preserve_whitespace": True,
"max_field_width": 500,
}
formatter.query = 'SELECT * FROM "user";'
output = adapter(data, header, table_format=table_format, **kwargs)
output_list = [l for l in output]
expected = [
'INSERT INTO "user" ("id", "name", "email", "phone", "description", "created_at", "updated_at") VALUES',
" ('1', 'Jackson', 'jackson_test@gmail.com', '132454789', NULL, "
+ "'2022-09-09 19:44:32.712343+08', '2022-09-09 19:44:32.712343+08')",
";",
]
assert expected == output_list
def test_output_sql_update():
global formatter
formatter = TabularOutputFormatter
register_new_formatter(formatter)
data = [
[
1,
"Jackson",
"jackson_test@gmail.com",
"132454789",
"",
"2022-09-09 19:44:32.712343+08",
"2022-09-09 19:44:32.712343+08",
]
]
header = ["id", "name", "email", "phone", "description", "created_at", "updated_at"]
table_format = "sql-update"
kwargs = {
"column_types": [int, str, str, str, str, str, str],
"sep_title": "RECORD {n}",
"sep_character": "-",
"sep_length": (1, 25),
"missing_value": "<null>",
"integer_format": "",
"float_format": "",
"disable_numparse": True,
"preserve_whitespace": True,
"max_field_width": 500,
}
formatter.query = 'SELECT * FROM "user";'
output = adapter(data, header, table_format=table_format, **kwargs)
output_list = [l for l in output]
print(output_list)
expected = [
'UPDATE "user" SET',
" \"name\" = 'Jackson'",
", \"email\" = 'jackson_test@gmail.com'",
", \"phone\" = '132454789'",
", \"description\" = ''",
", \"created_at\" = '2022-09-09 19:44:32.712343+08'",
", \"updated_at\" = '2022-09-09 19:44:32.712343+08'",
"WHERE \"id\" = '1';",
]
assert expected == output_list
|
