Skip to content

Commit a03602f

Browse files
co0lc0derco0lc0der
committed
add SQL unittests
1 parent bbe5ad9 commit a03602f

7 files changed

+548
-0
lines changed

‎tests/qb_delete_unittest.py

+34
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
import unittest
2+
from querybuilder import *
3+
4+
# test run
5+
# python .\qb_delete_unittest.py -v
6+
7+
8+
class QBDeleteTestCase(unittest.TestCase):
9+
def setUp(self):
10+
self.qb = QueryBuilder(DataBase(), ":memory:")
11+
12+
def test_sql_delete_eq(self):
13+
sql = self.qb.delete('comments').where([['user_id', '=', 10]]).get_sql()
14+
self.assertEqual(sql, "DELETE FROM `comments` WHERE (`user_id` = 10)")
15+
self.assertEqual(self.qb.get_params(), (10, ))
16+
17+
def test_sql_delete_no_eq(self):
18+
sql = self.qb.delete('comments').where([['user_id', 10]]).get_sql()
19+
self.assertEqual(sql, "DELETE FROM `comments` WHERE (`user_id` = 10)")
20+
self.assertEqual(self.qb.get_params(), (10, ))
21+
22+
# def test_sql_delete_limit_eq(self):
23+
# sql = self.qb.delete('users').where([['name', '=', 'John']]).limit().get_sql()
24+
# self.assertEqual(sql, "DELETE FROM `users` WHERE (`name` = 'John') LIMIT 1")
25+
# self.assertEqual(self.qb.get_params(), ('John',))
26+
#
27+
# def test_sql_delete_limit_no_eq(self):
28+
# sql = self.qb.delete('users').where([['name', 'John']]).limit().get_sql()
29+
# self.assertEqual(sql, "DELETE FROM `users` WHERE (`name` = 'John') LIMIT 1")
30+
# self.assertEqual(self.qb.get_params(), ('John',))
31+
32+
33+
if __name__ == "__main__":
34+
unittest.main()

‎tests/qb_insert_unittest.py

+28
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
import unittest
2+
from querybuilder import *
3+
4+
# test run
5+
# python .\qb_insert_unittest.py -v
6+
7+
8+
class QBInsertTestCase(unittest.TestCase):
9+
def setUp(self):
10+
self.qb = QueryBuilder(DataBase(), ":memory:")
11+
12+
def test_sql_insert(self):
13+
sql = self.qb.insert('groups', {'name': 'Moderator', 'permissions': 'moderator'}).get_sql()
14+
self.assertEqual(sql, "INSERT INTO `groups` (`name`, `permissions`) VALUES ('Moderator','moderator')")
15+
self.assertEqual(self.qb.get_params(), ('Moderator', 'moderator'))
16+
17+
def test_sql_insert_multiple(self):
18+
sql = self.qb.insert('groups', [
19+
['name', 'role'],
20+
['Moderator', 'moderator'], ['Moderator2', 'moderator'],
21+
['User', 'user'], ['User2', 'user']
22+
]).get_sql()
23+
self.assertEqual(sql, "INSERT INTO `groups` (`name`, `role`) VALUES ('Moderator','moderator'),('Moderator2','moderator'),('User','user'),('User2','user')")
24+
self.assertEqual(self.qb.get_params(), ('Moderator', 'moderator', 'Moderator2', 'moderator', 'User', 'user', 'User2', 'user'))
25+
26+
27+
if __name__ == "__main__":
28+
unittest.main()

‎tests/qb_select_ext_unittest.py

+185
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
import unittest
2+
from querybuilder import *
3+
4+
# test run
5+
# python .\qb_select_ext_unittest.py -v
6+
7+
8+
class QBSelectExtensionsTestCase(unittest.TestCase):
9+
def setUp(self):
10+
self.qb = QueryBuilder(DataBase(), ":memory:")
11+
12+
def test_sql_select_inner_join_list(self):
13+
sql = self.qb.select({'u': 'users'}, ['u.id', 'u.email', 'u.username', {'perms': 'groups.permissions'}])\
14+
.join('groups', ['u.group_id', 'groups.id']).get_sql()
15+
self.assertEqual(sql, "SELECT `u`.`id`, `u`.`email`, `u`.`username`, `groups`.`permissions` AS `perms` FROM `users` AS `u` INNER JOIN `groups` ON `u`.`group_id` = `groups`.`id`")
16+
self.assertEqual(self.qb.get_params(), ())
17+
18+
def test_sql_select_inner_join3_str(self):
19+
sql = self.qb.select({'cp': 'cabs_printers'}, [
20+
'cp.id', 'cp.cab_id', {'cab_name': 'cb.name'}, 'cp.printer_id',
21+
{'printer_name': 'p.name'}, {'cartridge_type': 'c.name'}, 'cp.comment'
22+
])\
23+
.join({'cb': 'cabs'}, ['cp.cab_id', 'cb.id'])\
24+
.join({'p': 'printer_models'}, ['cp.printer_id', 'p.id'])\
25+
.join({'c': 'cartridge_types'}, 'p.cartridge_id=c.id')\
26+
.where([['cp.cab_id', 'in', [11, 12, 13]], 'or', ['cp.cab_id', 5], 'and', ['p.id', '>', 'c.id']]).get_sql()
27+
self.assertEqual(sql, "SELECT `cp`.`id`, `cp`.`cab_id`, `cb`.`name` AS `cab_name`, `cp`.`printer_id`, `p`.`name` AS `printer_name`, `c`.`name` AS `cartridge_type`, `cp`.`comment` FROM `cabs_printers` AS `cp` INNER JOIN `cabs` AS `cb` ON `cp`.`cab_id` = `cb`.`id` INNER JOIN `printer_models` AS `p` ON `cp`.`printer_id` = `p`.`id` INNER JOIN `cartridge_types` AS `c` ON p.cartridge_id=c.id WHERE (`cp`.`cab_id` IN (11,12,13)) OR (`cp`.`cab_id` = 5) AND (`p`.`id` > 'c.id')")
28+
self.assertEqual(self.qb.get_params(), (11, 12, 13, 5, 'c.id'))
29+
30+
def test_sql_select_inner_join3_groupby_orederby(self):
31+
sql = self.qb.select({'cp': 'cabs_printers'}, [
32+
'cp.id', 'cp.cab_id', {'cab_name': 'cb.name'},
33+
'cp.printer_id', {'cartridge_id': 'c.id'},
34+
{'printer_name': 'p.name'}, {'cartridge_type': 'c.name'}, 'cp.comment'
35+
])\
36+
.join({'cb': 'cabs'}, ['cp.cab_id', 'cb.id'])\
37+
.join({'p': 'printer_models'}, ['cp.printer_id', 'p.id'])\
38+
.join({'c': 'cartridge_types'}, ['p.cartridge_id', 'c.id'])\
39+
.group_by(['cp.printer_id', 'cartridge_id'])\
40+
.order_by(['cp.cab_id', 'cp.printer_id desc']).get_sql()
41+
self.assertEqual(sql, "SELECT `cp`.`id`, `cp`.`cab_id`, `cb`.`name` AS `cab_name`, `cp`.`printer_id`, `c`.`id` AS `cartridge_id`, `p`.`name` AS `printer_name`, `c`.`name` AS `cartridge_type`, `cp`.`comment` FROM `cabs_printers` AS `cp` INNER JOIN `cabs` AS `cb` ON `cp`.`cab_id` = `cb`.`id` INNER JOIN `printer_models` AS `p` ON `cp`.`printer_id` = `p`.`id` INNER JOIN `cartridge_types` AS `c` ON `p`.`cartridge_id` = `c`.`id` GROUP BY `cp`.`printer_id`, `cartridge_id` ORDER BY `cp`.`cab_id` ASC, `cp`.`printer_id` DESC")
42+
self.assertEqual(self.qb.get_params(), ())
43+
44+
def test_sql_select_left_join(self):
45+
sql = self.qb.select('employees', ['employees.employee_id', 'employees.last_name', 'positions.title'])\
46+
.join('positions', ['employees.position_id', 'positions.position_id'], join_type="left").get_sql()
47+
self.assertEqual(sql, "SELECT `employees`.`employee_id`, `employees`.`last_name`, `positions`.`title` FROM `employees` LEFT JOIN `positions` ON `employees`.`position_id` = `positions`.`position_id`")
48+
self.assertEqual(self.qb.get_params(), ())
49+
50+
def test_sql_select_left_outer_join(self):
51+
sql = self.qb.select({'e': 'employees'}, ['e.employee_id', 'e.last_name', 'p.title'])\
52+
.join({'p': 'positions'}, ['e.position_id', 'p.position_id'], join_type="left outer").get_sql()
53+
self.assertEqual(sql, "SELECT `e`.`employee_id`, `e`.`last_name`, `p`.`title` FROM `employees` AS `e` LEFT OUTER JOIN `positions` AS `p` ON `e`.`position_id` = `p`.`position_id`")
54+
self.assertEqual(self.qb.get_params(), ())
55+
56+
def test_sql_select_cross_join(self):
57+
sql = self.qb.select('positions').join('departments', join_type="cross").get_sql()
58+
self.assertEqual(sql, "SELECT * FROM `positions` CROSS JOIN `departments`")
59+
self.assertEqual(self.qb.get_params(), ())
60+
61+
def test_sql_except_select(self):
62+
sql = self.qb.select('departments', ['department_id']).except_select('employees').get_sql()
63+
self.assertEqual(sql, "SELECT `department_id` FROM `departments` EXCEPT SELECT `department_id` FROM `employees`")
64+
self.assertEqual(self.qb.get_params(), ())
65+
66+
def test_sql_select_excepts_where(self):
67+
sql = self.qb.select('contacts', ['contact_id', 'last_name', 'first_name']).where([['contact_id', '>=', 74]])\
68+
.excepts()\
69+
.select('employees', ['employee_id', 'last_name', 'first_name']).where([['first_name', 'Sandra']])\
70+
.get_sql()
71+
self.assertEqual(sql, "SELECT `contact_id`, `last_name`, `first_name` FROM `contacts` WHERE (`contact_id` >= 74) EXCEPT SELECT `employee_id`, `last_name`, `first_name` FROM `employees` WHERE (`first_name` = 'Sandra')")
72+
self.assertEqual(self.qb.get_params(), (74, 'Sandra'))
73+
74+
def test_sql_select_excepts_where_order_by(self):
75+
sql = self.qb.select('suppliers', ['supplier_id', 'state']).where([['state', 'Nevada']]) \
76+
.excepts()\
77+
.select('companies', ['company_id', 'state']).where([['company_id', '<', 2000]]).order_by('1 desc')\
78+
.get_sql()
79+
self.assertEqual(sql, "SELECT `supplier_id`, `state` FROM `suppliers` WHERE (`state` = 'Nevada') EXCEPT SELECT `company_id`, `state` FROM `companies` WHERE (`company_id` < 2000) ORDER BY `1` DESC")
80+
self.assertEqual(self.qb.get_params(), ('Nevada', 2000))
81+
82+
def test_sql_intersect_select(self):
83+
sql = self.qb.select('departments', ['department_id']).intersect_select('employees').get_sql()
84+
self.assertEqual(sql, "SELECT `department_id` FROM `departments` INTERSECT SELECT `department_id` FROM `employees`")
85+
self.assertEqual(self.qb.get_params(), ())
86+
87+
def test_sql_select_intersect_where(self):
88+
sql = self.qb.select('departments', 'department_id').where([['department_id', '>=', 25]])\
89+
.intersect()\
90+
.select('employees', 'department_id').where([['last_name', '<>', 'Petrov']])\
91+
.get_sql()
92+
self.assertEqual(sql, "SELECT `department_id` FROM `departments` WHERE (`department_id` >= 25) INTERSECT SELECT `department_id` FROM `employees` WHERE (`last_name` <> 'Petrov')")
93+
self.assertEqual(self.qb.get_params(), (25, 'Petrov'))
94+
95+
def test_sql_select_intersect_where2(self):
96+
sql = self.qb.select('contacts', ['contact_id', 'last_name', 'first_name']).where([['contact_id', '>', 50]])\
97+
.intersect()\
98+
.select('customers', ['customer_id', 'last_name', 'first_name']).where([['last_name', '<>', 'Zagoskin']])\
99+
.get_sql()
100+
self.assertEqual(sql, "SELECT `contact_id`, `last_name`, `first_name` FROM `contacts` WHERE (`contact_id` > 50) INTERSECT SELECT `customer_id`, `last_name`, `first_name` FROM `customers` WHERE (`last_name` <> 'Zagoskin')")
101+
self.assertEqual(self.qb.get_params(), (50, 'Zagoskin'))
102+
103+
def test_sql_select_intersect_where_orderby(self):
104+
sql = self.qb.select('departments', ['department_id', 'state']).where([['department_id', '>=', 25]]) \
105+
.intersect()\
106+
.select('companies', ['company_id', 'state']).like('company_name', 'G%').order_by('1')\
107+
.get_sql()
108+
self.assertEqual(sql, "SELECT `department_id`, `state` FROM `departments` WHERE (`department_id` >= 25) INTERSECT SELECT `company_id`, `state` FROM `companies` WHERE (`company_name` LIKE 'G%') ORDER BY `1` ASC")
109+
self.assertEqual(self.qb.get_params(), (25, 'G%'))
110+
111+
def test_sql_select_union_where(self):
112+
sql = self.qb.select('clients', ['name', 'age', {'total_sum': 'account_sum + account_sum * 0.1'}])\
113+
.where([['account_sum', '<', 3000]])\
114+
.union()\
115+
.select('clients', ['name', 'age', {'total_sum': 'account_sum + account_sum * 0.3'}])\
116+
.where([['account_sum', '>=', 3000]]).get_sql()
117+
self.assertEqual(sql, "SELECT `name`, `age`, account_sum + account_sum * 0.1 AS `total_sum` FROM `clients` WHERE (`account_sum` < 3000) UNION SELECT `name`, `age`, account_sum + account_sum * 0.3 AS `total_sum` FROM `clients` WHERE (`account_sum` >= 3000)")
118+
self.assertEqual(self.qb.get_params(), (3000, 3000))
119+
120+
def test_sql_union_select_where(self):
121+
sql = self.qb.select('clients', ['name', 'age']).where([['id', '<', 10]])\
122+
.union_select('employees').where([['id', 1]]).get_sql()
123+
self.assertEqual(sql, "SELECT `name`, `age` FROM `clients` WHERE (`id` < 10) UNION SELECT `name`, `age` FROM `employees` WHERE (`id` = 1)")
124+
self.assertEqual(self.qb.get_params(), (10, 1))
125+
126+
def test_sql_select_union_where_orderby(self):
127+
sql = self.qb.select('departments', ['department_id', 'department_name']).where([['department_id', 'in', [1, 2]]])\
128+
.union()\
129+
.select('employees', ['employee_id', 'last_name']).where([['hire_date', '2024-02-08']]).order_by('2')\
130+
.get_sql()
131+
self.assertEqual(sql, "SELECT `department_id`, `department_name` FROM `departments` WHERE (`department_id` IN (1,2)) UNION SELECT `employee_id`, `last_name` FROM `employees` WHERE (`hire_date` = '2024-02-08') ORDER BY `2` ASC")
132+
self.assertEqual(self.qb.get_params(), (1, 2, '2024-02-08'))
133+
134+
def test_sql_select_union_all(self):
135+
sql = self.qb.select('clients', ['name', 'age', {'total_sum': 'account_sum + account_sum * 0.1'}])\
136+
.where([['account_sum', '<', 3000]])\
137+
.union(True)\
138+
.select('clients', ['name', 'age', {'total_sum': 'account_sum + account_sum * 0.3'}])\
139+
.where([['account_sum', '>=', 3000]]).get_sql()
140+
self.assertEqual(sql, "SELECT `name`, `age`, account_sum + account_sum * 0.1 AS `total_sum` FROM `clients` WHERE (`account_sum` < 3000) UNION ALL SELECT `name`, `age`, account_sum + account_sum * 0.3 AS `total_sum` FROM `clients` WHERE (`account_sum` >= 3000)")
141+
self.assertEqual(self.qb.get_params(), (3000, 3000))
142+
143+
def test_sql_union_select_all_where(self):
144+
sql = self.qb.select('cabs', ['id', 'name']) \
145+
.union_select('printer_models', True).where([['id', '<', 10]]) \
146+
.get_sql()
147+
self.assertEqual(sql, "SELECT `id`, `name` FROM `cabs` UNION ALL SELECT `id`, `name` FROM `printer_models` WHERE (`id` < 10)")
148+
self.assertEqual(self.qb.get_params(), (10, ))
149+
150+
def test_sql_select_union_all_where_orderby(self):
151+
sql = self.qb.select('departments', ['department_id', 'department_name']).where([['department_id', '>=', 10]])\
152+
.union(True)\
153+
.select('employees', ['employee_id', 'last_name']).where([['last_name', 'Rassohin']]).order_by('2')\
154+
.get_sql()
155+
self.assertEqual(sql, "SELECT `department_id`, `department_name` FROM `departments` WHERE (`department_id` >= 10) UNION ALL SELECT `employee_id`, `last_name` FROM `employees` WHERE (`last_name` = 'Rassohin') ORDER BY `2` ASC")
156+
self.assertEqual(self.qb.get_params(), (10, 'Rassohin'))
157+
158+
# def test_sql_select_in_select_str_one_param(self):
159+
# sql = self.qb.select(self.qb.select('categories').get_sql(), ['id', 'name']).where([['id', '<=', 5]]).get_sql()
160+
# self.assertEqual(sql, "SELECT `id`, `name` FROM (SELECT * FROM `categories`) WHERE (`id` <= 5)")
161+
# self.assertEqual(self.qb.get_params(), (5, ))
162+
#
163+
# def test_sql_select_in_select_str_two_params(self):
164+
# sql = self.qb.select(self.qb.select('categories').where([['parent_id', 0]]).get_sql(), ['id', 'name']).where([['id', '<=', 5]]).get_sql()
165+
# self.assertEqual(sql, "SELECT `id`, `name` FROM (SELECT * FROM `categories` WHERE (`parent_id` = 0)) WHERE (`id` <= 5)")
166+
# self.assertEqual(self.qb.get_params(), (0, 5))
167+
#
168+
# def test_sql_select_in_select_add_query(self):
169+
# q1 = self.qb.select('categories').where([['parent_id', 0]]).get_sql()
170+
# sql = self.qb.select(q1, ['id', 'name']).where([['id', '<=', 5]]).get_sql()
171+
# self.assertEqual(sql, "SELECT `id`, `name` FROM (SELECT * FROM `categories` WHERE (`parent_id` = 0)) WHERE (`id` <= 5)")
172+
# self.assertEqual(self.qb.get_params(), (0, 5))
173+
#
174+
# def test_sql_select_in_select_add_query_format(self):
175+
# q1 = self.qb.select('categories').where([['parent_id', 0]])
176+
# sql = self.qb.select(f'{q1}', ['id', 'name']).where([['id', '<=', 5]]).get_sql()
177+
# self.assertEqual(sql, "SELECT `id`, `name` FROM (SELECT * FROM `categories` WHERE (`parent_id` = 0)) WHERE (`id` <= 5)")
178+
# self.assertEqual(self.qb.get_params(), (0, 5))
179+
180+
def tearDown(self):
181+
pass
182+
183+
184+
if __name__ == "__main__":
185+
unittest.main()

0 commit comments

Comments
 (0)