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 @pytest.mark
.parametrize("table, filter", [
68 ("test", {"data": 1}),
69 ("test", {"_id": 1, "data": 1})])
70 def test_get_list_with_empty_db(db_memory
, table
, filter):
71 result
= db_memory
.get_list(table
, filter)
73 assert len(result
) == 0
75 @pytest.mark
.parametrize("table, filter, expected_data", [
76 ("test", {}, [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
77 ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
78 ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
79 ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
80 ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
81 ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
82 ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
83 ("test", {"_id": 4}, []),
84 ("test", {"data": 4}, []),
85 ("test", {"_id": 4, "data": 4}, []),
86 ("test_table", {}, []),
87 ("test_table", {"_id": 1}, []),
88 ("test_table", {"data": 1}, []),
89 ("test_table", {"_id": 1, "data": 1}, [])])
90 def test_get_list_with_non_empty_db(db_memory_with_data
, table
, filter, expected_data
):
91 result
= db_memory_with_data
.get_list(table
, filter)
93 assert len(result
) == len(expected_data
)
94 for data
in expected_data
:
97 def test_get_list_exception(db_memory_with_data
):
101 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
103 with pytest
.raises(DbException
) as excinfo
:
104 db_memory_with_data
.get_list(table
, filter)
105 assert str(excinfo
.value
) == empty_exception_message()
106 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
108 @pytest.mark
.parametrize("table, filter, expected_data", [
109 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
110 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
111 ("test", {"_id": 3}, {"_id": 3, "data": 3}),
112 ("test", {"data": 1}, {"_id": 1, "data": 1}),
113 ("test", {"data": 2}, {"_id": 2, "data": 2}),
114 ("test", {"data": 3}, {"_id": 3, "data": 3}),
115 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
116 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
117 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3})])
118 def test_get_one(db_memory_with_data
, table
, filter, expected_data
):
119 result
= db_memory_with_data
.get_one(table
, filter)
121 assert result
== expected_data
122 assert len(db_memory_with_data
.db
) == 1
123 assert table
in db_memory_with_data
.db
124 assert len(db_memory_with_data
.db
[table
]) == 3
125 assert result
in db_memory_with_data
.db
[table
]
127 @pytest.mark
.parametrize("table, filter, expected_data", [
128 ("test", {}, {"_id": 1, "data": 1})])
129 def test_get_one_with_multiple_results(db_memory_with_data
, table
, filter, expected_data
):
130 result
= db_memory_with_data
.get_one(table
, filter, fail_on_more
=False)
132 assert result
== expected_data
133 assert len(db_memory_with_data
.db
) == 1
134 assert table
in db_memory_with_data
.db
135 assert len(db_memory_with_data
.db
[table
]) == 3
136 assert result
in db_memory_with_data
.db
[table
]
138 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
142 with pytest
.raises(DbException
) as excinfo
:
143 db_memory_with_data
.get_one(table
, filter)
145 assert str(excinfo
.value
) == (empty_exception_message() + get_one_multiple_exception_message(filter))
146 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
148 @pytest.mark
.parametrize("table, filter", [
149 ("test", {"_id": 4}),
150 ("test", {"data": 4}),
151 ("test", {"_id": 4, "data": 4}),
152 ("test_table", {"_id": 4}),
153 ("test_table", {"data": 4}),
154 ("test_table", {"_id": 4, "data": 4})])
155 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, filter):
156 with pytest
.raises(DbException
) as excinfo
:
157 db_memory_with_data
.get_one(table
, filter)
158 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(filter))
159 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
161 @pytest.mark
.parametrize("table, filter", [
162 ("test", {"_id": 4}),
163 ("test", {"data": 4}),
164 ("test", {"_id": 4, "data": 4}),
165 ("test_table", {"_id": 4}),
166 ("test_table", {"data": 4}),
167 ("test_table", {"_id": 4, "data": 4})])
168 def test_get_one_with_non_empty_db_none(db_memory_with_data
, table
, filter):
169 result
= db_memory_with_data
.get_one(table
, filter, fail_on_empty
=False)
171 assert result
== None
173 @pytest.mark
.parametrize("table, filter", [
174 ("test", {"_id": 4}),
175 ("test", {"data": 4}),
176 ("test", {"_id": 4, "data": 4}),
177 ("test_table", {"_id": 4}),
178 ("test_table", {"data": 4}),
179 ("test_table", {"_id": 4, "data": 4})])
180 def test_get_one_with_empty_db_exception(db_memory
, table
, filter):
181 with pytest
.raises(DbException
) as excinfo
:
182 db_memory
.get_one(table
, filter)
183 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(filter))
184 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
186 @pytest.mark
.parametrize("table, filter", [
187 ("test", {"_id": 4}),
188 ("test", {"data": 4}),
189 ("test", {"_id": 4, "data": 4}),
190 ("test_table", {"_id": 4}),
191 ("test_table", {"data": 4}),
192 ("test_table", {"_id": 4, "data": 4})])
193 def test_get_one_with_empty_db_none(db_memory
, table
, filter):
194 result
= db_memory
.get_one(table
, filter, fail_on_empty
=False)
196 assert result
== None
198 def test_get_one_generic_exception(db_memory_with_data
):
202 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
204 with pytest
.raises(DbException
) as excinfo
:
205 db_memory_with_data
.get_one(table
, filter)
206 assert str(excinfo
.value
) == empty_exception_message()
207 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
209 @pytest.mark
.parametrize("table, filter, expected_data", [
211 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
212 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
213 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
214 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
215 def test_del_list_with_non_empty_db(db_memory_with_data
, table
, filter, expected_data
):
216 result
= db_memory_with_data
.del_list(table
, filter)
218 assert result
["deleted"] == (3 - len(expected_data
))
219 assert len(db_memory_with_data
.db
) == 1
220 assert table
in db_memory_with_data
.db
221 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
222 for data
in expected_data
:
223 assert data
in db_memory_with_data
.db
[table
]
225 @pytest.mark
.parametrize("table, filter", [
227 ("test", {"_id": 1}),
228 ("test", {"_id": 2}),
229 ("test", {"data": 1}),
230 ("test", {"data": 2}),
231 ("test", {"_id": 1, "data": 1}),
232 ("test", {"_id": 2, "data": 2})])
233 def test_del_list_with_empty_db(db_memory
, table
, filter):
234 result
= db_memory
.del_list(table
, filter)
235 assert result
['deleted'] == 0
237 def test_del_list_generic_exception(db_memory_with_data
):
241 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
243 with pytest
.raises(DbException
) as excinfo
:
244 db_memory_with_data
.del_list(table
, filter)
245 assert str(excinfo
.value
) == empty_exception_message()
246 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
248 @pytest.mark
.parametrize("table, filter, data", [
249 ("test", {}, {"_id": 1, "data": 1}),
250 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
251 ("test", {"data": 1}, {"_id": 1, "data": 1}),
252 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
253 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
254 ("test", {"data": 2}, {"_id": 2, "data": 2}),
255 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
256 def test_del_one(db_memory_with_data
, table
, filter, data
):
257 result
= db_memory_with_data
.del_one(table
, filter)
259 assert result
== {"deleted": 1}
260 assert len(db_memory_with_data
.db
) == 1
261 assert table
in db_memory_with_data
.db
262 assert len(db_memory_with_data
.db
[table
]) == 2
263 assert data
not in db_memory_with_data
.db
[table
]
265 @pytest.mark
.parametrize("table, filter", [
267 ("test", {"_id": 1}),
268 ("test", {"_id": 2}),
269 ("test", {"data": 1}),
270 ("test", {"data": 2}),
271 ("test", {"_id": 1, "data": 1}),
272 ("test", {"_id": 2, "data": 2}),
274 ("test_table", {"_id": 1}),
275 ("test_table", {"_id": 2}),
276 ("test_table", {"data": 1}),
277 ("test_table", {"data": 2}),
278 ("test_table", {"_id": 1, "data": 1}),
279 ("test_table", {"_id": 2, "data": 2})])
280 def test_del_one_with_empty_db_exception(db_memory
, table
, filter):
281 with pytest
.raises(DbException
) as excinfo
:
282 db_memory
.del_one(table
, filter)
283 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(filter))
284 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
286 @pytest.mark
.parametrize("table, filter", [
288 ("test", {"_id": 1}),
289 ("test", {"_id": 2}),
290 ("test", {"data": 1}),
291 ("test", {"data": 2}),
292 ("test", {"_id": 1, "data": 1}),
293 ("test", {"_id": 2, "data": 2}),
295 ("test_table", {"_id": 1}),
296 ("test_table", {"_id": 2}),
297 ("test_table", {"data": 1}),
298 ("test_table", {"data": 2}),
299 ("test_table", {"_id": 1, "data": 1}),
300 ("test_table", {"_id": 2, "data": 2})])
301 def test_del_one_with_empty_db_none(db_memory
, table
, filter):
302 result
= db_memory
.del_one(table
, filter, fail_on_empty
=False)
304 assert result
== None
306 @pytest.mark
.parametrize("table, filter", [
307 ("test", {"_id": 4}),
308 ("test", {"_id": 5}),
309 ("test", {"data": 4}),
310 ("test", {"data": 5}),
311 ("test", {"_id": 1, "data": 2}),
312 ("test", {"_id": 2, "data": 3}),
314 ("test_table", {"_id": 1}),
315 ("test_table", {"_id": 2}),
316 ("test_table", {"data": 1}),
317 ("test_table", {"data": 2}),
318 ("test_table", {"_id": 1, "data": 1}),
319 ("test_table", {"_id": 2, "data": 2})])
320 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, filter):
321 with pytest
.raises(DbException
) as excinfo
:
322 db_memory_with_data
.del_one(table
, filter)
323 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(filter))
324 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
326 @pytest.mark
.parametrize("table, filter", [
327 ("test", {"_id": 4}),
328 ("test", {"_id": 5}),
329 ("test", {"data": 4}),
330 ("test", {"data": 5}),
331 ("test", {"_id": 1, "data": 2}),
332 ("test", {"_id": 2, "data": 3}),
334 ("test_table", {"_id": 1}),
335 ("test_table", {"_id": 2}),
336 ("test_table", {"data": 1}),
337 ("test_table", {"data": 2}),
338 ("test_table", {"_id": 1, "data": 1}),
339 ("test_table", {"_id": 2, "data": 2})])
340 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, filter):
341 result
= db_memory_with_data
.del_one(table
, filter, fail_on_empty
=False)
343 assert result
== None
345 @pytest.mark
.parametrize("fail_on_empty", [
348 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
352 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
354 with pytest
.raises(DbException
) as excinfo
:
355 db_memory_with_data
.del_one(table
, filter, fail_on_empty
=fail_on_empty
)
356 assert str(excinfo
.value
) == empty_exception_message()
357 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
359 @pytest.mark
.parametrize("table, filter, indata", [
360 ("test", {}, {"_id": 1, "data": 42}),
361 ("test", {}, {"_id": 3, "data": 42}),
362 ("test", {"_id": 1}, {"_id": 3, "data": 42}),
363 ("test", {"_id": 3}, {"_id": 3, "data": 42}),
364 ("test", {"data": 1}, {"_id": 3, "data": 42}),
365 ("test", {"data": 3}, {"_id": 3, "data": 42}),
366 ("test", {"_id": 1, "data": 1}, {"_id": 3, "data": 42}),
367 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 42})])
368 def test_replace(db_memory_with_data
, table
, filter, indata
):
369 result
= db_memory_with_data
.replace(table
, filter, indata
)
371 assert result
== {"updated": 1}
372 assert len(db_memory_with_data
.db
) == 1
373 assert table
in db_memory_with_data
.db
374 assert len(db_memory_with_data
.db
[table
]) == 3
375 assert indata
in db_memory_with_data
.db
[table
]
377 @pytest.mark
.parametrize("table, filter, indata", [
378 ("test", {}, {'_id': 1, 'data': 1}),
379 ("test", {}, {'_id': 2, 'data': 1}),
380 ("test", {}, {'_id': 1, 'data': 2}),
381 ("test", {'_id': 1}, {'_id': 1, 'data': 1}),
382 ("test", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
383 ("test_table", {}, {'_id': 1, 'data': 1}),
384 ("test_table", {}, {'_id': 2, 'data': 1}),
385 ("test_table", {}, {'_id': 1, 'data': 2}),
386 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
387 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
388 def test_replace_without_data_exception(db_memory
, table
, filter, indata
):
389 with pytest
.raises(DbException
) as excinfo
:
390 db_memory
.replace(table
, filter, indata
, fail_on_empty
=True)
391 assert str(excinfo
.value
) == (empty_exception_message() + replace_exception_message(filter))
392 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
394 @pytest.mark
.parametrize("table, filter, indata", [
395 ("test", {}, {'_id': 1, 'data': 1}),
396 ("test", {}, {'_id': 2, 'data': 1}),
397 ("test", {}, {'_id': 1, 'data': 2}),
398 ("test", {'_id': 1}, {'_id': 1, 'data': 1}),
399 ("test", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1}),
400 ("test_table", {}, {'_id': 1, 'data': 1}),
401 ("test_table", {}, {'_id': 2, 'data': 1}),
402 ("test_table", {}, {'_id': 1, 'data': 2}),
403 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
404 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
405 def test_replace_without_data_none(db_memory
, table
, filter, indata
):
406 result
= db_memory
.replace(table
, filter, indata
, fail_on_empty
=False)
407 assert result
== None
409 @pytest.mark
.parametrize("table, filter, indata", [
410 ("test_table", {}, {'_id': 1, 'data': 1}),
411 ("test_table", {}, {'_id': 2, 'data': 1}),
412 ("test_table", {}, {'_id': 1, 'data': 2}),
413 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
414 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
415 def test_replace_with_data_exception(db_memory_with_data
, table
, filter, indata
):
416 with pytest
.raises(DbException
) as excinfo
:
417 db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=True)
418 assert str(excinfo
.value
) == (empty_exception_message() + replace_exception_message(filter))
419 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
421 @pytest.mark
.parametrize("table, filter, indata", [
422 ("test_table", {}, {'_id': 1, 'data': 1}),
423 ("test_table", {}, {'_id': 2, 'data': 1}),
424 ("test_table", {}, {'_id': 1, 'data': 2}),
425 ("test_table", {'_id': 1}, {'_id': 1, 'data': 1}),
426 ("test_table", {'_id': 1, 'data': 1}, {'_id': 1, 'data': 1})])
427 def test_replace_with_data_none(db_memory_with_data
, table
, filter, indata
):
428 result
= db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=False)
429 assert result
== None
431 @pytest.mark
.parametrize("fail_on_empty", [
434 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
437 indata
= {'_id': 1, 'data': 1}
439 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
441 with pytest
.raises(DbException
) as excinfo
:
442 db_memory_with_data
.replace(table
, filter, indata
, fail_on_empty
=fail_on_empty
)
443 assert str(excinfo
.value
) == empty_exception_message()
444 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
446 @pytest.mark
.parametrize("table, id, data", [
447 ("test", "1", {"data": 1}),
448 ("test", "1", {"data": 2}),
449 ("test", "2", {"data": 1}),
450 ("test", "2", {"data": 2}),
451 ("test_table", "1", {"data": 1}),
452 ("test_table", "1", {"data": 2}),
453 ("test_table", "2", {"data": 1}),
454 ("test_table", "2", {"data": 2}),
455 ("test", "1", {"data_1": 1, "data_2": 2}),
456 ("test", "1", {"data_1": 2, "data_2": 1}),
457 ("test", "2", {"data_1": 1, "data_2": 2}),
458 ("test", "2", {"data_1": 2, "data_2": 1}),
459 ("test_table", "1", {"data_1": 1, "data_2": 2}),
460 ("test_table", "1", {"data_1": 2, "data_2": 1}),
461 ("test_table", "2", {"data_1": 1, "data_2": 2}),
462 ("test_table", "2", {"data_1": 2, "data_2": 1})])
463 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
464 data_to_insert
= data
465 data_to_insert
['_id'] = id
467 returned_id
= db_memory
.create(table
, data_to_insert
)
469 assert returned_id
== id
470 assert len(db_memory
.db
) == 1
471 assert table
in db_memory
.db
472 assert len(db_memory
.db
[table
]) == 1
473 assert data_to_insert
in db_memory
.db
[table
]
475 @pytest.mark
.parametrize("table, id, data", [
476 ("test", "4", {"data": 1}),
477 ("test", "5", {"data": 2}),
478 ("test", "4", {"data": 1}),
479 ("test", "5", {"data": 2}),
480 ("test_table", "4", {"data": 1}),
481 ("test_table", "5", {"data": 2}),
482 ("test_table", "4", {"data": 1}),
483 ("test_table", "5", {"data": 2}),
484 ("test", "4", {"data_1": 1, "data_2": 2}),
485 ("test", "5", {"data_1": 2, "data_2": 1}),
486 ("test", "4", {"data_1": 1, "data_2": 2}),
487 ("test", "5", {"data_1": 2, "data_2": 1}),
488 ("test_table", "4", {"data_1": 1, "data_2": 2}),
489 ("test_table", "5", {"data_1": 2, "data_2": 1}),
490 ("test_table", "4", {"data_1": 1, "data_2": 2}),
491 ("test_table", "5", {"data_1": 2, "data_2": 1})])
492 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
493 data_to_insert
= data
494 data_to_insert
['_id'] = id
496 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
498 assert returned_id
== id
499 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
500 assert table
in db_memory_with_data
.db
501 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
502 assert data_to_insert
in db_memory_with_data
.db
[table
]
504 @pytest.mark
.parametrize("table, data", [
505 ("test", {"data": 1}),
506 ("test", {"data": 2}),
507 ("test", {"data": 1}),
508 ("test", {"data": 2}),
509 ("test_table", {"data": 1}),
510 ("test_table", {"data": 2}),
511 ("test_table", {"data": 1}),
512 ("test_table", {"data": 2}),
513 ("test", {"data_1": 1, "data_2": 2}),
514 ("test", {"data_1": 2, "data_2": 1}),
515 ("test", {"data_1": 1, "data_2": 2}),
516 ("test", {"data_1": 2, "data_2": 1}),
517 ("test_table", {"data_1": 1, "data_2": 2}),
518 ("test_table", {"data_1": 2, "data_2": 1}),
519 ("test_table", {"data_1": 1, "data_2": 2}),
520 ("test_table", {"data_1": 2, "data_2": 1})])
521 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
522 returned_id
= db_memory
.create(table
, data
)
524 assert len(db_memory
.db
) == 1
525 assert table
in db_memory
.db
526 assert len(db_memory
.db
[table
]) == 1
529 data_inserted
['_id'] = returned_id
531 assert data_inserted
in db_memory
.db
[table
]
533 @pytest.mark
.parametrize("table, data", [
534 ("test", {"data": 1}),
535 ("test", {"data": 2}),
536 ("test", {"data": 1}),
537 ("test", {"data": 2}),
538 ("test_table", {"data": 1}),
539 ("test_table", {"data": 2}),
540 ("test_table", {"data": 1}),
541 ("test_table", {"data": 2}),
542 ("test", {"data_1": 1, "data_2": 2}),
543 ("test", {"data_1": 2, "data_2": 1}),
544 ("test", {"data_1": 1, "data_2": 2}),
545 ("test", {"data_1": 2, "data_2": 1}),
546 ("test_table", {"data_1": 1, "data_2": 2}),
547 ("test_table", {"data_1": 2, "data_2": 1}),
548 ("test_table", {"data_1": 1, "data_2": 2}),
549 ("test_table", {"data_1": 2, "data_2": 1})])
550 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
551 returned_id
= db_memory_with_data
.create(table
, data
)
553 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
554 assert table
in db_memory_with_data
.db
555 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
558 data_inserted
['_id'] = returned_id
560 assert data_inserted
in db_memory_with_data
.db
[table
]
562 def test_create_with_exception(db_memory
):
564 data
= {"_id": 1, "data": 1}
566 db_memory
.db
= MagicMock()
567 db_memory
.db
.__contains
__.side_effect
= Exception()
569 with pytest
.raises(DbException
) as excinfo
:
570 db_memory
.create(table
, data
)
571 assert str(excinfo
.value
) == empty_exception_message()
572 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND