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
19 from copy
import deepcopy
23 from unittest
.mock
import MagicMock
, Mock
25 from osm_common
.dbbase
import DbException
26 from osm_common
.dbmemory
import DbMemory
29 __author__
= "Eduardo Sousa <eduardosousa@av.it.pt>"
32 @pytest.fixture(scope
="function", params
=[True, False])
33 def db_memory(request
):
34 db
= DbMemory(lock
=request
.param
)
38 @pytest.fixture(scope
="function", params
=[True, False])
39 def db_memory_with_data(request
):
40 db
= DbMemory(lock
=request
.param
)
42 db
.create("test", {"_id": 1, "data": 1})
43 db
.create("test", {"_id": 2, "data": 2})
44 db
.create("test", {"_id": 3, "data": 3})
49 @pytest.fixture(scope
="function")
50 def db_memory_with_many_data(request
):
51 db
= DbMemory(lock
=False)
58 "data": {"data2": {"data3": 1}},
64 "data": {"data2": {"data3": 2}},
68 {"_id": 3, "data": {"data2": {"data3": 3}}, "list": [{"a": 3}]},
69 {"_id": 4, "data": {"data2": {"data3": 4}}, "list": [{"a": 4}, {"a": 0}]},
70 {"_id": 5, "data": {"data2": {"data3": 5}}, "list": [{"a": 5}]},
71 {"_id": 6, "data": {"data2": {"data3": 6}}, "list": [{"0": {"a": 1}}]},
72 {"_id": 7, "data": {"data2": {"data3": 7}}, "0": {"a": 0}},
76 {"a": 3, "b": 0, "c": [{"a": 3, "b": 1}, {"a": 0, "b": "v"}]},
85 def empty_exception_message():
86 return "database exception "
89 def get_one_exception_message(db_filter
):
90 return "database exception Not found entry with filter='{}'".format(db_filter
)
93 def get_one_multiple_exception_message(db_filter
):
94 return "database exception Found more than one entry with filter='{}'".format(
99 def del_one_exception_message(db_filter
):
100 return "database exception Not found entry with filter='{}'".format(db_filter
)
103 def replace_exception_message(value
):
104 return "database exception Not found entry with _id='{}'".format(value
)
107 def test_constructor():
109 assert db
.logger
== logging
.getLogger("db")
113 def test_constructor_with_logger():
114 logger_name
= "db_local"
115 db
= DbMemory(logger_name
=logger_name
)
116 assert db
.logger
== logging
.getLogger(logger_name
)
120 def test_db_connect():
121 logger_name
= "db_local"
122 config
= {"logger_name": logger_name
}
124 db
.db_connect(config
)
125 assert db
.logger
== logging
.getLogger(logger_name
)
129 def test_db_disconnect(db_memory
):
130 db_memory
.db_disconnect()
133 @pytest.mark
.parametrize(
137 ("test", {"_id": 1}),
138 ("test", {"data": 1}),
139 ("test", {"_id": 1, "data": 1}),
142 def test_get_list_with_empty_db(db_memory
, table
, db_filter
):
143 result
= db_memory
.get_list(table
, db_filter
)
144 assert len(result
) == 0
147 @pytest.mark
.parametrize(
148 "table, db_filter, expected_data",
153 [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}],
155 ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
156 ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
157 ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
158 ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
159 ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
160 ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
161 ("test", {"_id": 4}, []),
162 ("test", {"data": 4}, []),
163 ("test", {"_id": 4, "data": 4}, []),
164 ("test_table", {}, []),
165 ("test_table", {"_id": 1}, []),
166 ("test_table", {"data": 1}, []),
167 ("test_table", {"_id": 1, "data": 1}, []),
170 def test_get_list_with_non_empty_db(
171 db_memory_with_data
, table
, db_filter
, expected_data
173 result
= db_memory_with_data
.get_list(table
, db_filter
)
174 assert len(result
) == len(expected_data
)
175 for data
in expected_data
:
176 assert data
in result
179 def test_get_list_exception(db_memory_with_data
):
182 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
183 with pytest
.raises(DbException
) as excinfo
:
184 db_memory_with_data
.get_list(table
, db_filter
)
185 assert str(excinfo
.value
) == empty_exception_message()
186 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
189 @pytest.mark
.parametrize(
190 "table, db_filter, expected_data",
192 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
193 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
194 ("test", {"_id": 3}, {"_id": 3, "data": 3}),
195 ("test", {"data": 1}, {"_id": 1, "data": 1}),
196 ("test", {"data": 2}, {"_id": 2, "data": 2}),
197 ("test", {"data": 3}, {"_id": 3, "data": 3}),
198 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
199 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
200 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3}),
203 def test_get_one(db_memory_with_data
, table
, db_filter
, expected_data
):
204 result
= db_memory_with_data
.get_one(table
, db_filter
)
205 assert result
== expected_data
206 assert len(db_memory_with_data
.db
) == 1
207 assert table
in db_memory_with_data
.db
208 assert len(db_memory_with_data
.db
[table
]) == 3
209 assert result
in db_memory_with_data
.db
[table
]
212 @pytest.mark
.parametrize(
213 "db_filter, expected_ids",
215 ({}, [1, 2, 3, 4, 5, 6, 7, 8]),
217 ({"data.data2.data3": 2}, [2]),
218 ({"data.data2.data3.eq": 2}, [2]),
219 ({"data.data2.data3": [2]}, [2]),
220 ({"data.data2.data3.cont": [2]}, [2]),
221 ({"data.data2.data3.neq": 2}, [1, 3, 4, 5, 6, 7, 8]),
222 ({"data.data2.data3.neq": [2]}, [1, 3, 4, 5, 6, 7, 8]),
223 ({"data.data2.data3.ncont": [2]}, [1, 3, 4, 5, 6, 7, 8]),
224 ({"data.data2.data3": [2, 3]}, [2, 3]),
225 ({"data.data2.data3.gt": 4}, [5, 6, 7]),
226 ({"data.data2.data3.gte": 4}, [4, 5, 6, 7]),
227 ({"data.data2.data3.lt": 4}, [1, 2, 3]),
228 ({"data.data2.data3.lte": 4}, [1, 2, 3, 4]),
229 ({"data.data2.data3.lte": 4.5}, [1, 2, 3, 4]),
230 ({"data.data2.data3.gt": "text"}, []),
231 ({"nonexist.nonexist": "4"}, []),
232 ({"nonexist.nonexist": None}, [1, 2, 3, 4, 5, 6, 7, 8]),
233 ({"nonexist.nonexist.neq": "4"}, [1, 2, 3, 4, 5, 6, 7, 8]),
234 ({"nonexist.nonexist.neq": None}, []),
235 ({"text.eq": "sometext"}, [1]),
236 ({"text.neq": "sometext"}, [2, 3, 4, 5, 6, 7, 8]),
237 ({"text.eq": "somet"}, []),
238 ({"text.gte": "a"}, [1]),
239 ({"text.gte": "somet"}, [1]),
240 ({"text.gte": "sometext"}, [1]),
241 ({"text.lt": "somet"}, []),
242 ({"data.data2.data3": 2, "data.data2.data4": None}, [2]),
243 ({"data.data2.data3": 2, "data.data2.data4": 5}, []),
244 ({"data.data2.data3": 4}, [4]),
245 ({"data.data2.data3": [3, 4, "e"]}, [3, 4]),
246 ({"data.data2.data3": None}, [8]),
247 ({"data.data2": "4"}, []),
248 ({"list.0.a": 1}, [1, 6]),
250 ({"list2": [1, 5]}, [2]),
251 ({"list2": [1, 2]}, [2]),
252 ({"list2": [5, 7]}, []),
253 ({"list.ANYINDEX.a": 1}, [1]),
254 ({"list.a": 3, "list.b": 1}, [8]),
255 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 1}, []),
256 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.a": 3}, [8]),
257 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 0}, [8]),
260 "list.ANYINDEX.a": 3,
261 "list.ANYINDEX.c.ANYINDEX.a": 0,
262 "list.ANYINDEX.c.ANYINDEX.b": "v",
268 "list.ANYINDEX.a": 3,
269 "list.ANYINDEX.c.ANYINDEX.a": 0,
270 "list.ANYINDEX.c.ANYINDEX.b": 1,
274 ({"list.c.b": 1}, [8]),
275 ({"list.c.b": None}, [1, 2, 3, 4, 5, 6, 7]),
276 # ({"data.data2.data3": 4}, []),
277 # ({"data.data2.data3": 4}, []),
280 def test_get_list(db_memory_with_many_data
, db_filter
, expected_ids
):
281 result
= db_memory_with_many_data
.get_list("test", db_filter
)
282 assert isinstance(result
, list)
283 result_ids
= [item
["_id"] for item
in result
]
284 assert len(result
) == len(
286 ), "for db_filter={} result={} expected_ids={}".format(
287 db_filter
, result
, result_ids
289 assert result_ids
== expected_ids
290 for i
in range(len(result
)):
291 assert result
[i
] in db_memory_with_many_data
.db
["test"]
293 assert len(db_memory_with_many_data
.db
) == 1
294 assert "test" in db_memory_with_many_data
.db
295 assert len(db_memory_with_many_data
.db
["test"]) == 8
296 result
= db_memory_with_many_data
.count("test", db_filter
)
297 assert result
== len(expected_ids
)
300 @pytest.mark
.parametrize(
301 "table, db_filter, expected_data", [("test", {}, {"_id": 1, "data": 1})]
303 def test_get_one_with_multiple_results(
304 db_memory_with_data
, table
, db_filter
, expected_data
306 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_more
=False)
307 assert result
== expected_data
308 assert len(db_memory_with_data
.db
) == 1
309 assert table
in db_memory_with_data
.db
310 assert len(db_memory_with_data
.db
[table
]) == 3
311 assert result
in db_memory_with_data
.db
[table
]
314 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
317 with pytest
.raises(DbException
) as excinfo
:
318 db_memory_with_data
.get_one(table
, db_filter
)
319 assert str(excinfo
.value
) == (
320 empty_exception_message() + get_one_multiple_exception_message(db_filter
)
322 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
325 @pytest.mark
.parametrize(
328 ("test", {"_id": 4}),
329 ("test", {"data": 4}),
330 ("test", {"_id": 4, "data": 4}),
331 ("test_table", {"_id": 4}),
332 ("test_table", {"data": 4}),
333 ("test_table", {"_id": 4, "data": 4}),
336 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
337 with pytest
.raises(DbException
) as excinfo
:
338 db_memory_with_data
.get_one(table
, db_filter
)
339 assert str(excinfo
.value
) == (
340 empty_exception_message() + get_one_exception_message(db_filter
)
342 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
345 @pytest.mark
.parametrize(
348 ("test", {"_id": 4}),
349 ("test", {"data": 4}),
350 ("test", {"_id": 4, "data": 4}),
351 ("test_table", {"_id": 4}),
352 ("test_table", {"data": 4}),
353 ("test_table", {"_id": 4, "data": 4}),
356 def test_get_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
357 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_empty
=False)
358 assert result
is None
361 @pytest.mark
.parametrize(
364 ("test", {"_id": 4}),
365 ("test", {"data": 4}),
366 ("test", {"_id": 4, "data": 4}),
367 ("test_table", {"_id": 4}),
368 ("test_table", {"data": 4}),
369 ("test_table", {"_id": 4, "data": 4}),
372 def test_get_one_with_empty_db_exception(db_memory
, table
, db_filter
):
373 with pytest
.raises(DbException
) as excinfo
:
374 db_memory
.get_one(table
, db_filter
)
375 assert str(excinfo
.value
) == (
376 empty_exception_message() + get_one_exception_message(db_filter
)
378 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
381 @pytest.mark
.parametrize(
384 ("test", {"_id": 4}),
385 ("test", {"data": 4}),
386 ("test", {"_id": 4, "data": 4}),
387 ("test_table", {"_id": 4}),
388 ("test_table", {"data": 4}),
389 ("test_table", {"_id": 4, "data": 4}),
392 def test_get_one_with_empty_db_none(db_memory
, table
, db_filter
):
393 result
= db_memory
.get_one(table
, db_filter
, fail_on_empty
=False)
394 assert result
is None
397 def test_get_one_generic_exception(db_memory_with_data
):
400 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
401 with pytest
.raises(DbException
) as excinfo
:
402 db_memory_with_data
.get_one(table
, db_filter
)
403 assert str(excinfo
.value
) == empty_exception_message()
404 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
407 @pytest.mark
.parametrize(
408 "table, db_filter, expected_data",
411 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
412 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
413 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
414 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
417 def test_del_list_with_non_empty_db(
418 db_memory_with_data
, table
, db_filter
, expected_data
420 result
= db_memory_with_data
.del_list(table
, db_filter
)
421 assert result
["deleted"] == (3 - len(expected_data
))
422 assert len(db_memory_with_data
.db
) == 1
423 assert table
in db_memory_with_data
.db
424 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
425 for data
in expected_data
:
426 assert data
in db_memory_with_data
.db
[table
]
429 @pytest.mark
.parametrize(
433 ("test", {"_id": 1}),
434 ("test", {"_id": 2}),
435 ("test", {"data": 1}),
436 ("test", {"data": 2}),
437 ("test", {"_id": 1, "data": 1}),
438 ("test", {"_id": 2, "data": 2}),
441 def test_del_list_with_empty_db(db_memory
, table
, db_filter
):
442 result
= db_memory
.del_list(table
, db_filter
)
443 assert result
["deleted"] == 0
446 def test_del_list_generic_exception(db_memory_with_data
):
449 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
450 with pytest
.raises(DbException
) as excinfo
:
451 db_memory_with_data
.del_list(table
, db_filter
)
452 assert str(excinfo
.value
) == empty_exception_message()
453 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
456 @pytest.mark
.parametrize(
457 "table, db_filter, data",
459 ("test", {}, {"_id": 1, "data": 1}),
460 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
461 ("test", {"data": 1}, {"_id": 1, "data": 1}),
462 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
463 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
464 ("test", {"data": 2}, {"_id": 2, "data": 2}),
465 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
468 def test_del_one(db_memory_with_data
, table
, db_filter
, data
):
469 result
= db_memory_with_data
.del_one(table
, db_filter
)
470 assert result
== {"deleted": 1}
471 assert len(db_memory_with_data
.db
) == 1
472 assert table
in db_memory_with_data
.db
473 assert len(db_memory_with_data
.db
[table
]) == 2
474 assert data
not in db_memory_with_data
.db
[table
]
477 @pytest.mark
.parametrize(
481 ("test", {"_id": 1}),
482 ("test", {"_id": 2}),
483 ("test", {"data": 1}),
484 ("test", {"data": 2}),
485 ("test", {"_id": 1, "data": 1}),
486 ("test", {"_id": 2, "data": 2}),
488 ("test_table", {"_id": 1}),
489 ("test_table", {"_id": 2}),
490 ("test_table", {"data": 1}),
491 ("test_table", {"data": 2}),
492 ("test_table", {"_id": 1, "data": 1}),
493 ("test_table", {"_id": 2, "data": 2}),
496 def test_del_one_with_empty_db_exception(db_memory
, table
, db_filter
):
497 with pytest
.raises(DbException
) as excinfo
:
498 db_memory
.del_one(table
, db_filter
)
499 assert str(excinfo
.value
) == (
500 empty_exception_message() + del_one_exception_message(db_filter
)
502 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
505 @pytest.mark
.parametrize(
509 ("test", {"_id": 1}),
510 ("test", {"_id": 2}),
511 ("test", {"data": 1}),
512 ("test", {"data": 2}),
513 ("test", {"_id": 1, "data": 1}),
514 ("test", {"_id": 2, "data": 2}),
516 ("test_table", {"_id": 1}),
517 ("test_table", {"_id": 2}),
518 ("test_table", {"data": 1}),
519 ("test_table", {"data": 2}),
520 ("test_table", {"_id": 1, "data": 1}),
521 ("test_table", {"_id": 2, "data": 2}),
524 def test_del_one_with_empty_db_none(db_memory
, table
, db_filter
):
525 result
= db_memory
.del_one(table
, db_filter
, fail_on_empty
=False)
526 assert result
is None
529 @pytest.mark
.parametrize(
532 ("test", {"_id": 4}),
533 ("test", {"_id": 5}),
534 ("test", {"data": 4}),
535 ("test", {"data": 5}),
536 ("test", {"_id": 1, "data": 2}),
537 ("test", {"_id": 2, "data": 3}),
539 ("test_table", {"_id": 1}),
540 ("test_table", {"_id": 2}),
541 ("test_table", {"data": 1}),
542 ("test_table", {"data": 2}),
543 ("test_table", {"_id": 1, "data": 1}),
544 ("test_table", {"_id": 2, "data": 2}),
547 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
548 with pytest
.raises(DbException
) as excinfo
:
549 db_memory_with_data
.del_one(table
, db_filter
)
550 assert str(excinfo
.value
) == (
551 empty_exception_message() + del_one_exception_message(db_filter
)
553 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
556 @pytest.mark
.parametrize(
559 ("test", {"_id": 4}),
560 ("test", {"_id": 5}),
561 ("test", {"data": 4}),
562 ("test", {"data": 5}),
563 ("test", {"_id": 1, "data": 2}),
564 ("test", {"_id": 2, "data": 3}),
566 ("test_table", {"_id": 1}),
567 ("test_table", {"_id": 2}),
568 ("test_table", {"data": 1}),
569 ("test_table", {"data": 2}),
570 ("test_table", {"_id": 1, "data": 1}),
571 ("test_table", {"_id": 2, "data": 2}),
574 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
575 result
= db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=False)
576 assert result
is None
579 @pytest.mark
.parametrize("fail_on_empty", [(True), (False)])
580 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
583 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
584 with pytest
.raises(DbException
) as excinfo
:
585 db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=fail_on_empty
)
586 assert str(excinfo
.value
) == empty_exception_message()
587 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
590 @pytest.mark
.parametrize(
591 "table, _id, indata",
593 ("test", 1, {"_id": 1, "data": 42}),
594 ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
595 ("test", 1, {"_id": 1}),
596 ("test", 2, {"_id": 2, "data": 42}),
597 ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
598 ("test", 2, {"_id": 2}),
599 ("test", 3, {"_id": 3, "data": 42}),
600 ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
601 ("test", 3, {"_id": 3}),
604 def test_replace(db_memory_with_data
, table
, _id
, indata
):
605 result
= db_memory_with_data
.replace(table
, _id
, indata
)
606 assert result
== {"updated": 1}
607 assert len(db_memory_with_data
.db
) == 1
608 assert table
in db_memory_with_data
.db
609 assert len(db_memory_with_data
.db
[table
]) == 3
610 assert indata
in db_memory_with_data
.db
[table
]
613 @pytest.mark
.parametrize(
614 "table, _id, indata",
616 ("test", 1, {"_id": 1, "data": 42}),
617 ("test", 2, {"_id": 2}),
618 ("test", 3, {"_id": 3}),
621 def test_replace_without_data_exception(db_memory
, table
, _id
, indata
):
622 with pytest
.raises(DbException
) as excinfo
:
623 db_memory
.replace(table
, _id
, indata
, fail_on_empty
=True)
624 assert str(excinfo
.value
) == (replace_exception_message(_id
))
625 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
628 @pytest.mark
.parametrize(
629 "table, _id, indata",
631 ("test", 1, {"_id": 1, "data": 42}),
632 ("test", 2, {"_id": 2}),
633 ("test", 3, {"_id": 3}),
636 def test_replace_without_data_none(db_memory
, table
, _id
, indata
):
637 result
= db_memory
.replace(table
, _id
, indata
, fail_on_empty
=False)
638 assert result
is None
641 @pytest.mark
.parametrize(
642 "table, _id, indata",
644 ("test", 11, {"_id": 11, "data": 42}),
645 ("test", 12, {"_id": 12}),
646 ("test", 33, {"_id": 33}),
649 def test_replace_with_data_exception(db_memory_with_data
, table
, _id
, indata
):
650 with pytest
.raises(DbException
) as excinfo
:
651 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=True)
652 assert str(excinfo
.value
) == (replace_exception_message(_id
))
653 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
656 @pytest.mark
.parametrize(
657 "table, _id, indata",
659 ("test", 11, {"_id": 11, "data": 42}),
660 ("test", 12, {"_id": 12}),
661 ("test", 33, {"_id": 33}),
664 def test_replace_with_data_none(db_memory_with_data
, table
, _id
, indata
):
665 result
= db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=False)
666 assert result
is None
669 @pytest.mark
.parametrize("fail_on_empty", [True, False])
670 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
673 indata
= {"_id": 1, "data": 1}
674 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
675 with pytest
.raises(DbException
) as excinfo
:
676 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=fail_on_empty
)
677 assert str(excinfo
.value
) == empty_exception_message()
678 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
681 @pytest.mark
.parametrize(
684 ("test", "1", {"data": 1}),
685 ("test", "1", {"data": 2}),
686 ("test", "2", {"data": 1}),
687 ("test", "2", {"data": 2}),
688 ("test_table", "1", {"data": 1}),
689 ("test_table", "1", {"data": 2}),
690 ("test_table", "2", {"data": 1}),
691 ("test_table", "2", {"data": 2}),
692 ("test", "1", {"data_1": 1, "data_2": 2}),
693 ("test", "1", {"data_1": 2, "data_2": 1}),
694 ("test", "2", {"data_1": 1, "data_2": 2}),
695 ("test", "2", {"data_1": 2, "data_2": 1}),
696 ("test_table", "1", {"data_1": 1, "data_2": 2}),
697 ("test_table", "1", {"data_1": 2, "data_2": 1}),
698 ("test_table", "2", {"data_1": 1, "data_2": 2}),
699 ("test_table", "2", {"data_1": 2, "data_2": 1}),
702 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
703 data_to_insert
= data
704 data_to_insert
["_id"] = id
705 returned_id
= db_memory
.create(table
, data_to_insert
)
706 assert returned_id
== id
707 assert len(db_memory
.db
) == 1
708 assert table
in db_memory
.db
709 assert len(db_memory
.db
[table
]) == 1
710 assert data_to_insert
in db_memory
.db
[table
]
713 @pytest.mark
.parametrize(
716 ("test", "4", {"data": 1}),
717 ("test", "5", {"data": 2}),
718 ("test", "4", {"data": 1}),
719 ("test", "5", {"data": 2}),
720 ("test_table", "4", {"data": 1}),
721 ("test_table", "5", {"data": 2}),
722 ("test_table", "4", {"data": 1}),
723 ("test_table", "5", {"data": 2}),
724 ("test", "4", {"data_1": 1, "data_2": 2}),
725 ("test", "5", {"data_1": 2, "data_2": 1}),
726 ("test", "4", {"data_1": 1, "data_2": 2}),
727 ("test", "5", {"data_1": 2, "data_2": 1}),
728 ("test_table", "4", {"data_1": 1, "data_2": 2}),
729 ("test_table", "5", {"data_1": 2, "data_2": 1}),
730 ("test_table", "4", {"data_1": 1, "data_2": 2}),
731 ("test_table", "5", {"data_1": 2, "data_2": 1}),
734 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
735 data_to_insert
= data
736 data_to_insert
["_id"] = id
737 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
738 assert returned_id
== id
739 assert len(db_memory_with_data
.db
) == (1 if table
== "test" else 2)
740 assert table
in db_memory_with_data
.db
741 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== "test" else 1)
742 assert data_to_insert
in db_memory_with_data
.db
[table
]
745 @pytest.mark
.parametrize(
748 ("test", {"data": 1}),
749 ("test", {"data": 2}),
750 ("test", {"data": 1}),
751 ("test", {"data": 2}),
752 ("test_table", {"data": 1}),
753 ("test_table", {"data": 2}),
754 ("test_table", {"data": 1}),
755 ("test_table", {"data": 2}),
756 ("test", {"data_1": 1, "data_2": 2}),
757 ("test", {"data_1": 2, "data_2": 1}),
758 ("test", {"data_1": 1, "data_2": 2}),
759 ("test", {"data_1": 2, "data_2": 1}),
760 ("test_table", {"data_1": 1, "data_2": 2}),
761 ("test_table", {"data_1": 2, "data_2": 1}),
762 ("test_table", {"data_1": 1, "data_2": 2}),
763 ("test_table", {"data_1": 2, "data_2": 1}),
766 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
767 returned_id
= db_memory
.create(table
, data
)
768 assert len(db_memory
.db
) == 1
769 assert table
in db_memory
.db
770 assert len(db_memory
.db
[table
]) == 1
772 data_inserted
["_id"] = returned_id
773 assert data_inserted
in db_memory
.db
[table
]
776 @pytest.mark
.parametrize(
779 ("test", {"data": 1}),
780 ("test", {"data": 2}),
781 ("test", {"data": 1}),
782 ("test", {"data": 2}),
783 ("test_table", {"data": 1}),
784 ("test_table", {"data": 2}),
785 ("test_table", {"data": 1}),
786 ("test_table", {"data": 2}),
787 ("test", {"data_1": 1, "data_2": 2}),
788 ("test", {"data_1": 2, "data_2": 1}),
789 ("test", {"data_1": 1, "data_2": 2}),
790 ("test", {"data_1": 2, "data_2": 1}),
791 ("test_table", {"data_1": 1, "data_2": 2}),
792 ("test_table", {"data_1": 2, "data_2": 1}),
793 ("test_table", {"data_1": 1, "data_2": 2}),
794 ("test_table", {"data_1": 2, "data_2": 1}),
797 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
798 returned_id
= db_memory_with_data
.create(table
, data
)
799 assert len(db_memory_with_data
.db
) == (1 if table
== "test" else 2)
800 assert table
in db_memory_with_data
.db
801 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== "test" else 1)
803 data_inserted
["_id"] = returned_id
804 assert data_inserted
in db_memory_with_data
.db
[table
]
807 def test_create_with_exception(db_memory
):
809 data
= {"_id": 1, "data": 1}
810 db_memory
.db
= MagicMock()
811 db_memory
.db
.__contains
__.side_effect
= Exception()
812 with pytest
.raises(DbException
) as excinfo
:
813 db_memory
.create(table
, data
)
814 assert str(excinfo
.value
) == empty_exception_message()
815 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
818 @pytest.mark
.parametrize(
819 "db_content, update_dict, expected, message",
822 {"a": {"none": None}},
824 {"a": {"none": None, "b": {"num": "v"}}},
828 {"a": {"none": None}},
830 {"a": {"none": {"num": "v"}}},
831 "create dict over none",
834 {"a": {"b": {"num": 4}}},
836 {"a": {"b": {"num": "v"}}},
840 {"a": {"b": {"num": 4}}},
841 {"a.b.num.c.d": "v"},
843 "create dict over number should fail",
846 {"a": {"b": {"num": 4}}},
849 "replace dict with a string",
852 {"a": {"b": {"num": 4}}},
855 "replace dict with None",
858 {"a": [{"b": {"num": 4}}]},
861 "create dict over list should fail",
864 {"a": [{"b": {"num": 4}}]},
866 {"a": [{"b": {"num": "v"}}]},
870 {"a": [{"b": {"num": 4}}]},
872 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
875 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
876 ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
880 {"a": [[4], None, {"2": "v"}]},
881 "expand list and add number key",
885 def test_set_one(db_memory
, db_content
, update_dict
, expected
, message
):
886 db_memory
._find
= Mock(return_value
=((0, db_content
),))
888 with pytest
.raises(DbException
) as excinfo
:
889 db_memory
.set_one("table", {}, update_dict
)
890 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
, message
892 db_memory
.set_one("table", {}, update_dict
)
893 assert db_content
== expected
, message
896 class TestDbMemory(unittest
.TestCase
):
897 # TODO to delete. This is cover with pytest test_set_one.
898 def test_set_one(self
):
900 # (database content, set-content, expected database content (None=fails), message)
902 {"a": {"none": None}},
904 {"a": {"none": None, "b": {"num": "v"}}},
908 {"a": {"none": None}},
910 {"a": {"none": {"num": "v"}}},
911 "create dict over none",
914 {"a": {"b": {"num": 4}}},
916 {"a": {"b": {"num": "v"}}},
920 {"a": {"b": {"num": 4}}},
921 {"a.b.num.c.d": "v"},
923 "create dict over number should fail",
926 {"a": {"b": {"num": 4}}},
929 "replace dict with a string",
932 {"a": {"b": {"num": 4}}},
935 "replace dict with None",
938 {"a": [{"b": {"num": 4}}]},
941 "create dict over list should fail",
944 {"a": [{"b": {"num": 4}}]},
946 {"a": [{"b": {"num": "v"}}]},
950 {"a": [{"b": {"num": 4}}]},
952 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]},
955 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
959 {"a": [[4, None, "v"]]},
960 "expand nested list",
965 {"a": [[4], None, {"2": "v"}]},
966 "expand list and add number key",
968 ({"a": None}, {"b.c": "v"}, {"a": None, "b": {"c": "v"}}, "expand at root"),
971 db_men
._find
= Mock()
972 for db_content
, update_dict
, expected
, message
in test_set
:
973 db_men
._find
.return_value
= ((0, db_content
),)
975 self
.assertRaises(DbException
, db_men
.set_one
, "table", {}, update_dict
)
977 db_men
.set_one("table", {}, update_dict
)
978 self
.assertEqual(db_content
, expected
, message
)
980 def test_set_one_pull(self
):
981 example
= {"a": [1, "1", 1], "d": {}, "n": None}
983 # (database content, set-content, expected database content (None=fails), message)
984 (example
, {"a": "1"}, {"a": [1, 1], "d": {}, "n": None}, "pull one item"),
985 (example
, {"a": 1}, {"a": ["1"], "d": {}, "n": None}, "pull two items"),
986 (example
, {"a": "v"}, example
, "pull non existing item"),
987 (example
, {"a.6": 1}, example
, "pull non existing arrray"),
988 (example
, {"d.b.c": 1}, example
, "pull non existing arrray2"),
989 (example
, {"b": 1}, example
, "pull non existing arrray3"),
990 (example
, {"d": 1}, None, "pull over dict"),
991 (example
, {"n": 1}, None, "pull over None"),
994 db_men
._find
= Mock()
995 for db_content
, pull_dict
, expected
, message
in test_set
:
996 db_content
= deepcopy(db_content
)
997 db_men
._find
.return_value
= ((0, db_content
),)
1005 fail_on_empty
=False,
1009 db_men
.set_one("table", {}, None, pull
=pull_dict
)
1010 self
.assertEqual(db_content
, expected
, message
)
1012 def test_set_one_push(self
):
1013 example
= {"a": [1, "1", 1], "d": {}, "n": None}
1015 # (database content, set-content, expected database content (None=fails), message)
1019 {"a": [1, "1", 1], "d": {"b": {"c": [1]}}, "n": None},
1020 "push non existing arrray2",
1025 {"a": [1, "1", 1], "d": {}, "b": [1], "n": None},
1026 "push non existing arrray3",
1031 {"a": [1, "1", 1, None, None, None, [1]], "d": {}, "n": None},
1032 "push non existing arrray",
1037 {"a": [1, "1", 1, 2], "d": {}, "n": None},
1043 {"a": [1, "1", 1, {1: 1}], "d": {}, "n": None},
1046 (example
, {"d": 1}, None, "push over dict"),
1047 (example
, {"n": 1}, None, "push over None"),
1050 db_men
._find
= Mock()
1051 for db_content
, push_dict
, expected
, message
in test_set
:
1052 db_content
= deepcopy(db_content
)
1053 db_men
._find
.return_value
= ((0, db_content
),)
1054 if expected
is None:
1061 fail_on_empty
=False,
1065 db_men
.set_one("table", {}, None, push
=push_dict
)
1066 self
.assertEqual(db_content
, expected
, message
)
1068 def test_set_one_push_list(self
):
1069 example
= {"a": [1, "1", 1], "d": {}, "n": None}
1071 # (database content, set-content, expected database content (None=fails), message)
1075 {"a": [1, "1", 1], "d": {"b": {"c": [1]}}, "n": None},
1076 "push non existing arrray2",
1081 {"a": [1, "1", 1], "d": {}, "b": [1], "n": None},
1082 "push non existing arrray3",
1087 {"a": [1, "1", 1, None, None, None, [1]], "d": {}, "n": None},
1088 "push non existing arrray",
1093 {"a": [1, "1", 1, 2, 3], "d": {}, "n": None},
1099 {"a": [1, "1", 1, {1: 1}], "d": {}, "n": None},
1102 (example
, {"d": [1]}, None, "push over dict"),
1103 (example
, {"n": [1]}, None, "push over None"),
1104 (example
, {"a": 1}, None, "invalid push list non an array"),
1107 db_men
._find
= Mock()
1108 for db_content
, push_list
, expected
, message
in test_set
:
1109 db_content
= deepcopy(db_content
)
1110 db_men
._find
.return_value
= ((0, db_content
),)
1111 if expected
is None:
1118 fail_on_empty
=False,
1119 push_list
=push_list
,
1122 db_men
.set_one("table", {}, None, push_list
=push_list
)
1123 self
.assertEqual(db_content
, expected
, message
)
1125 def test_unset_one(self
):
1126 example
= {"a": [1, "1", 1], "d": {}, "n": None}
1128 # (database content, set-content, expected database content (None=fails), message)
1129 (example
, {"d.b.c": 1}, example
, "unset non existing"),
1130 (example
, {"b": 1}, example
, "unset non existing"),
1131 (example
, {"a.6": 1}, example
, "unset non existing arrray"),
1132 (example
, {"a": 2}, {"d": {}, "n": None}, "unset array"),
1133 (example
, {"d": 1}, {"a": [1, "1", 1], "n": None}, "unset dict"),
1134 (example
, {"n": 1}, {"a": [1, "1", 1], "d": {}}, "unset None"),
1137 db_men
._find
= Mock()
1138 for db_content
, unset_dict
, expected
, message
in test_set
:
1139 db_content
= deepcopy(db_content
)
1140 db_men
._find
.return_value
= ((0, db_content
),)
1141 if expected
is None:
1148 fail_on_empty
=False,
1152 db_men
.set_one("table", {}, None, unset
=unset_dict
)
1153 self
.assertEqual(db_content
, expected
, message
)