-
Notifications
You must be signed in to change notification settings - Fork 28
Expand file tree
/
Copy pathtest_sqlbuilder.py
More file actions
134 lines (102 loc) · 4.89 KB
/
test_sqlbuilder.py
File metadata and controls
134 lines (102 loc) · 4.89 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
from sqlobject import IntCol, SQLObject, StringCol
from sqlobject.compat import PY2
from sqlobject.sqlbuilder import AND, ANY, CONCAT, Delete, Insert, \
SQLOp, Select, Union, Update, const, func, sqlrepr
from sqlobject.tests.dbtest import getConnection, raises, setupClass
class SOTestSQLBuilder(SQLObject):
name = StringCol()
so_value = IntCol()
def test_Select():
setupClass(SOTestSQLBuilder)
select1 = Select([const.id, func.MAX(const.salary)],
staticTables=['employees'])
assert sqlrepr(select1) == 'SELECT id, MAX(salary) FROM employees'
select2 = Select([SOTestSQLBuilder.q.name, SOTestSQLBuilder.q.so_value])
assert sqlrepr(select2) == \
'SELECT so_test_sql_builder.name, so_test_sql_builder.so_value ' \
'FROM so_test_sql_builder'
union = Union(select1, select2)
assert sqlrepr(union) == \
'SELECT id, MAX(salary) FROM employees ' \
'UNION SELECT so_test_sql_builder.name, ' \
'so_test_sql_builder.so_value ' \
'FROM so_test_sql_builder'
union = Union(SOTestSQLBuilder.select().queryForSelect())
assert sqlrepr(union) == \
'SELECT so_test_sql_builder.id, so_test_sql_builder.name, ' \
'so_test_sql_builder.so_value FROM so_test_sql_builder WHERE 1 = 1'
def test_empty_AND():
assert AND() is None
assert AND(True) is True
# sqlrepr() is needed because AND() returns an SQLExpression that overrides
# comparison. The following
# AND('x', 'y') == "foo bar"
# is True! (-: Eeek!
assert sqlrepr(AND(1, 2)) == sqlrepr(SQLOp("AND", 1, 2)) == "((1) AND (2))"
assert sqlrepr(AND(1, 2, '3'), "sqlite") == \
sqlrepr(SQLOp("AND", 1, SQLOp("AND", 2, '3')), "sqlite") == \
"((1) AND ((2) AND ('3')))"
def test_modulo():
setupClass(SOTestSQLBuilder)
assert sqlrepr(SOTestSQLBuilder.q.so_value % 2 == 0, 'mysql') == \
"((MOD(so_test_sql_builder.so_value, 2)) = (0))"
assert sqlrepr(SOTestSQLBuilder.q.so_value % 2 == 0, 'sqlite') == \
"(((so_test_sql_builder.so_value) % (2)) = (0))"
def test_division():
SOTestSQLBuilder(name='test', so_value=-11)
assert sqlrepr(SOTestSQLBuilder.q.so_value / 4 == -2.75, 'mysql') == \
"(((so_test_sql_builder.so_value) / (4)) = (-2.75))"
assert sqlrepr(SOTestSQLBuilder.q.so_value // 4 == -3, 'mysql') == \
"((FLOOR(((so_test_sql_builder.so_value) / (4)))) = (-3))"
def test_str_or_sqlrepr():
select = Select(['id', 'name'], staticTables=['employees'],
where='value>0', orderBy='id')
assert sqlrepr(select, 'sqlite') == \
'SELECT id, name FROM employees WHERE value>0 ORDER BY id'
select = Select(['id', 'name'], staticTables=['employees'],
where='value>0', orderBy='id', lazyColumns=True)
assert sqlrepr(select, 'sqlite') == \
'SELECT id FROM employees WHERE value>0 ORDER BY id'
insert = Insert('employees', values={'id': 1, 'name': 'test'})
assert sqlrepr(insert, 'sqlite') == \
"INSERT INTO employees (id, name) VALUES (1, 'test')"
update = Update('employees', {'name': 'test'}, where='id=1')
assert sqlrepr(update, 'sqlite') == \
"UPDATE employees SET name='test' WHERE id=1"
update = Update('employees', {'name': 'test', 'age': 42}, where='id=1')
assert sqlrepr(update, 'sqlite') == \
"UPDATE employees SET age=42, name='test' WHERE id=1"
delete = Delete('employees', where='id=1')
assert sqlrepr(delete, 'sqlite') == \
"DELETE FROM employees WHERE id=1"
raises(TypeError, Delete, 'employees')
delete = Delete('employees', where=None)
assert sqlrepr(delete, 'sqlite') == \
"DELETE FROM employees"
def test_CONCAT():
setupClass(SOTestSQLBuilder)
SOTestSQLBuilder(name='test', so_value=42)
assert sqlrepr(CONCAT('a', 'b'), 'mysql') == "CONCAT('a', 'b')"
assert sqlrepr(CONCAT('a', 'b'), 'mssql') == "'a' + 'b'"
assert sqlrepr(CONCAT('a', 'b'), 'sqlite') == "'a' || 'b'"
assert sqlrepr(CONCAT('prefix', SOTestSQLBuilder.q.name), 'mysql') == \
"CONCAT('prefix', so_test_sql_builder.name)"
assert sqlrepr(CONCAT('prefix', SOTestSQLBuilder.q.name), 'sqlite') == \
"'prefix' || so_test_sql_builder.name"
select = Select([CONCAT(SOTestSQLBuilder.q.name, '-suffix')],
staticTables=['so_test_sql_builder'])
connection = getConnection()
rows = connection.queryAll(connection.sqlrepr(select))
result = rows[0][0]
if not PY2 and not isinstance(result, str):
result = result.decode('ascii')
assert result == "test-suffix"
def test_ANY():
setupClass(SOTestSQLBuilder)
select = Select(
[SOTestSQLBuilder.q.name],
'value' == ANY(SOTestSQLBuilder.q.so_value),
)
assert sqlrepr(select, 'mysql') == \
"SELECT so_test_sql_builder.name FROM so_test_sql_builder " \
"WHERE (('value') = ANY (so_test_sql_builder.so_value))"