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})
26 def empty_exception_message():
27 return 'database exception '
29 def get_one_exception_message(filter):
30 return "database exception Not found entry with filter='{}'".format(filter)
32 def get_one_multiple_exception_message(filter):
33 return "database exception Found more than one entry with filter='{}'".format(filter)
35 def del_one_exception_message(filter):
36 return "database exception Not found entry with filter='{}'".format(filter)
38 def replace_exception_message(filter):
39 return "database exception Not found entry with filter='{}'".format(filter)
41 def test_constructor():
44 assert db
.logger
== logging
.getLogger('db')
45 assert len(db
.db
) == 0
47 def test_constructor_with_logger():
48 logger_name
= 'db_local'
50 db
= DbMemory(logger_name
=logger_name
)
52 assert db
.logger
== logging
.getLogger(logger_name
)
53 assert len(db
.db
) == 0
55 def test_db_connect():
56 logger_name
= 'db_local'
57 config
= {'logger_name': logger_name
}
62 assert db
.logger
== logging
.getLogger(logger_name
)
63 assert len(db
.db
) == 0
65 def test_db_disconnect(db_memory
):
66 db_memory
.db_disconnect()
68 @pytest.mark
.parametrize("table, filter", [
71 ("test", {"data": 1}),
72 ("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}, [])])
93 def test_get_list_with_non_empty_db(db_memory_with_data
, table
, filter, expected_data
):
94 result
= db_memory_with_data
.get_list(table
, filter)
96 assert len(result
) == len(expected_data
)
97 for data
in expected_data
:
100 def test_get_list_exception(db_memory_with_data
):
104 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
106 with pytest
.raises(DbException
) as excinfo
:
107 db_memory_with_data
.get_list(table
, filter)
108 assert str(excinfo
.value
) == empty_exception_message()
109 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
111 @pytest.mark
.parametrize("table, filter, expected_data", [
112 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
113 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
114 ("test", {"_id": 3}, {"_id": 3, "data": 3}),
115 ("test", {"data": 1}, {"_id": 1, "data": 1}),
116 ("test", {"data": 2}, {"_id": 2, "data": 2}),
117 ("test", {"data": 3}, {"_id": 3, "data": 3}),
118 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
119 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
120 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3})])
121 def test_get_one(db_memory_with_data
, table
, filter, expected_data
):
122 result
= db_memory_with_data
.get_one(table
, filter)
124 assert result
== expected_data
125 assert len(db_memory_with_data
.db
) == 1
126 assert table
in db_memory_with_data
.db
127 assert len(db_memory_with_data
.db
[table
]) == 3
128 assert result
in db_memory_with_data
.db
[table
]
130 @pytest.mark
.parametrize("table, filter, expected_data", [
131 ("test", {}, {"_id": 1, "data": 1})])
132 def test_get_one_with_multiple_results(db_memory_with_data
, table
, filter, expected_data
):
133 result
= db_memory_with_data
.get_one(table
, filter, fail_on_more
=False)
135 assert result
== expected_data
136 assert len(db_memory_with_data
.db
) == 1
137 assert table
in db_memory_with_data
.db
138 assert len(db_memory_with_data
.db
[table
]) == 3
139 assert result
in db_memory_with_data
.db
[table
]
141 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
145 with pytest
.raises(DbException
) as excinfo
:
146 db_memory_with_data
.get_one(table
, filter)
148 assert str(excinfo
.value
) == (empty_exception_message() + get_one_multiple_exception_message(filter))
149 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
151 @pytest.mark
.parametrize("table, filter", [
152 ("test", {"_id": 4}),
153 ("test", {"data": 4}),
154 ("test", {"_id": 4, "data": 4}),
155 ("test_table", {"_id": 4}),
156 ("test_table", {"data": 4}),
157 ("test_table", {"_id": 4, "data": 4})])
158 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, filter):
159 with pytest
.raises(DbException
) as excinfo
:
160 db_memory_with_data
.get_one(table
, filter)
161 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(filter))
162 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
164 @pytest.mark
.parametrize("table, filter", [
165 ("test", {"_id": 4}),
166 ("test", {"data": 4}),
167 ("test", {"_id": 4, "data": 4}),
168 ("test_table", {"_id": 4}),
169 ("test_table", {"data": 4}),
170 ("test_table", {"_id": 4, "data": 4})])
171 def test_get_one_with_non_empty_db_none(db_memory_with_data
, table
, filter):
172 result
= db_memory_with_data
.get_one(table
, filter, fail_on_empty
=False)
174 assert result
== None
176 @pytest.mark
.parametrize("table, filter", [
177 ("test", {"_id": 4}),
178 ("test", {"data": 4}),
179 ("test", {"_id": 4, "data": 4}),
180 ("test_table", {"_id": 4}),
181 ("test_table", {"data": 4}),
182 ("test_table", {"_id": 4, "data": 4})])
183 def test_get_one_with_empty_db_exception(db_memory
, table
, filter):
184 with pytest
.raises(DbException
) as excinfo
:
185 db_memory
.get_one(table
, filter)
186 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(filter))
187 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
189 @pytest.mark
.parametrize("table, filter", [
190 ("test", {"_id": 4}),
191 ("test", {"data": 4}),
192 ("test", {"_id": 4, "data": 4}),
193 ("test_table", {"_id": 4}),
194 ("test_table", {"data": 4}),
195 ("test_table", {"_id": 4, "data": 4})])
196 def test_get_one_with_empty_db_none(db_memory
, table
, filter):
197 result
= db_memory
.get_one(table
, filter, fail_on_empty
=False)
199 assert result
== None
201 def test_get_one_generic_exception(db_memory_with_data
):
205 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
207 with pytest
.raises(DbException
) as excinfo
:
208 db_memory_with_data
.get_one(table
, filter)
209 assert str(excinfo
.value
) == empty_exception_message()
210 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
212 @pytest.mark
.parametrize("table, filter, expected_data", [
214 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
215 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
216 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
217 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
218 def test_del_list_with_non_empty_db(db_memory_with_data
, table
, filter, expected_data
):
219 result
= db_memory_with_data
.del_list(table
, filter)
221 assert result
["deleted"] == (3 - len(expected_data
))
222 assert len(db_memory_with_data
.db
) == 1
223 assert table
in db_memory_with_data
.db
224 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
225 for data
in expected_data
:
226 assert data
in db_memory_with_data
.db
[table
]
228 @pytest.mark
.parametrize("table, filter", [
230 ("test", {"_id": 1}),
231 ("test", {"_id": 2}),
232 ("test", {"data": 1}),
233 ("test", {"data": 2}),
234 ("test", {"_id": 1, "data": 1}),
235 ("test", {"_id": 2, "data": 2})])
236 def test_del_list_with_empty_db(db_memory
, table
, filter):
237 result
= db_memory
.del_list(table
, filter)
238 assert result
['deleted'] == 0
240 def test_del_list_generic_exception(db_memory_with_data
):
244 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
246 with pytest
.raises(DbException
) as excinfo
:
247 db_memory_with_data
.del_list(table
, filter)
248 assert str(excinfo
.value
) == empty_exception_message()
249 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
251 @pytest.mark
.parametrize("table, filter, data", [
252 ("test", {}, {"_id": 1, "data": 1}),
253 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
254 ("test", {"data": 1}, {"_id": 1, "data": 1}),
255 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
256 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
257 ("test", {"data": 2}, {"_id": 2, "data": 2}),
258 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
259 def test_del_one(db_memory_with_data
, table
, filter, data
):
260 result
= db_memory_with_data
.del_one(table
, filter)
262 assert result
== {"deleted": 1}
263 assert len(db_memory_with_data
.db
) == 1
264 assert table
in db_memory_with_data
.db
265 assert len(db_memory_with_data
.db
[table
]) == 2
266 assert data
not in db_memory_with_data
.db
[table
]
268 @pytest.mark
.parametrize("table, filter", [
270 ("test", {"_id": 1}),
271 ("test", {"_id": 2}),
272 ("test", {"data": 1}),
273 ("test", {"data": 2}),
274 ("test", {"_id": 1, "data": 1}),
275 ("test", {"_id": 2, "data": 2}),
277 ("test_table", {"_id": 1}),
278 ("test_table", {"_id": 2}),
279 ("test_table", {"data": 1}),
280 ("test_table", {"data": 2}),
281 ("test_table", {"_id": 1, "data": 1}),
282 ("test_table", {"_id": 2, "data": 2})])
283 def test_del_one_with_empty_db_exception(db_memory
, table
, filter):
284 with pytest
.raises(DbException
) as excinfo
:
285 db_memory
.del_one(table
, filter)
286 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(filter))
287 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
289 @pytest.mark
.parametrize("table, filter", [
291 ("test", {"_id": 1}),
292 ("test", {"_id": 2}),
293 ("test", {"data": 1}),
294 ("test", {"data": 2}),
295 ("test", {"_id": 1, "data": 1}),
296 ("test", {"_id": 2, "data": 2}),
298 ("test_table", {"_id": 1}),
299 ("test_table", {"_id": 2}),
300 ("test_table", {"data": 1}),
301 ("test_table", {"data": 2}),
302 ("test_table", {"_id": 1, "data": 1}),
303 ("test_table", {"_id": 2, "data": 2})])
304 def test_del_one_with_empty_db_none(db_memory
, table
, filter):
305 result
= db_memory
.del_one(table
, filter, fail_on_empty
=False)
307 assert result
== None
309 @pytest.mark
.parametrize("table, filter", [
310 ("test", {"_id": 4}),
311 ("test", {"_id": 5}),
312 ("test", {"data": 4}),
313 ("test", {"data": 5}),
314 ("test", {"_id": 1, "data": 2}),
315 ("test", {"_id": 2, "data": 3}),
317 ("test_table", {"_id": 1}),
318 ("test_table", {"_id": 2}),
319 ("test_table", {"data": 1}),
320 ("test_table", {"data": 2}),
321 ("test_table", {"_id": 1, "data": 1}),
322 ("test_table", {"_id": 2, "data": 2})])
323 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, filter):
324 with pytest
.raises(DbException
) as excinfo
:
325 db_memory_with_data
.del_one(table
, filter)
326 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(filter))
327 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
329 @pytest.mark
.parametrize("table, filter", [
330 ("test", {"_id": 4}),
331 ("test", {"_id": 5}),
332 ("test", {"data": 4}),
333 ("test", {"data": 5}),
334 ("test", {"_id": 1, "data": 2}),
335 ("test", {"_id": 2, "data": 3}),
337 ("test_table", {"_id": 1}),
338 ("test_table", {"_id": 2}),
339 ("test_table", {"data": 1}),
340 ("test_table", {"data": 2}),
341 ("test_table", {"_id": 1, "data": 1}),
342 ("test_table", {"_id": 2, "data": 2})])
343 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, filter):
344 result
= db_memory_with_data
.del_one(table
, filter, fail_on_empty
=False)
346 assert result
== None
348 @pytest.mark
.parametrize("fail_on_empty", [
351 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
355 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
357 with pytest
.raises(DbException
) as excinfo
:
358 db_memory_with_data
.del_one(table
, filter, fail_on_empty
=fail_on_empty
)
359 assert str(excinfo
.value
) == empty_exception_message()
360 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
362 @pytest.mark
.parametrize("table, filter, indata", [
363 ("test", {}, {"_id": 1, "data": 42}),
364 ("test", {}, {"_id": 3, "data": 42}),
365 ("test", {"_id": 1}, {"_id": 3, "data": 42}),
366 ("test", {"_id": 3}, {"_id": 3, "data": 42}),
367 ("test", {"data": 1}, {"_id": 3, "data": 42}),
368 ("test", {"data": 3}, {"_id": 3, "data": 42}),
369 ("test", {"_id": 1, "data": 1}, {"_id": 3, "data": 42}),
370 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 42})])
371 def test_replace(db_memory_with_data
, table
, filter, indata
):
372 result
= db_memory_with_data
.replace(table
, filter, indata
)
374 assert result
== {"updated": 1}
375 assert len(db_memory_with_data
.db
) == 1
376 assert table
in db_memory_with_data
.db
377 assert len(db_memory_with_data
.db
[table
]) == 3
378 assert indata
in db_memory_with_data
.db
[table
]
380 @pytest.mark
.parametrize("table, filter, indata", [
381 ("test", {}, {'_id': 1, 'data': 1}),
382 ("test", {}, {'_id': 2, 'data': 1}),
383 ("test", {}, {'_id': 1, 'data': 2}),
384 ("test", {'_id': 1}, {'_id': 1, 'data': 1}),
385 ("test", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
386 ("test_table", {}, {'_id': 1, 'data': 1}),
387 ("test_table", {}, {'_id': 2, 'data': 1}),
388 ("test_table", {}, {'_id': 1, 'data': 2}),
389 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
390 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
391 def test_replace_without_data_exception(db_memory
, table
, filter, indata
):
392 with pytest
.raises(DbException
) as excinfo
:
393 db_memory
.replace(table
, filter, indata
, fail_on_empty
=True)
394 assert str(excinfo
.value
) == (empty_exception_message() + replace_exception_message(filter))
395 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
397 @pytest.mark
.parametrize("table, filter, indata", [
398 ("test", {}, {'_id': 1, 'data': 1}),
399 ("test", {}, {'_id': 2, 'data': 1}),
400 ("test", {}, {'_id': 1, 'data': 2}),
401 ("test", {'_id': 1}, {'_id': 1, 'data': 1}),
402 ("test", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
403 ("test_table", {}, {'_id': 1, 'data': 1}),
404 ("test_table", {}, {'_id': 2, 'data': 1}),
405 ("test_table", {}, {'_id': 1, 'data': 2}),
406 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
407 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
408 def test_replace_without_data_none(db_memory
, table
, filter, indata
):
409 result
= db_memory
.replace(table
, filter, indata
, fail_on_empty
=False)
410 assert result
== None
412 @pytest.mark
.parametrize("table, filter, indata", [
413 ("test_table", {}, {'_id': 1, 'data': 1}),
414 ("test_table", {}, {'_id': 2, 'data': 1}),
415 ("test_table", {}, {'_id': 1, 'data': 2}),
416 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
417 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
418 def test_replace_with_data_exception(db_memory_with_data
, table
, filter, indata
):
419 with pytest
.raises(DbException
) as excinfo
:
420 db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=True)
421 assert str(excinfo
.value
) == (empty_exception_message() + replace_exception_message(filter))
422 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
424 @pytest.mark
.parametrize("table, filter, indata", [
425 ("test_table", {}, {'_id': 1, 'data': 1}),
426 ("test_table", {}, {'_id': 2, 'data': 1}),
427 ("test_table", {}, {'_id': 1, 'data': 2}),
428 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
429 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
430 def test_replace_with_data_none(db_memory_with_data
, table
, filter, indata
):
431 result
= db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=False)
432 assert result
== None
434 @pytest.mark
.parametrize("fail_on_empty", [
437 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
440 indata
= {'_id': 1, 'data': 1}
442 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
444 with pytest
.raises(DbException
) as excinfo
:
445 db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=fail_on_empty
)
446 assert str(excinfo
.value
) == empty_exception_message()
447 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
449 @pytest.mark
.parametrize("table, id, data", [
450 ("test", "1", {"data": 1}),
451 ("test", "1", {"data": 2}),
452 ("test", "2", {"data": 1}),
453 ("test", "2", {"data": 2}),
454 ("test_table", "1", {"data": 1}),
455 ("test_table", "1", {"data": 2}),
456 ("test_table", "2", {"data": 1}),
457 ("test_table", "2", {"data": 2}),
458 ("test", "1", {"data_1": 1, "data_2": 2}),
459 ("test", "1", {"data_1": 2, "data_2": 1}),
460 ("test", "2", {"data_1": 1, "data_2": 2}),
461 ("test", "2", {"data_1": 2, "data_2": 1}),
462 ("test_table", "1", {"data_1": 1, "data_2": 2}),
463 ("test_table", "1", {"data_1": 2, "data_2": 1}),
464 ("test_table", "2", {"data_1": 1, "data_2": 2}),
465 ("test_table", "2", {"data_1": 2, "data_2": 1})])
466 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
467 data_to_insert
= data
468 data_to_insert
['_id'] = id
470 returned_id
= db_memory
.create(table
, data_to_insert
)
472 assert returned_id
== id
473 assert len(db_memory
.db
) == 1
474 assert table
in db_memory
.db
475 assert len(db_memory
.db
[table
]) == 1
476 assert data_to_insert
in db_memory
.db
[table
]
478 @pytest.mark
.parametrize("table, id, data", [
479 ("test", "4", {"data": 1}),
480 ("test", "5", {"data": 2}),
481 ("test", "4", {"data": 1}),
482 ("test", "5", {"data": 2}),
483 ("test_table", "4", {"data": 1}),
484 ("test_table", "5", {"data": 2}),
485 ("test_table", "4", {"data": 1}),
486 ("test_table", "5", {"data": 2}),
487 ("test", "4", {"data_1": 1, "data_2": 2}),
488 ("test", "5", {"data_1": 2, "data_2": 1}),
489 ("test", "4", {"data_1": 1, "data_2": 2}),
490 ("test", "5", {"data_1": 2, "data_2": 1}),
491 ("test_table", "4", {"data_1": 1, "data_2": 2}),
492 ("test_table", "5", {"data_1": 2, "data_2": 1}),
493 ("test_table", "4", {"data_1": 1, "data_2": 2}),
494 ("test_table", "5", {"data_1": 2, "data_2": 1})])
495 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
496 data_to_insert
= data
497 data_to_insert
['_id'] = id
499 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
501 assert returned_id
== id
502 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
503 assert table
in db_memory_with_data
.db
504 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
505 assert data_to_insert
in db_memory_with_data
.db
[table
]
507 @pytest.mark
.parametrize("table, data", [
508 ("test", {"data": 1}),
509 ("test", {"data": 2}),
510 ("test", {"data": 1}),
511 ("test", {"data": 2}),
512 ("test_table", {"data": 1}),
513 ("test_table", {"data": 2}),
514 ("test_table", {"data": 1}),
515 ("test_table", {"data": 2}),
516 ("test", {"data_1": 1, "data_2": 2}),
517 ("test", {"data_1": 2, "data_2": 1}),
518 ("test", {"data_1": 1, "data_2": 2}),
519 ("test", {"data_1": 2, "data_2": 1}),
520 ("test_table", {"data_1": 1, "data_2": 2}),
521 ("test_table", {"data_1": 2, "data_2": 1}),
522 ("test_table", {"data_1": 1, "data_2": 2}),
523 ("test_table", {"data_1": 2, "data_2": 1})])
524 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
525 returned_id
= db_memory
.create(table
, data
)
527 assert len(db_memory
.db
) == 1
528 assert table
in db_memory
.db
529 assert len(db_memory
.db
[table
]) == 1
532 data_inserted
['_id'] = returned_id
534 assert data_inserted
in db_memory
.db
[table
]
536 @pytest.mark
.parametrize("table, data", [
537 ("test", {"data": 1}),
538 ("test", {"data": 2}),
539 ("test", {"data": 1}),
540 ("test", {"data": 2}),
541 ("test_table", {"data": 1}),
542 ("test_table", {"data": 2}),
543 ("test_table", {"data": 1}),
544 ("test_table", {"data": 2}),
545 ("test", {"data_1": 1, "data_2": 2}),
546 ("test", {"data_1": 2, "data_2": 1}),
547 ("test", {"data_1": 1, "data_2": 2}),
548 ("test", {"data_1": 2, "data_2": 1}),
549 ("test_table", {"data_1": 1, "data_2": 2}),
550 ("test_table", {"data_1": 2, "data_2": 1}),
551 ("test_table", {"data_1": 1, "data_2": 2}),
552 ("test_table", {"data_1": 2, "data_2": 1})])
553 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
554 returned_id
= db_memory_with_data
.create(table
, data
)
556 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
557 assert table
in db_memory_with_data
.db
558 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
561 data_inserted
['_id'] = returned_id
563 assert data_inserted
in db_memory_with_data
.db
[table
]
565 def test_create_with_exception(db_memory
):
567 data
= {"_id": 1, "data": 1}
569 db_memory
.db
= MagicMock()
570 db_memory
.db
.__contains
__.side_effect
= Exception()
572 with pytest
.raises(DbException
) as excinfo
:
573 db_memory
.create(table
, data
)
574 assert str(excinfo
.value
) == empty_exception_message()
575 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND