1 # Copyright 2018 Whitestack, LLC
2 # Copyright 2018 Telefonica S.A.
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
16 # For those usages not covered by the Apache License, Version 2.0 please
17 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
24 from unittest
.mock
import MagicMock
25 from osm_common
.dbbase
import DbException
26 from osm_common
.dbmemory
import DbMemory
28 __author__
= 'Eduardo Sousa <eduardosousa@av.it.pt>'
31 @pytest.fixture(scope
="function", params
=[True, False])
32 def db_memory(request
):
33 db
= DbMemory(lock
=request
.param
)
37 @pytest.fixture(scope
="function", params
=[True, False])
38 def db_memory_with_data(request
):
39 db
= DbMemory(lock
=request
.param
)
41 db
.create("test", {"_id": 1, "data": 1})
42 db
.create("test", {"_id": 2, "data": 2})
43 db
.create("test", {"_id": 3, "data": 3})
48 def empty_exception_message():
49 return 'database exception '
52 def get_one_exception_message(db_filter
):
53 return "database exception Not found entry with filter='{}'".format(db_filter
)
56 def get_one_multiple_exception_message(db_filter
):
57 return "database exception Found more than one entry with filter='{}'".format(db_filter
)
60 def del_one_exception_message(db_filter
):
61 return "database exception Not found entry with filter='{}'".format(db_filter
)
64 def replace_exception_message(value
):
65 return "database exception Not found entry with _id='{}'".format(value
)
68 def test_constructor():
70 assert db
.logger
== logging
.getLogger('db')
71 assert len(db
.db
) == 0
74 def test_constructor_with_logger():
75 logger_name
= 'db_local'
76 db
= DbMemory(logger_name
=logger_name
)
77 assert db
.logger
== logging
.getLogger(logger_name
)
78 assert len(db
.db
) == 0
81 def test_db_connect():
82 logger_name
= 'db_local'
83 config
= {'logger_name': logger_name
}
86 assert db
.logger
== logging
.getLogger(logger_name
)
87 assert len(db
.db
) == 0
90 def test_db_disconnect(db_memory
):
91 db_memory
.db_disconnect()
94 @pytest.mark
.parametrize("table, db_filter", [
97 ("test", {"data": 1}),
98 ("test", {"_id": 1, "data": 1})])
99 def test_get_list_with_empty_db(db_memory
, table
, db_filter
):
100 result
= db_memory
.get_list(table
, db_filter
)
101 assert len(result
) == 0
104 @pytest.mark
.parametrize("table, db_filter, expected_data", [
105 ("test", {}, [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
106 ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
107 ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
108 ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
109 ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
110 ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
111 ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
112 ("test", {"_id": 4}, []),
113 ("test", {"data": 4}, []),
114 ("test", {"_id": 4, "data": 4}, []),
115 ("test_table", {}, []),
116 ("test_table", {"_id": 1}, []),
117 ("test_table", {"data": 1}, []),
118 ("test_table", {"_id": 1, "data": 1}, [])])
119 def test_get_list_with_non_empty_db(db_memory_with_data
, table
, db_filter
, expected_data
):
120 result
= db_memory_with_data
.get_list(table
, db_filter
)
121 assert len(result
) == len(expected_data
)
122 for data
in expected_data
:
123 assert data
in result
126 def test_get_list_exception(db_memory_with_data
):
129 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
130 with pytest
.raises(DbException
) as excinfo
:
131 db_memory_with_data
.get_list(table
, db_filter
)
132 assert str(excinfo
.value
) == empty_exception_message()
133 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
136 @pytest.mark
.parametrize("table, db_filter, expected_data", [
137 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
138 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
139 ("test", {"_id": 3}, {"_id": 3, "data": 3}),
140 ("test", {"data": 1}, {"_id": 1, "data": 1}),
141 ("test", {"data": 2}, {"_id": 2, "data": 2}),
142 ("test", {"data": 3}, {"_id": 3, "data": 3}),
143 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
144 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
145 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3})])
146 def test_get_one(db_memory_with_data
, table
, db_filter
, expected_data
):
147 result
= db_memory_with_data
.get_one(table
, db_filter
)
148 assert result
== expected_data
149 assert len(db_memory_with_data
.db
) == 1
150 assert table
in db_memory_with_data
.db
151 assert len(db_memory_with_data
.db
[table
]) == 3
152 assert result
in db_memory_with_data
.db
[table
]
155 @pytest.mark
.parametrize("table, db_filter, expected_data", [
156 ("test", {}, {"_id": 1, "data": 1})])
157 def test_get_one_with_multiple_results(db_memory_with_data
, table
, db_filter
, expected_data
):
158 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_more
=False)
159 assert result
== expected_data
160 assert len(db_memory_with_data
.db
) == 1
161 assert table
in db_memory_with_data
.db
162 assert len(db_memory_with_data
.db
[table
]) == 3
163 assert result
in db_memory_with_data
.db
[table
]
166 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
169 with pytest
.raises(DbException
) as excinfo
:
170 db_memory_with_data
.get_one(table
, db_filter
)
171 assert str(excinfo
.value
) == (empty_exception_message() + get_one_multiple_exception_message(db_filter
))
172 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
175 @pytest.mark
.parametrize("table, db_filter", [
176 ("test", {"_id": 4}),
177 ("test", {"data": 4}),
178 ("test", {"_id": 4, "data": 4}),
179 ("test_table", {"_id": 4}),
180 ("test_table", {"data": 4}),
181 ("test_table", {"_id": 4, "data": 4})])
182 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
183 with pytest
.raises(DbException
) as excinfo
:
184 db_memory_with_data
.get_one(table
, db_filter
)
185 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(db_filter
))
186 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
189 @pytest.mark
.parametrize("table, db_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_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
197 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_empty
=False)
198 assert result
is None
201 @pytest.mark
.parametrize("table, db_filter", [
202 ("test", {"_id": 4}),
203 ("test", {"data": 4}),
204 ("test", {"_id": 4, "data": 4}),
205 ("test_table", {"_id": 4}),
206 ("test_table", {"data": 4}),
207 ("test_table", {"_id": 4, "data": 4})])
208 def test_get_one_with_empty_db_exception(db_memory
, table
, db_filter
):
209 with pytest
.raises(DbException
) as excinfo
:
210 db_memory
.get_one(table
, db_filter
)
211 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(db_filter
))
212 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
215 @pytest.mark
.parametrize("table, db_filter", [
216 ("test", {"_id": 4}),
217 ("test", {"data": 4}),
218 ("test", {"_id": 4, "data": 4}),
219 ("test_table", {"_id": 4}),
220 ("test_table", {"data": 4}),
221 ("test_table", {"_id": 4, "data": 4})])
222 def test_get_one_with_empty_db_none(db_memory
, table
, db_filter
):
223 result
= db_memory
.get_one(table
, db_filter
, fail_on_empty
=False)
224 assert result
is None
227 def test_get_one_generic_exception(db_memory_with_data
):
230 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
231 with pytest
.raises(DbException
) as excinfo
:
232 db_memory_with_data
.get_one(table
, db_filter
)
233 assert str(excinfo
.value
) == empty_exception_message()
234 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
237 @pytest.mark
.parametrize("table, db_filter, expected_data", [
239 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
240 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
241 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
242 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
243 def test_del_list_with_non_empty_db(db_memory_with_data
, table
, db_filter
, expected_data
):
244 result
= db_memory_with_data
.del_list(table
, db_filter
)
245 assert result
["deleted"] == (3 - len(expected_data
))
246 assert len(db_memory_with_data
.db
) == 1
247 assert table
in db_memory_with_data
.db
248 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
249 for data
in expected_data
:
250 assert data
in db_memory_with_data
.db
[table
]
253 @pytest.mark
.parametrize("table, db_filter", [
255 ("test", {"_id": 1}),
256 ("test", {"_id": 2}),
257 ("test", {"data": 1}),
258 ("test", {"data": 2}),
259 ("test", {"_id": 1, "data": 1}),
260 ("test", {"_id": 2, "data": 2})])
261 def test_del_list_with_empty_db(db_memory
, table
, db_filter
):
262 result
= db_memory
.del_list(table
, db_filter
)
263 assert result
['deleted'] == 0
266 def test_del_list_generic_exception(db_memory_with_data
):
269 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
270 with pytest
.raises(DbException
) as excinfo
:
271 db_memory_with_data
.del_list(table
, db_filter
)
272 assert str(excinfo
.value
) == empty_exception_message()
273 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
276 @pytest.mark
.parametrize("table, db_filter, data", [
277 ("test", {}, {"_id": 1, "data": 1}),
278 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
279 ("test", {"data": 1}, {"_id": 1, "data": 1}),
280 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
281 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
282 ("test", {"data": 2}, {"_id": 2, "data": 2}),
283 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
284 def test_del_one(db_memory_with_data
, table
, db_filter
, data
):
285 result
= db_memory_with_data
.del_one(table
, db_filter
)
286 assert result
== {"deleted": 1}
287 assert len(db_memory_with_data
.db
) == 1
288 assert table
in db_memory_with_data
.db
289 assert len(db_memory_with_data
.db
[table
]) == 2
290 assert data
not in db_memory_with_data
.db
[table
]
293 @pytest.mark
.parametrize("table, db_filter", [
295 ("test", {"_id": 1}),
296 ("test", {"_id": 2}),
297 ("test", {"data": 1}),
298 ("test", {"data": 2}),
299 ("test", {"_id": 1, "data": 1}),
300 ("test", {"_id": 2, "data": 2}),
302 ("test_table", {"_id": 1}),
303 ("test_table", {"_id": 2}),
304 ("test_table", {"data": 1}),
305 ("test_table", {"data": 2}),
306 ("test_table", {"_id": 1, "data": 1}),
307 ("test_table", {"_id": 2, "data": 2})])
308 def test_del_one_with_empty_db_exception(db_memory
, table
, db_filter
):
309 with pytest
.raises(DbException
) as excinfo
:
310 db_memory
.del_one(table
, db_filter
)
311 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(db_filter
))
312 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
315 @pytest.mark
.parametrize("table, db_filter", [
317 ("test", {"_id": 1}),
318 ("test", {"_id": 2}),
319 ("test", {"data": 1}),
320 ("test", {"data": 2}),
321 ("test", {"_id": 1, "data": 1}),
322 ("test", {"_id": 2, "data": 2}),
324 ("test_table", {"_id": 1}),
325 ("test_table", {"_id": 2}),
326 ("test_table", {"data": 1}),
327 ("test_table", {"data": 2}),
328 ("test_table", {"_id": 1, "data": 1}),
329 ("test_table", {"_id": 2, "data": 2})])
330 def test_del_one_with_empty_db_none(db_memory
, table
, db_filter
):
331 result
= db_memory
.del_one(table
, db_filter
, fail_on_empty
=False)
332 assert result
is None
335 @pytest.mark
.parametrize("table, db_filter", [
336 ("test", {"_id": 4}),
337 ("test", {"_id": 5}),
338 ("test", {"data": 4}),
339 ("test", {"data": 5}),
340 ("test", {"_id": 1, "data": 2}),
341 ("test", {"_id": 2, "data": 3}),
343 ("test_table", {"_id": 1}),
344 ("test_table", {"_id": 2}),
345 ("test_table", {"data": 1}),
346 ("test_table", {"data": 2}),
347 ("test_table", {"_id": 1, "data": 1}),
348 ("test_table", {"_id": 2, "data": 2})])
349 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
350 with pytest
.raises(DbException
) as excinfo
:
351 db_memory_with_data
.del_one(table
, db_filter
)
352 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(db_filter
))
353 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
356 @pytest.mark
.parametrize("table, db_filter", [
357 ("test", {"_id": 4}),
358 ("test", {"_id": 5}),
359 ("test", {"data": 4}),
360 ("test", {"data": 5}),
361 ("test", {"_id": 1, "data": 2}),
362 ("test", {"_id": 2, "data": 3}),
364 ("test_table", {"_id": 1}),
365 ("test_table", {"_id": 2}),
366 ("test_table", {"data": 1}),
367 ("test_table", {"data": 2}),
368 ("test_table", {"_id": 1, "data": 1}),
369 ("test_table", {"_id": 2, "data": 2})])
370 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
371 result
= db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=False)
372 assert result
is None
375 @pytest.mark
.parametrize("fail_on_empty", [
378 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
381 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
382 with pytest
.raises(DbException
) as excinfo
:
383 db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=fail_on_empty
)
384 assert str(excinfo
.value
) == empty_exception_message()
385 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
388 @pytest.mark
.parametrize("table, _id, indata", [
389 ("test", 1, {"_id": 1, "data": 42}),
390 ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
391 ("test", 1, {"_id": 1}),
392 ("test", 2, {"_id": 2, "data": 42}),
393 ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
394 ("test", 2, {"_id": 2}),
395 ("test", 3, {"_id": 3, "data": 42}),
396 ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
397 ("test", 3, {"_id": 3})])
398 def test_replace(db_memory_with_data
, table
, _id
, indata
):
399 result
= db_memory_with_data
.replace(table
, _id
, indata
)
400 assert result
== {"updated": 1}
401 assert len(db_memory_with_data
.db
) == 1
402 assert table
in db_memory_with_data
.db
403 assert len(db_memory_with_data
.db
[table
]) == 3
404 assert indata
in db_memory_with_data
.db
[table
]
407 @pytest.mark
.parametrize("table, _id, indata", [
408 ("test", 1, {"_id": 1, "data": 42}),
409 ("test", 2, {"_id": 2}),
410 ("test", 3, {"_id": 3})])
411 def test_replace_without_data_exception(db_memory
, table
, _id
, indata
):
412 with pytest
.raises(DbException
) as excinfo
:
413 db_memory
.replace(table
, _id
, indata
, fail_on_empty
=True)
414 assert str(excinfo
.value
) == (replace_exception_message(_id
))
415 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
418 @pytest.mark
.parametrize("table, _id, indata", [
419 ("test", 1, {"_id": 1, "data": 42}),
420 ("test", 2, {"_id": 2}),
421 ("test", 3, {"_id": 3})])
422 def test_replace_without_data_none(db_memory
, table
, _id
, indata
):
423 result
= db_memory
.replace(table
, _id
, indata
, fail_on_empty
=False)
424 assert result
is None
427 @pytest.mark
.parametrize("table, _id, indata", [
428 ("test", 11, {"_id": 11, "data": 42}),
429 ("test", 12, {"_id": 12}),
430 ("test", 33, {"_id": 33})])
431 def test_replace_with_data_exception(db_memory_with_data
, table
, _id
, indata
):
432 with pytest
.raises(DbException
) as excinfo
:
433 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=True)
434 assert str(excinfo
.value
) == (replace_exception_message(_id
))
435 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
438 @pytest.mark
.parametrize("table, _id, indata", [
439 ("test", 11, {"_id": 11, "data": 42}),
440 ("test", 12, {"_id": 12}),
441 ("test", 33, {"_id": 33})])
442 def test_replace_with_data_none(db_memory_with_data
, table
, _id
, indata
):
443 result
= db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=False)
444 assert result
is None
447 @pytest.mark
.parametrize("fail_on_empty", [
450 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
453 indata
= {'_id': 1, 'data': 1}
454 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
455 with pytest
.raises(DbException
) as excinfo
:
456 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=fail_on_empty
)
457 assert str(excinfo
.value
) == empty_exception_message()
458 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
461 @pytest.mark
.parametrize("table, id, data", [
462 ("test", "1", {"data": 1}),
463 ("test", "1", {"data": 2}),
464 ("test", "2", {"data": 1}),
465 ("test", "2", {"data": 2}),
466 ("test_table", "1", {"data": 1}),
467 ("test_table", "1", {"data": 2}),
468 ("test_table", "2", {"data": 1}),
469 ("test_table", "2", {"data": 2}),
470 ("test", "1", {"data_1": 1, "data_2": 2}),
471 ("test", "1", {"data_1": 2, "data_2": 1}),
472 ("test", "2", {"data_1": 1, "data_2": 2}),
473 ("test", "2", {"data_1": 2, "data_2": 1}),
474 ("test_table", "1", {"data_1": 1, "data_2": 2}),
475 ("test_table", "1", {"data_1": 2, "data_2": 1}),
476 ("test_table", "2", {"data_1": 1, "data_2": 2}),
477 ("test_table", "2", {"data_1": 2, "data_2": 1})])
478 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
479 data_to_insert
= data
480 data_to_insert
['_id'] = id
481 returned_id
= db_memory
.create(table
, data_to_insert
)
482 assert returned_id
== id
483 assert len(db_memory
.db
) == 1
484 assert table
in db_memory
.db
485 assert len(db_memory
.db
[table
]) == 1
486 assert data_to_insert
in db_memory
.db
[table
]
489 @pytest.mark
.parametrize("table, id, data", [
490 ("test", "4", {"data": 1}),
491 ("test", "5", {"data": 2}),
492 ("test", "4", {"data": 1}),
493 ("test", "5", {"data": 2}),
494 ("test_table", "4", {"data": 1}),
495 ("test_table", "5", {"data": 2}),
496 ("test_table", "4", {"data": 1}),
497 ("test_table", "5", {"data": 2}),
498 ("test", "4", {"data_1": 1, "data_2": 2}),
499 ("test", "5", {"data_1": 2, "data_2": 1}),
500 ("test", "4", {"data_1": 1, "data_2": 2}),
501 ("test", "5", {"data_1": 2, "data_2": 1}),
502 ("test_table", "4", {"data_1": 1, "data_2": 2}),
503 ("test_table", "5", {"data_1": 2, "data_2": 1}),
504 ("test_table", "4", {"data_1": 1, "data_2": 2}),
505 ("test_table", "5", {"data_1": 2, "data_2": 1})])
506 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
507 data_to_insert
= data
508 data_to_insert
['_id'] = id
509 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
510 assert returned_id
== id
511 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
512 assert table
in db_memory_with_data
.db
513 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
514 assert data_to_insert
in db_memory_with_data
.db
[table
]
517 @pytest.mark
.parametrize("table, data", [
518 ("test", {"data": 1}),
519 ("test", {"data": 2}),
520 ("test", {"data": 1}),
521 ("test", {"data": 2}),
522 ("test_table", {"data": 1}),
523 ("test_table", {"data": 2}),
524 ("test_table", {"data": 1}),
525 ("test_table", {"data": 2}),
526 ("test", {"data_1": 1, "data_2": 2}),
527 ("test", {"data_1": 2, "data_2": 1}),
528 ("test", {"data_1": 1, "data_2": 2}),
529 ("test", {"data_1": 2, "data_2": 1}),
530 ("test_table", {"data_1": 1, "data_2": 2}),
531 ("test_table", {"data_1": 2, "data_2": 1}),
532 ("test_table", {"data_1": 1, "data_2": 2}),
533 ("test_table", {"data_1": 2, "data_2": 1})])
534 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
535 returned_id
= db_memory
.create(table
, data
)
536 assert len(db_memory
.db
) == 1
537 assert table
in db_memory
.db
538 assert len(db_memory
.db
[table
]) == 1
540 data_inserted
['_id'] = returned_id
541 assert data_inserted
in db_memory
.db
[table
]
544 @pytest.mark
.parametrize("table, data", [
545 ("test", {"data": 1}),
546 ("test", {"data": 2}),
547 ("test", {"data": 1}),
548 ("test", {"data": 2}),
549 ("test_table", {"data": 1}),
550 ("test_table", {"data": 2}),
551 ("test_table", {"data": 1}),
552 ("test_table", {"data": 2}),
553 ("test", {"data_1": 1, "data_2": 2}),
554 ("test", {"data_1": 2, "data_2": 1}),
555 ("test", {"data_1": 1, "data_2": 2}),
556 ("test", {"data_1": 2, "data_2": 1}),
557 ("test_table", {"data_1": 1, "data_2": 2}),
558 ("test_table", {"data_1": 2, "data_2": 1}),
559 ("test_table", {"data_1": 1, "data_2": 2}),
560 ("test_table", {"data_1": 2, "data_2": 1})])
561 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
562 returned_id
= db_memory_with_data
.create(table
, data
)
563 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
564 assert table
in db_memory_with_data
.db
565 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
567 data_inserted
['_id'] = returned_id
568 assert data_inserted
in db_memory_with_data
.db
[table
]
571 def test_create_with_exception(db_memory
):
573 data
= {"_id": 1, "data": 1}
574 db_memory
.db
= MagicMock()
575 db_memory
.db
.__contains
__.side_effect
= Exception()
576 with pytest
.raises(DbException
) as excinfo
:
577 db_memory
.create(table
, data
)
578 assert str(excinfo
.value
) == empty_exception_message()
579 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND