/
test_database_access.py
363 lines (337 loc) · 12.6 KB
/
test_database_access.py
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
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
# coding: utf-8
# Copyright (c) Max-Planck-Institut für Eisenforschung GmbH - Computational Materials Design (CM) Department
# Distributed under the terms of "New BSD License", see the LICENSE file.
"""
Normally, test-based development is done in reversed order unlike this DatabaseClass.
First the Unittest should be created and then the wanted class which should pass all the tests.
Because of that, this Unittest is based on the DatabaseClass making the executed test cases a bit strange
and specific extra for this class.
Murat Han Celik
"""
import unittest
import os
from datetime import datetime
from random import choice
from string import ascii_uppercase
from pyiron_base.database.generic import DatabaseAccess
from pyiron_base._tests import PyironTestCase
from sqlalchemy import text
class TestDatabaseAccess(PyironTestCase):
"""
Standard Unittest of the DatabaseAccess class
"""
@classmethod
def setUpClass(cls):
"""
Set up whole class for testing
Returns:
"""
# we assume everything working on sqlite, should also work on postgres in sqlalchemy
cls.database = DatabaseAccess("sqlite:///test_database.db", "simulation")
@classmethod
def tearDownClass(cls):
"""
Tear down whole class after testing
Returns:
"""
cls.database.conn.close()
if os.name != 'nt':
# On windows we get PermissionError: [WinError 32] The process cannot access the
# file because it is being used by another process: 'test_database.db'
os.remove("test_database.db")
def tearDown(self):
"""
Deletes all entries after every tested function
Returns:
"""
self.database.conn.execute(text("delete from simulation"))
def test_get_table_headings(self):
"""
Tests get_table_headings
Returns:
"""
heading_list = [
"id",
"parentid",
"masterid",
"projectpath",
"project",
"job",
"subjob",
"chemicalformula",
"status",
"hamilton",
"hamversion",
"username",
"computer",
"timestart",
"timestop",
"totalcputime",
]
# general headings have to be at least a part of get_table_headings
for item in heading_list:
self.assertTrue(item in self.database.get_table_headings())
def test_get_items_sql(self):
"""
Tests get_items_sql function
Returns:
"""
self.add_items("Blub")
self.add_items("Bluk")
# has to return a list
self.assertIsInstance(
self.database.get_items_sql("chemicalformula LIKE 'Blu%'"), list
)
self.assertRaises(
Exception, self.database.get_items_sql, "A Wrong where Clause"
) # Where clause must be right
# A valid sqlstatement should also return a valid list
self.assertIsInstance(
self.database.get_items_sql(
where_condition="", sql_statement="select * from simulation"
),
list,
)
par_dict = self.add_items("BO")
key = par_dict["id"]
# be sure that get_items_sql returns right result with right statement
result = self.database.get_items_sql(
where_condition="",
sql_statement="select * from simulation " "where id=%s" % key,
)[-1]
self.assertTrue(par_dict.items() <= result.items())
def test_get_items_sql_like_regex(self):
"""
Tests the regex functionality of 'like'
Returns:
"""
elem1 = self.add_items("H4Ni2")
elem2 = self.add_items("H2")
elem3 = self.add_items("H6")
elem4 = self.add_items("HeNi2")
elem5 = self.add_items("H12Ni5")
elem6 = self.add_items("H12")
elem7 = self.add_items("He2")
# H([0-9]*) matches H2, H6 and H12
self.assertEqual(
[elem2, elem3, elem6],
self.database.get_items_sql(r"chemicalformula like 'H([0-9]*)'"),
)
# He(\d)*(Ni)?\d* matches HeNi2, He2
self.assertEqual(
[elem4, elem7],
self.database.get_items_sql(r"chemicalformula like 'He(\d)*(Ni)?\d*'"),
)
# H\d*Ni\d* matches H4Ni2, H12Ni5
self.assertEqual(
[elem1, elem5],
self.database.get_items_sql(r"chemicalformula like 'H\d*Ni\d*'"),
)
# assert that not something random really is in the Database, recommended by Samuel Hartke
# Murat: 'Just ignore the line!'
self.assertEqual(
[], self.database.get_items_sql(r"chemicalformula like 'B\d[a-z]'")
)
def test_add_item_dict(self):
"""
Tests add_item_dict function
Returns:
"""
par_dict = self.add_items("BO")
key = par_dict["id"]
# list as parameter shall not work
self.assertRaises(
Exception, self.database.add_item_dict, [{"chemicalformula": "BO"}]
)
self.assertIsInstance(key, int) # returned value must be int
# added and got dict must be(almost) the same
result = self.database.get_item_by_id(key)
self.assertTrue(par_dict.items() <= result.items())
def test_item_update(self):
"""
Tests item_update function
Returns:
"""
par_dict = self.add_items("BO")
key = par_dict["id"]
# function does only accept a dict, no list
self.assertRaises(
Exception, self.database.item_update, [{"job": "testing2"}], key
)
try: # Function works with int, str and list, normally I would test against list, but then our project
# would not work anymore.
self.database.item_update({"job": "testing2"}, key)
self.database.item_update({"job": "testing2"}, [key])
self.database.item_update({"job": "testing2"}, str(key))
except TypeError:
self.fail(
"Unexpectedly, item_update raises an Error with types of ids which should be usable"
)
def test_delete_item(self):
"""
Tests delete_item function
Returns:
"""
par_dict = self.add_items("BO")
key = par_dict["id"]
self.database.delete_item(key)
self.assertRaises(
Exception, self.database.delete_item, [key]
) # use only str or int
# self.assertRaises(Exception, self.database.get_item_by_id, key) # ensure item does not exist anymore
def test_get_item_by_id(self):
"""
Tests get_item_by_id function
Returns:
"""
par_dict = self.add_items("BO")
key = par_dict["id"]
self.assertRaises(
Exception, self.database.get_item_by_id, [key]
) # given key must be int or str
# self.assertRaises(Exception, self.database.get_item_by_id,
# str(key + 1)) # must give Error, if id does not exist
self.assertIsInstance(
self.database.get_item_by_id(key), dict
) # return value has to be a dict
# added dict must (almost) be same as the got ones
result = self.database.get_item_by_id(key)
self.assertTrue(par_dict.items() <= result.items())
def test_get_items_dict_and(self):
"""
Tests the 'and' functionality of get_items_dict function
Returns:
"""
self.add_items("Blub")
# tests general and statements
item_dict = {"hamilton": "VAMPE", "hamversion": "1.1"}
self.assertEqual(
self.database.get_items_dict(item_dict),
self.database.get_items_sql("hamilton='VAMPE' and hamversion='1.1'"),
)
def test_get_items_dict_project(self):
"""
Tests whether a query for {'project': 'Projecta%'} gives Projecta, Projecta/b/c , but not Projectas
Returns:
"""
par_dict = {
"chemicalformula": "H2",
"computer": "localhost",
"hamilton": "VAMPE",
"hamversion": "1.1",
"job": "testing",
"parentid": 0,
"project": "Projecta/",
"projectpath": "/TESTING",
"status": "KAAAA",
"timestart": datetime(2016, 5, 2, 11, 31, 4, 253377),
"timestop": datetime(2016, 5, 2, 11, 31, 4, 371165),
"totalcputime": 0.117788,
"username": "User",
"masterid": 0,
"subjob": "testJob",
}
second_dict = dict(par_dict, project="Projecta/b/c/")
third_dict = dict(par_dict, project="Projectas")
par_dict["id"] = self.database.add_item_dict(par_dict)
second_dict["id"] = self.database.add_item_dict(second_dict)
third_dict["id"] = self.database.add_item_dict(third_dict)
self.assertEqual(
[par_dict, second_dict],
self.database.get_items_dict({"project": "Projecta/%"}),
)
def test_get_items_dict_or(self):
"""
Tests 'or' functionality of get_items_dict function
Returns:
"""
self.add_items("Blub")
self.add_items("Blab")
# tests an example or statement
item_dict = {"chemicalformula": ["Blub", "Blab"]}
# assert that both the sql and non-sql methods give the same result
sql_db = self.database.get_items_sql(
"chemicalformula='Blub' or chemicalformula='Blab'"
)
dict_db = self.database.get_items_dict(item_dict)
for item in sql_db:
self.assertTrue(item in dict_db)
def test_get_items_dict_like(self):
"""
Tests 'like' functionality of get_items_dict function
Returns:
"""
self.add_items("Blub")
# tests an example like statement
item_dict = {"status": "%AA%"}
# assert that both the sql and non-sql methods give the same result
sql_db = self.database.get_items_sql("status like '%AA%'")
dict_db = self.database.get_items_dict(item_dict)
for item in sql_db:
self.assertTrue(item in dict_db)
def test_get_items_dict_datatype(self):
"""
Tests datatype error functionality of get_items_dict function
Returns:
"""
# ensures right datatype
item_dict = ["test", "test"]
self.assertRaises(TypeError, self.database.get_items_dict, item_dict)
def test_z_add_column(self):
"""
Tests add_column function
Name includes a z so that it is run last. Altering a table can lead
the other tests to fail.
Returns:
"""
self.add_items("blub")
column = "myColumn5"
if column not in self.database.get_table_headings():
self.database.add_column(column, "varchar(50)")
self.assertRaises(
Exception, self.database.add_column, column
) # cannot add column with same name
self.assertTrue(
column in self.database.get_table_headings()
) # see whether myColumn has been included
try:
# list should be usable, but function will just take last element of lists
second_column = "".join(
choice(ascii_uppercase) + str(i) for i in range(12)
) # random generator for columns
self.database.add_column([second_column], ["varchar(50)"])
self.database.add_column([second_column + "2"], "varchar(50)")
except TypeError:
self.fail("Unexpectedly add_column cannot take lists as parameter.")
self.assertRaises(
Exception, self.database.add_column, ["mycolumn"], 10
) # cannot use int as params
# NOT A TEST #
def add_items(self, formula):
"""
Simple generic helper function to add items to DB
Args:
formula: string for chemicalformula
Returns:
"""
par_dict = {
"chemicalformula": formula,
"computer": "localhost",
"hamilton": "VAMPE",
"hamversion": "1.1",
"job": "testing",
"parentid": 0,
"project": "database.testing/",
"projectpath": "/TESTING",
"status": "KAAAA",
"timestart": datetime(2016, 5, 2, 11, 31, 4, 253377),
"timestop": datetime(2016, 5, 2, 11, 31, 4, 371165),
"totalcputime": 0.117788,
"username": "User",
"masterid": 0,
"subjob": "testJob",
}
par_dict["id"] = self.database.add_item_dict(par_dict)
return par_dict
if __name__ == "__main__":
unittest.main()