0b62485d74641d5b250f94dc67549dfc5110d157
5 from unittest
.mock
import MagicMock
6 from osm_common
.dbbase
import DbException
7 from osm_common
.dbmemory
import DbMemory
9 __author__
= 'Eduardo Sousa <eduardosousa@av.it.pt>'
17 def db_memory_with_data():
20 db
.create("test", {"_id": 1, "data": 1})
21 db
.create("test", {"_id": 2, "data": 2})
22 db
.create("test", {"_id": 3, "data": 3})
28 def empty_exception_message():
29 return 'database exception '
31 def get_one_exception_message(filter):
32 return "database exception Not found entry with filter='{}'".format(filter)
34 def get_one_multiple_exception_message(filter):
35 return "database exception Found more than one entry with filter='{}'".format(filter)
37 def del_one_exception_message(filter):
38 return "database exception Not found entry with filter='{}'".format(filter)
40 def replace_exception_message(filter):
41 return "database exception Not found entry with filter='{}'".format(filter)
43 def test_constructor():
46 assert db
.logger
== logging
.getLogger('db')
47 assert len(db
.db
) == 0
49 def test_constructor_with_logger():
50 logger_name
= 'db_local'
52 db
= DbMemory(logger_name
=logger_name
)
54 assert db
.logger
== logging
.getLogger(logger_name
)
55 assert len(db
.db
) == 0
57 def test_db_connect():
58 logger_name
= 'db_local'
59 config
= {'logger_name': logger_name
}
64 assert db
.logger
== logging
.getLogger(logger_name
)
65 assert len(db
.db
) == 0
67 @pytest.mark
.parametrize("table, filter", [
70 ("test", {"data": 1}),
71 ("test", {"_id": 1, "data": 1}),
73 def test_get_list_with_empty_db(db_memory
, table
, filter):
74 result
= db_memory
.get_list(table
, filter)
76 assert len(result
) == 0
78 @pytest.mark
.parametrize("table, filter, expected_data", [
79 ("test", {}, [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
80 ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
81 ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
82 ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
83 ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
84 ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
85 ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
86 ("test", {"_id": 4}, []),
87 ("test", {"data": 4}, []),
88 ("test", {"_id": 4, "data": 4}, []),
89 ("test_table", {}, []),
90 ("test_table", {"_id": 1}, []),
91 ("test_table", {"data": 1}, []),
92 ("test_table", {"_id": 1, "data": 1}, []),
94 def test_get_list_with_non_empty_db(db_memory_with_data
, table
, filter, expected_data
):
95 result
= db_memory_with_data
.get_list(table
, filter)
97 assert len(result
) == len(expected_data
)
98 for data
in expected_data
:
101 def test_get_list_exception(db_memory_with_data
):
105 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
107 with pytest
.raises(DbException
) as excinfo
:
108 db_memory_with_data
.get_list(table
, filter)
109 assert str(excinfo
.value
) == empty_exception_message()
110 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
112 @pytest.mark
.parametrize("table, filter, expected_data", [
113 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
114 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
115 ("test", {"_id": 3}, {"_id": 3, "data": 3}),
116 ("test", {"data": 1}, {"_id": 1, "data": 1}),
117 ("test", {"data": 2}, {"_id": 2, "data": 2}),
118 ("test", {"data": 3}, {"_id": 3, "data": 3}),
119 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
120 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
121 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3}),
123 def test_get_one(db_memory_with_data
, table
, filter, expected_data
):
124 result
= db_memory_with_data
.get_one(table
, filter)
126 assert result
== expected_data
127 assert len(db_memory_with_data
.db
) == 1
128 assert table
in db_memory_with_data
.db
129 assert len(db_memory_with_data
.db
[table
]) == 3
130 assert result
in db_memory_with_data
.db
[table
]
132 @pytest.mark
.parametrize("table, filter, expected_data", [
133 ("test", {}, {"_id": 1, "data": 1}),
135 def test_get_one_with_multiple_results(db_memory_with_data
, table
, filter, expected_data
):
136 result
= db_memory_with_data
.get_one(table
, filter, fail_on_more
=False)
138 assert result
== expected_data
139 assert len(db_memory_with_data
.db
) == 1
140 assert table
in db_memory_with_data
.db
141 assert len(db_memory_with_data
.db
[table
]) == 3
142 assert result
in db_memory_with_data
.db
[table
]
144 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
148 with pytest
.raises(DbException
) as excinfo
:
149 db_memory_with_data
.get_one(table
, filter)
151 assert str(excinfo
.value
) == (empty_exception_message() + get_one_multiple_exception_message(filter))
152 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
154 @pytest.mark
.parametrize("table, filter", [
155 ("test", {"_id": 4}),
156 ("test", {"data": 4}),
157 ("test", {"_id": 4, "data": 4}),
158 ("test_table", {"_id": 4}),
159 ("test_table", {"data": 4}),
160 ("test_table", {"_id": 4, "data": 4}),
162 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, filter):
163 with pytest
.raises(DbException
) as excinfo
:
164 db_memory_with_data
.get_one(table
, filter)
165 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(filter))
166 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
168 @pytest.mark
.parametrize("table, filter", [
169 ("test", {"_id": 4}),
170 ("test", {"data": 4}),
171 ("test", {"_id": 4, "data": 4}),
172 ("test_table", {"_id": 4}),
173 ("test_table", {"data": 4}),
174 ("test_table", {"_id": 4, "data": 4}),
176 def test_get_one_with_non_empty_db_none(db_memory_with_data
, table
, filter):
177 result
= db_memory_with_data
.get_one(table
, filter, fail_on_empty
=False)
179 assert result
== None
181 @pytest.mark
.parametrize("table, filter", [
182 ("test", {"_id": 4}),
183 ("test", {"data": 4}),
184 ("test", {"_id": 4, "data": 4}),
185 ("test_table", {"_id": 4}),
186 ("test_table", {"data": 4}),
187 ("test_table", {"_id": 4, "data": 4}),
189 def test_get_one_with_empty_db_exception(db_memory
, table
, filter):
190 with pytest
.raises(DbException
) as excinfo
:
191 db_memory
.get_one(table
, filter)
192 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(filter))
193 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
195 @pytest.mark
.parametrize("table, filter", [
196 ("test", {"_id": 4}),
197 ("test", {"data": 4}),
198 ("test", {"_id": 4, "data": 4}),
199 ("test_table", {"_id": 4}),
200 ("test_table", {"data": 4}),
201 ("test_table", {"_id": 4, "data": 4}),
203 def test_get_one_with_empty_db_none(db_memory
, table
, filter):
204 result
= db_memory
.get_one(table
, filter, fail_on_empty
=False)
206 assert result
== None
208 def test_get_one_generic_exception(db_memory_with_data
):
212 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
214 with pytest
.raises(DbException
) as excinfo
:
215 db_memory_with_data
.get_one(table
, filter)
216 assert str(excinfo
.value
) == empty_exception_message()
217 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
219 @pytest.mark
.parametrize("table, filter, expected_data", [
221 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
222 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
223 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
224 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
226 def test_del_list_with_non_empty_db(db_memory_with_data
, table
, filter, expected_data
):
227 result
= db_memory_with_data
.del_list(table
, filter)
229 assert result
["deleted"] == (3 - len(expected_data
))
230 assert len(db_memory_with_data
.db
) == 1
231 assert table
in db_memory_with_data
.db
232 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
233 for data
in expected_data
:
234 assert data
in db_memory_with_data
.db
[table
]
236 @pytest.mark
.parametrize("table, filter", [
238 ("test", {"_id": 1}),
239 ("test", {"_id": 2}),
240 ("test", {"data": 1}),
241 ("test", {"data": 2}),
242 ("test", {"_id": 1, "data": 1}),
243 ("test", {"_id": 2, "data": 2}),
245 def test_del_list_with_empty_db(db_memory
, table
, filter):
246 result
= db_memory
.del_list(table
, filter)
247 assert result
['deleted'] == 0
249 def test_del_list_generic_exception(db_memory_with_data
):
253 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
255 with pytest
.raises(DbException
) as excinfo
:
256 db_memory_with_data
.del_list(table
, filter)
257 assert str(excinfo
.value
) == empty_exception_message()
258 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
260 @pytest.mark
.parametrize("table, filter, data", [
261 ("test", {}, {"_id": 1, "data": 1}),
262 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
263 ("test", {"data": 1}, {"_id": 1, "data": 1}),
264 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
265 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
266 ("test", {"data": 2}, {"_id": 2, "data": 2}),
267 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
269 def test_del_one(db_memory_with_data
, table
, filter, data
):
270 result
= db_memory_with_data
.del_one(table
, filter)
272 assert result
== {"deleted": 1}
273 assert len(db_memory_with_data
.db
) == 1
274 assert table
in db_memory_with_data
.db
275 assert len(db_memory_with_data
.db
[table
]) == 2
276 assert data
not in db_memory_with_data
.db
[table
]
278 @pytest.mark
.parametrize("table, filter", [
280 ("test", {"_id": 1}),
281 ("test", {"_id": 2}),
282 ("test", {"data": 1}),
283 ("test", {"data": 2}),
284 ("test", {"_id": 1, "data": 1}),
285 ("test", {"_id": 2, "data": 2}),
287 ("test_table", {"_id": 1}),
288 ("test_table", {"_id": 2}),
289 ("test_table", {"data": 1}),
290 ("test_table", {"data": 2}),
291 ("test_table", {"_id": 1, "data": 1}),
292 ("test_table", {"_id": 2, "data": 2}),
294 def test_del_one_with_empty_db_exception(db_memory
, table
, filter):
295 with pytest
.raises(DbException
) as excinfo
:
296 db_memory
.del_one(table
, filter)
297 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(filter))
298 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
300 @pytest.mark
.parametrize("table, filter", [
302 ("test", {"_id": 1}),
303 ("test", {"_id": 2}),
304 ("test", {"data": 1}),
305 ("test", {"data": 2}),
306 ("test", {"_id": 1, "data": 1}),
307 ("test", {"_id": 2, "data": 2}),
309 ("test_table", {"_id": 1}),
310 ("test_table", {"_id": 2}),
311 ("test_table", {"data": 1}),
312 ("test_table", {"data": 2}),
313 ("test_table", {"_id": 1, "data": 1}),
314 ("test_table", {"_id": 2, "data": 2}),
316 def test_del_one_with_empty_db_none(db_memory
, table
, filter):
317 result
= db_memory
.del_one(table
, filter, fail_on_empty
=False)
319 assert result
== None
321 @pytest.mark
.parametrize("table, filter", [
322 ("test", {"_id": 4}),
323 ("test", {"_id": 5}),
324 ("test", {"data": 4}),
325 ("test", {"data": 5}),
326 ("test", {"_id": 1, "data": 2}),
327 ("test", {"_id": 2, "data": 3}),
329 ("test_table", {"_id": 1}),
330 ("test_table", {"_id": 2}),
331 ("test_table", {"data": 1}),
332 ("test_table", {"data": 2}),
333 ("test_table", {"_id": 1, "data": 1}),
334 ("test_table", {"_id": 2, "data": 2}),
336 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, filter):
337 with pytest
.raises(DbException
) as excinfo
:
338 db_memory_with_data
.del_one(table
, filter)
339 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(filter))
340 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
342 @pytest.mark
.parametrize("table, filter", [
343 ("test", {"_id": 4}),
344 ("test", {"_id": 5}),
345 ("test", {"data": 4}),
346 ("test", {"data": 5}),
347 ("test", {"_id": 1, "data": 2}),
348 ("test", {"_id": 2, "data": 3}),
350 ("test_table", {"_id": 1}),
351 ("test_table", {"_id": 2}),
352 ("test_table", {"data": 1}),
353 ("test_table", {"data": 2}),
354 ("test_table", {"_id": 1, "data": 1}),
355 ("test_table", {"_id": 2, "data": 2}),
357 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, filter):
358 result
= db_memory_with_data
.del_one(table
, filter, fail_on_empty
=False)
360 assert result
== None
362 @pytest.mark
.parametrize("fail_on_empty", [
366 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
370 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
372 with pytest
.raises(DbException
) as excinfo
:
373 db_memory_with_data
.del_one(table
, filter, fail_on_empty
=fail_on_empty
)
374 assert str(excinfo
.value
) == empty_exception_message()
375 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
377 @pytest.mark
.parametrize("table, filter, indata", [
378 ("test", {}, {"_id": 1, "data": 42}),
379 ("test", {}, {"_id": 3, "data": 42}),
380 ("test", {"_id": 1}, {"_id": 3, "data": 42}),
381 ("test", {"_id": 3}, {"_id": 3, "data": 42}),
382 ("test", {"data": 1}, {"_id": 3, "data": 42}),
383 ("test", {"data": 3}, {"_id": 3, "data": 42}),
384 ("test", {"_id": 1, "data": 1}, {"_id": 3, "data": 42}),
385 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 42}),
387 def test_replace(db_memory_with_data
, table
, filter, indata
):
388 result
= db_memory_with_data
.replace(table
, filter, indata
)
390 assert result
== {"updated": 1}
391 assert len(db_memory_with_data
.db
) == 1
392 assert table
in db_memory_with_data
.db
393 assert len(db_memory_with_data
.db
[table
]) == 3
394 assert indata
in db_memory_with_data
.db
[table
]
396 @pytest.mark
.parametrize("table, filter, indata", [
397 ("test", {}, {'_id': 1, 'data': 1}),
398 ("test", {}, {'_id': 2, 'data': 1}),
399 ("test", {}, {'_id': 1, 'data': 2}),
400 ("test", {'_id': 1}, {'_id': 1, 'data': 1}),
401 ("test", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
402 ("test_table", {}, {'_id': 1, 'data': 1}),
403 ("test_table", {}, {'_id': 2, 'data': 1}),
404 ("test_table", {}, {'_id': 1, 'data': 2}),
405 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
406 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
408 def test_replace_without_data_exception(db_memory
, table
, filter, indata
):
409 with pytest
.raises(DbException
) as excinfo
:
410 db_memory
.replace(table
, filter, indata
, fail_on_empty
=True)
411 assert str(excinfo
.value
) == (empty_exception_message() + replace_exception_message(filter))
412 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
414 @pytest.mark
.parametrize("table, filter, indata", [
415 ("test", {}, {'_id': 1, 'data': 1}),
416 ("test", {}, {'_id': 2, 'data': 1}),
417 ("test", {}, {'_id': 1, 'data': 2}),
418 ("test", {'_id': 1}, {'_id': 1, 'data': 1}),
419 ("test", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
420 ("test_table", {}, {'_id': 1, 'data': 1}),
421 ("test_table", {}, {'_id': 2, 'data': 1}),
422 ("test_table", {}, {'_id': 1, 'data': 2}),
423 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
424 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
426 def test_replace_without_data_none(db_memory
, table
, filter, indata
):
427 result
= db_memory
.replace(table
, filter, indata
, fail_on_empty
=False)
428 assert result
== None
430 @pytest.mark
.parametrize("table, filter, indata", [
431 ("test_table", {}, {'_id': 1, 'data': 1}),
432 ("test_table", {}, {'_id': 2, 'data': 1}),
433 ("test_table", {}, {'_id': 1, 'data': 2}),
434 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
435 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
437 def test_replace_with_data_exception(db_memory_with_data
, table
, filter, indata
):
438 with pytest
.raises(DbException
) as excinfo
:
439 db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=True)
440 assert str(excinfo
.value
) == (empty_exception_message() + replace_exception_message(filter))
441 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
443 @pytest.mark
.parametrize("table, filter, indata", [
444 ("test_table", {}, {'_id': 1, 'data': 1}),
445 ("test_table", {}, {'_id': 2, 'data': 1}),
446 ("test_table", {}, {'_id': 1, 'data': 2}),
447 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
448 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
450 def test_replace_with_data_none(db_memory_with_data
, table
, filter, indata
):
451 result
= db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=False)
452 assert result
== None
454 @pytest.mark
.parametrize("fail_on_empty", [
458 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
461 indata
= {'_id': 1, 'data': 1}
463 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
465 with pytest
.raises(DbException
) as excinfo
:
466 db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=fail_on_empty
)
467 assert str(excinfo
.value
) == empty_exception_message()
468 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
470 @pytest.mark
.parametrize("table, id, data", [
471 ("test", "1", {"data": 1}),
472 ("test", "1", {"data": 2}),
473 ("test", "2", {"data": 1}),
474 ("test", "2", {"data": 2}),
475 ("test_table", "1", {"data": 1}),
476 ("test_table", "1", {"data": 2}),
477 ("test_table", "2", {"data": 1}),
478 ("test_table", "2", {"data": 2}),
479 ("test", "1", {"data_1": 1, "data_2": 2}),
480 ("test", "1", {"data_1": 2, "data_2": 1}),
481 ("test", "2", {"data_1": 1, "data_2": 2}),
482 ("test", "2", {"data_1": 2, "data_2": 1}),
483 ("test_table", "1", {"data_1": 1, "data_2": 2}),
484 ("test_table", "1", {"data_1": 2, "data_2": 1}),
485 ("test_table", "2", {"data_1": 1, "data_2": 2}),
486 ("test_table", "2", {"data_1": 2, "data_2": 1}),
488 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
489 data_to_insert
= data
490 data_to_insert
['_id'] = id
492 returned_id
= db_memory
.create(table
, data_to_insert
)
494 assert returned_id
== id
495 assert len(db_memory
.db
) == 1
496 assert table
in db_memory
.db
497 assert len(db_memory
.db
[table
]) == 1
498 assert data_to_insert
in db_memory
.db
[table
]
500 @pytest.mark
.parametrize("table, id, data", [
501 ("test", "4", {"data": 1}),
502 ("test", "5", {"data": 2}),
503 ("test", "4", {"data": 1}),
504 ("test", "5", {"data": 2}),
505 ("test_table", "4", {"data": 1}),
506 ("test_table", "5", {"data": 2}),
507 ("test_table", "4", {"data": 1}),
508 ("test_table", "5", {"data": 2}),
509 ("test", "4", {"data_1": 1, "data_2": 2}),
510 ("test", "5", {"data_1": 2, "data_2": 1}),
511 ("test", "4", {"data_1": 1, "data_2": 2}),
512 ("test", "5", {"data_1": 2, "data_2": 1}),
513 ("test_table", "4", {"data_1": 1, "data_2": 2}),
514 ("test_table", "5", {"data_1": 2, "data_2": 1}),
515 ("test_table", "4", {"data_1": 1, "data_2": 2}),
516 ("test_table", "5", {"data_1": 2, "data_2": 1}),
518 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
519 data_to_insert
= data
520 data_to_insert
['_id'] = id
522 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
524 assert returned_id
== id
525 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
526 assert table
in db_memory_with_data
.db
527 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
528 assert data_to_insert
in db_memory_with_data
.db
[table
]
530 @pytest.mark
.parametrize("table, data", [
531 ("test", {"data": 1}),
532 ("test", {"data": 2}),
533 ("test", {"data": 1}),
534 ("test", {"data": 2}),
535 ("test_table", {"data": 1}),
536 ("test_table", {"data": 2}),
537 ("test_table", {"data": 1}),
538 ("test_table", {"data": 2}),
539 ("test", {"data_1": 1, "data_2": 2}),
540 ("test", {"data_1": 2, "data_2": 1}),
541 ("test", {"data_1": 1, "data_2": 2}),
542 ("test", {"data_1": 2, "data_2": 1}),
543 ("test_table", {"data_1": 1, "data_2": 2}),
544 ("test_table", {"data_1": 2, "data_2": 1}),
545 ("test_table", {"data_1": 1, "data_2": 2}),
546 ("test_table", {"data_1": 2, "data_2": 1}),
548 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
549 returned_id
= db_memory
.create(table
, data
)
551 assert len(db_memory
.db
) == 1
552 assert table
in db_memory
.db
553 assert len(db_memory
.db
[table
]) == 1
556 data_inserted
['_id'] = returned_id
558 assert data_inserted
in db_memory
.db
[table
]
560 @pytest.mark
.parametrize("table, data", [
561 ("test", {"data": 1}),
562 ("test", {"data": 2}),
563 ("test", {"data": 1}),
564 ("test", {"data": 2}),
565 ("test_table", {"data": 1}),
566 ("test_table", {"data": 2}),
567 ("test_table", {"data": 1}),
568 ("test_table", {"data": 2}),
569 ("test", {"data_1": 1, "data_2": 2}),
570 ("test", {"data_1": 2, "data_2": 1}),
571 ("test", {"data_1": 1, "data_2": 2}),
572 ("test", {"data_1": 2, "data_2": 1}),
573 ("test_table", {"data_1": 1, "data_2": 2}),
574 ("test_table", {"data_1": 2, "data_2": 1}),
575 ("test_table", {"data_1": 1, "data_2": 2}),
576 ("test_table", {"data_1": 2, "data_2": 1}),
578 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
579 returned_id
= db_memory_with_data
.create(table
, data
)
581 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
582 assert table
in db_memory_with_data
.db
583 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
586 data_inserted
['_id'] = returned_id
588 assert data_inserted
in db_memory_with_data
.db
[table
]
590 def test_create_with_exception(db_memory
):
592 data
= {"_id": 1, "data": 1}
594 db_memory
.db
= MagicMock()
595 db_memory
.db
.__contains
__.side_effect
= Exception()
597 with pytest
.raises(DbException
) as excinfo
:
598 db_memory
.create(table
, data
)
599 assert str(excinfo
.value
) == empty_exception_message()
600 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND