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 Mock
26 from unittest
.mock
import MagicMock
27 from osm_common
.dbbase
import DbException
28 from osm_common
.dbmemory
import DbMemory
30 __author__
= 'Eduardo Sousa <eduardosousa@av.it.pt>'
33 @pytest.fixture(scope
="function", params
=[True, False])
34 def db_memory(request
):
35 db
= DbMemory(lock
=request
.param
)
39 @pytest.fixture(scope
="function", params
=[True, False])
40 def db_memory_with_data(request
):
41 db
= DbMemory(lock
=request
.param
)
43 db
.create("test", {"_id": 1, "data": 1})
44 db
.create("test", {"_id": 2, "data": 2})
45 db
.create("test", {"_id": 3, "data": 3})
50 @pytest.fixture(scope
="function")
51 def db_memory_with_many_data(request
):
52 db
= DbMemory(lock
=False)
54 db
.create_list("test", [
55 {"_id": 1, "data": {"data2": {"data3": 1}}, "list": [{"a": 1}], "text": "sometext"},
56 {"_id": 2, "data": {"data2": {"data3": 2}}, "list": [{"a": 2}], "list2": [1, 2, 3]},
57 {"_id": 3, "data": {"data2": {"data3": 3}}, "list": [{"a": 3}]},
58 {"_id": 4, "data": {"data2": {"data3": 4}}, "list": [{"a": 4}, {"a": 0}]},
59 {"_id": 5, "data": {"data2": {"data3": 5}}, "list": [{"a": 5}]},
60 {"_id": 6, "data": {"data2": {"data3": 6}}, "list": [{"0": {"a": 1}}]},
61 {"_id": 7, "data": {"data2": {"data3": 7}}, "0": {"a": 0}},
62 {"_id": 8, "list": [{"a": 3, "b": 0, "c": [{"a": 3, "b": 1}, {"a": 0, "b": "v"}]}, {"a": 0, "b": 1}]},
67 def empty_exception_message():
68 return 'database exception '
71 def get_one_exception_message(db_filter
):
72 return "database exception Not found entry with filter='{}'".format(db_filter
)
75 def get_one_multiple_exception_message(db_filter
):
76 return "database exception Found more than one entry with filter='{}'".format(db_filter
)
79 def del_one_exception_message(db_filter
):
80 return "database exception Not found entry with filter='{}'".format(db_filter
)
83 def replace_exception_message(value
):
84 return "database exception Not found entry with _id='{}'".format(value
)
87 def test_constructor():
89 assert db
.logger
== logging
.getLogger('db')
93 def test_constructor_with_logger():
94 logger_name
= 'db_local'
95 db
= DbMemory(logger_name
=logger_name
)
96 assert db
.logger
== logging
.getLogger(logger_name
)
100 def test_db_connect():
101 logger_name
= 'db_local'
102 config
= {'logger_name': logger_name
}
104 db
.db_connect(config
)
105 assert db
.logger
== logging
.getLogger(logger_name
)
109 def test_db_disconnect(db_memory
):
110 db_memory
.db_disconnect()
113 @pytest.mark
.parametrize("table, db_filter", [
115 ("test", {"_id": 1}),
116 ("test", {"data": 1}),
117 ("test", {"_id": 1, "data": 1})])
118 def test_get_list_with_empty_db(db_memory
, table
, db_filter
):
119 result
= db_memory
.get_list(table
, db_filter
)
120 assert len(result
) == 0
123 @pytest.mark
.parametrize("table, db_filter, expected_data", [
124 ("test", {}, [{"_id": 1, "data": 1}, {"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
125 ("test", {"_id": 1}, [{"_id": 1, "data": 1}]),
126 ("test", {"data": 1}, [{"_id": 1, "data": 1}]),
127 ("test", {"_id": 1, "data": 1}, [{"_id": 1, "data": 1}]),
128 ("test", {"_id": 2}, [{"_id": 2, "data": 2}]),
129 ("test", {"data": 2}, [{"_id": 2, "data": 2}]),
130 ("test", {"_id": 2, "data": 2}, [{"_id": 2, "data": 2}]),
131 ("test", {"_id": 4}, []),
132 ("test", {"data": 4}, []),
133 ("test", {"_id": 4, "data": 4}, []),
134 ("test_table", {}, []),
135 ("test_table", {"_id": 1}, []),
136 ("test_table", {"data": 1}, []),
137 ("test_table", {"_id": 1, "data": 1}, [])])
138 def test_get_list_with_non_empty_db(db_memory_with_data
, table
, db_filter
, expected_data
):
139 result
= db_memory_with_data
.get_list(table
, db_filter
)
140 assert len(result
) == len(expected_data
)
141 for data
in expected_data
:
142 assert data
in result
145 def test_get_list_exception(db_memory_with_data
):
148 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
149 with pytest
.raises(DbException
) as excinfo
:
150 db_memory_with_data
.get_list(table
, db_filter
)
151 assert str(excinfo
.value
) == empty_exception_message()
152 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
155 @pytest.mark
.parametrize("table, db_filter, expected_data", [
156 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
157 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
158 ("test", {"_id": 3}, {"_id": 3, "data": 3}),
159 ("test", {"data": 1}, {"_id": 1, "data": 1}),
160 ("test", {"data": 2}, {"_id": 2, "data": 2}),
161 ("test", {"data": 3}, {"_id": 3, "data": 3}),
162 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
163 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2}),
164 ("test", {"_id": 3, "data": 3}, {"_id": 3, "data": 3})])
165 def test_get_one(db_memory_with_data
, table
, db_filter
, expected_data
):
166 result
= db_memory_with_data
.get_one(table
, db_filter
)
167 assert result
== expected_data
168 assert len(db_memory_with_data
.db
) == 1
169 assert table
in db_memory_with_data
.db
170 assert len(db_memory_with_data
.db
[table
]) == 3
171 assert result
in db_memory_with_data
.db
[table
]
174 @pytest.mark
.parametrize("db_filter, expected_ids", [
175 ({}, [1, 2, 3, 4, 5, 6, 7, 8]),
177 ({"data.data2.data3": 2}, [2]),
178 ({"data.data2.data3.eq": 2}, [2]),
179 ({"data.data2.data3": [2]}, [2]),
180 ({"data.data2.data3.cont": [2]}, [2]),
181 ({"data.data2.data3.neq": 2}, [1, 3, 4, 5, 6, 7, 8]),
182 ({"data.data2.data3.neq": [2]}, [1, 3, 4, 5, 6, 7, 8]),
183 ({"data.data2.data3.ncont": [2]}, [1, 3, 4, 5, 6, 7, 8]),
184 ({"data.data2.data3": [2, 3]}, [2, 3]),
185 ({"data.data2.data3.gt": 4}, [5, 6, 7]),
186 ({"data.data2.data3.gte": 4}, [4, 5, 6, 7]),
187 ({"data.data2.data3.lt": 4}, [1, 2, 3]),
188 ({"data.data2.data3.lte": 4}, [1, 2, 3, 4]),
189 ({"data.data2.data3.lte": 4.5}, [1, 2, 3, 4]),
190 ({"data.data2.data3.gt": "text"}, []),
191 ({"nonexist.nonexist": "4"}, []),
192 ({"nonexist.nonexist": None}, [1, 2, 3, 4, 5, 6, 7, 8]),
193 ({"nonexist.nonexist.neq": "4"}, [1, 2, 3, 4, 5, 6, 7, 8]),
194 ({"nonexist.nonexist.neq": None}, []),
195 ({"text.eq": "sometext"}, [1]),
196 ({"text.neq": "sometext"}, [2, 3, 4, 5, 6, 7, 8]),
197 ({"text.eq": "somet"}, []),
198 ({"text.gte": "a"}, [1]),
199 ({"text.gte": "somet"}, [1]),
200 ({"text.gte": "sometext"}, [1]),
201 ({"text.lt": "somet"}, []),
202 ({"data.data2.data3": 2, "data.data2.data4": None}, [2]),
203 ({"data.data2.data3": 2, "data.data2.data4": 5}, []),
204 ({"data.data2.data3": 4}, [4]),
205 ({"data.data2.data3": [3, 4, "e"]}, [3, 4]),
206 ({"data.data2.data3": None}, [8]),
207 ({"data.data2": "4"}, []),
208 ({"list.0.a": 1}, [1, 6]),
210 ({"list2": [1, 5]}, [2]),
211 ({"list2": [1, 2]}, [2]),
212 ({"list2": [5, 7]}, []),
213 ({"list.ANYINDEX.a": 1}, [1]),
214 ({"list.a": 3, "list.b": 1}, [8]),
215 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 1}, []),
216 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.a": 3}, [8]),
217 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 0}, [8]),
218 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.ANYINDEX.a": 0, "list.ANYINDEX.c.ANYINDEX.b": "v"}, [8]),
219 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.ANYINDEX.a": 0, "list.ANYINDEX.c.ANYINDEX.b": 1}, []),
220 ({"list.c.b": 1}, [8]),
221 ({"list.c.b": None}, [1, 2, 3, 4, 5, 6, 7]),
222 # ({"data.data2.data3": 4}, []),
223 # ({"data.data2.data3": 4}, []),
225 def test_get_list(db_memory_with_many_data
, db_filter
, expected_ids
):
226 result
= db_memory_with_many_data
.get_list("test", db_filter
)
227 assert isinstance(result
, list)
228 result_ids
= [item
["_id"] for item
in result
]
229 assert len(result
) == len(expected_ids
), "for db_filter={} result={} expected_ids={}".format(db_filter
, result
,
231 assert result_ids
== expected_ids
232 for i
in range(len(result
)):
233 assert result
[i
] in db_memory_with_many_data
.db
["test"]
235 assert len(db_memory_with_many_data
.db
) == 1
236 assert "test" in db_memory_with_many_data
.db
237 assert len(db_memory_with_many_data
.db
["test"]) == 8
238 result
= db_memory_with_many_data
.count("test", db_filter
)
239 assert result
== len(expected_ids
)
242 @pytest.mark
.parametrize("table, db_filter, expected_data", [
243 ("test", {}, {"_id": 1, "data": 1})])
244 def test_get_one_with_multiple_results(db_memory_with_data
, table
, db_filter
, expected_data
):
245 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_more
=False)
246 assert result
== expected_data
247 assert len(db_memory_with_data
.db
) == 1
248 assert table
in db_memory_with_data
.db
249 assert len(db_memory_with_data
.db
[table
]) == 3
250 assert result
in db_memory_with_data
.db
[table
]
253 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
256 with pytest
.raises(DbException
) as excinfo
:
257 db_memory_with_data
.get_one(table
, db_filter
)
258 assert str(excinfo
.value
) == (empty_exception_message() + get_one_multiple_exception_message(db_filter
))
259 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
262 @pytest.mark
.parametrize("table, db_filter", [
263 ("test", {"_id": 4}),
264 ("test", {"data": 4}),
265 ("test", {"_id": 4, "data": 4}),
266 ("test_table", {"_id": 4}),
267 ("test_table", {"data": 4}),
268 ("test_table", {"_id": 4, "data": 4})])
269 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
270 with pytest
.raises(DbException
) as excinfo
:
271 db_memory_with_data
.get_one(table
, db_filter
)
272 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(db_filter
))
273 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
276 @pytest.mark
.parametrize("table, db_filter", [
277 ("test", {"_id": 4}),
278 ("test", {"data": 4}),
279 ("test", {"_id": 4, "data": 4}),
280 ("test_table", {"_id": 4}),
281 ("test_table", {"data": 4}),
282 ("test_table", {"_id": 4, "data": 4})])
283 def test_get_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
284 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_empty
=False)
285 assert result
is None
288 @pytest.mark
.parametrize("table, db_filter", [
289 ("test", {"_id": 4}),
290 ("test", {"data": 4}),
291 ("test", {"_id": 4, "data": 4}),
292 ("test_table", {"_id": 4}),
293 ("test_table", {"data": 4}),
294 ("test_table", {"_id": 4, "data": 4})])
295 def test_get_one_with_empty_db_exception(db_memory
, table
, db_filter
):
296 with pytest
.raises(DbException
) as excinfo
:
297 db_memory
.get_one(table
, db_filter
)
298 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(db_filter
))
299 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
302 @pytest.mark
.parametrize("table, db_filter", [
303 ("test", {"_id": 4}),
304 ("test", {"data": 4}),
305 ("test", {"_id": 4, "data": 4}),
306 ("test_table", {"_id": 4}),
307 ("test_table", {"data": 4}),
308 ("test_table", {"_id": 4, "data": 4})])
309 def test_get_one_with_empty_db_none(db_memory
, table
, db_filter
):
310 result
= db_memory
.get_one(table
, db_filter
, fail_on_empty
=False)
311 assert result
is None
314 def test_get_one_generic_exception(db_memory_with_data
):
317 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
318 with pytest
.raises(DbException
) as excinfo
:
319 db_memory_with_data
.get_one(table
, db_filter
)
320 assert str(excinfo
.value
) == empty_exception_message()
321 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
324 @pytest.mark
.parametrize("table, db_filter, expected_data", [
326 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
327 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
328 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
329 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
330 def test_del_list_with_non_empty_db(db_memory_with_data
, table
, db_filter
, expected_data
):
331 result
= db_memory_with_data
.del_list(table
, db_filter
)
332 assert result
["deleted"] == (3 - len(expected_data
))
333 assert len(db_memory_with_data
.db
) == 1
334 assert table
in db_memory_with_data
.db
335 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
336 for data
in expected_data
:
337 assert data
in db_memory_with_data
.db
[table
]
340 @pytest.mark
.parametrize("table, db_filter", [
342 ("test", {"_id": 1}),
343 ("test", {"_id": 2}),
344 ("test", {"data": 1}),
345 ("test", {"data": 2}),
346 ("test", {"_id": 1, "data": 1}),
347 ("test", {"_id": 2, "data": 2})])
348 def test_del_list_with_empty_db(db_memory
, table
, db_filter
):
349 result
= db_memory
.del_list(table
, db_filter
)
350 assert result
['deleted'] == 0
353 def test_del_list_generic_exception(db_memory_with_data
):
356 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
357 with pytest
.raises(DbException
) as excinfo
:
358 db_memory_with_data
.del_list(table
, db_filter
)
359 assert str(excinfo
.value
) == empty_exception_message()
360 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
363 @pytest.mark
.parametrize("table, db_filter, data", [
364 ("test", {}, {"_id": 1, "data": 1}),
365 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
366 ("test", {"data": 1}, {"_id": 1, "data": 1}),
367 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
368 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
369 ("test", {"data": 2}, {"_id": 2, "data": 2}),
370 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
371 def test_del_one(db_memory_with_data
, table
, db_filter
, data
):
372 result
= db_memory_with_data
.del_one(table
, db_filter
)
373 assert result
== {"deleted": 1}
374 assert len(db_memory_with_data
.db
) == 1
375 assert table
in db_memory_with_data
.db
376 assert len(db_memory_with_data
.db
[table
]) == 2
377 assert data
not in db_memory_with_data
.db
[table
]
380 @pytest.mark
.parametrize("table, db_filter", [
382 ("test", {"_id": 1}),
383 ("test", {"_id": 2}),
384 ("test", {"data": 1}),
385 ("test", {"data": 2}),
386 ("test", {"_id": 1, "data": 1}),
387 ("test", {"_id": 2, "data": 2}),
389 ("test_table", {"_id": 1}),
390 ("test_table", {"_id": 2}),
391 ("test_table", {"data": 1}),
392 ("test_table", {"data": 2}),
393 ("test_table", {"_id": 1, "data": 1}),
394 ("test_table", {"_id": 2, "data": 2})])
395 def test_del_one_with_empty_db_exception(db_memory
, table
, db_filter
):
396 with pytest
.raises(DbException
) as excinfo
:
397 db_memory
.del_one(table
, db_filter
)
398 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(db_filter
))
399 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
402 @pytest.mark
.parametrize("table, db_filter", [
404 ("test", {"_id": 1}),
405 ("test", {"_id": 2}),
406 ("test", {"data": 1}),
407 ("test", {"data": 2}),
408 ("test", {"_id": 1, "data": 1}),
409 ("test", {"_id": 2, "data": 2}),
411 ("test_table", {"_id": 1}),
412 ("test_table", {"_id": 2}),
413 ("test_table", {"data": 1}),
414 ("test_table", {"data": 2}),
415 ("test_table", {"_id": 1, "data": 1}),
416 ("test_table", {"_id": 2, "data": 2})])
417 def test_del_one_with_empty_db_none(db_memory
, table
, db_filter
):
418 result
= db_memory
.del_one(table
, db_filter
, fail_on_empty
=False)
419 assert result
is None
422 @pytest.mark
.parametrize("table, db_filter", [
423 ("test", {"_id": 4}),
424 ("test", {"_id": 5}),
425 ("test", {"data": 4}),
426 ("test", {"data": 5}),
427 ("test", {"_id": 1, "data": 2}),
428 ("test", {"_id": 2, "data": 3}),
430 ("test_table", {"_id": 1}),
431 ("test_table", {"_id": 2}),
432 ("test_table", {"data": 1}),
433 ("test_table", {"data": 2}),
434 ("test_table", {"_id": 1, "data": 1}),
435 ("test_table", {"_id": 2, "data": 2})])
436 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
437 with pytest
.raises(DbException
) as excinfo
:
438 db_memory_with_data
.del_one(table
, db_filter
)
439 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(db_filter
))
440 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
443 @pytest.mark
.parametrize("table, db_filter", [
444 ("test", {"_id": 4}),
445 ("test", {"_id": 5}),
446 ("test", {"data": 4}),
447 ("test", {"data": 5}),
448 ("test", {"_id": 1, "data": 2}),
449 ("test", {"_id": 2, "data": 3}),
451 ("test_table", {"_id": 1}),
452 ("test_table", {"_id": 2}),
453 ("test_table", {"data": 1}),
454 ("test_table", {"data": 2}),
455 ("test_table", {"_id": 1, "data": 1}),
456 ("test_table", {"_id": 2, "data": 2})])
457 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
458 result
= db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=False)
459 assert result
is None
462 @pytest.mark
.parametrize("fail_on_empty", [
465 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
468 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
469 with pytest
.raises(DbException
) as excinfo
:
470 db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=fail_on_empty
)
471 assert str(excinfo
.value
) == empty_exception_message()
472 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
475 @pytest.mark
.parametrize("table, _id, indata", [
476 ("test", 1, {"_id": 1, "data": 42}),
477 ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
478 ("test", 1, {"_id": 1}),
479 ("test", 2, {"_id": 2, "data": 42}),
480 ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
481 ("test", 2, {"_id": 2}),
482 ("test", 3, {"_id": 3, "data": 42}),
483 ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
484 ("test", 3, {"_id": 3})])
485 def test_replace(db_memory_with_data
, table
, _id
, indata
):
486 result
= db_memory_with_data
.replace(table
, _id
, indata
)
487 assert result
== {"updated": 1}
488 assert len(db_memory_with_data
.db
) == 1
489 assert table
in db_memory_with_data
.db
490 assert len(db_memory_with_data
.db
[table
]) == 3
491 assert indata
in db_memory_with_data
.db
[table
]
494 @pytest.mark
.parametrize("table, _id, indata", [
495 ("test", 1, {"_id": 1, "data": 42}),
496 ("test", 2, {"_id": 2}),
497 ("test", 3, {"_id": 3})])
498 def test_replace_without_data_exception(db_memory
, table
, _id
, indata
):
499 with pytest
.raises(DbException
) as excinfo
:
500 db_memory
.replace(table
, _id
, indata
, fail_on_empty
=True)
501 assert str(excinfo
.value
) == (replace_exception_message(_id
))
502 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
505 @pytest.mark
.parametrize("table, _id, indata", [
506 ("test", 1, {"_id": 1, "data": 42}),
507 ("test", 2, {"_id": 2}),
508 ("test", 3, {"_id": 3})])
509 def test_replace_without_data_none(db_memory
, table
, _id
, indata
):
510 result
= db_memory
.replace(table
, _id
, indata
, fail_on_empty
=False)
511 assert result
is None
514 @pytest.mark
.parametrize("table, _id, indata", [
515 ("test", 11, {"_id": 11, "data": 42}),
516 ("test", 12, {"_id": 12}),
517 ("test", 33, {"_id": 33})])
518 def test_replace_with_data_exception(db_memory_with_data
, table
, _id
, indata
):
519 with pytest
.raises(DbException
) as excinfo
:
520 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=True)
521 assert str(excinfo
.value
) == (replace_exception_message(_id
))
522 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
525 @pytest.mark
.parametrize("table, _id, indata", [
526 ("test", 11, {"_id": 11, "data": 42}),
527 ("test", 12, {"_id": 12}),
528 ("test", 33, {"_id": 33})])
529 def test_replace_with_data_none(db_memory_with_data
, table
, _id
, indata
):
530 result
= db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=False)
531 assert result
is None
534 @pytest.mark
.parametrize("fail_on_empty", [
537 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
540 indata
= {'_id': 1, 'data': 1}
541 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
542 with pytest
.raises(DbException
) as excinfo
:
543 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=fail_on_empty
)
544 assert str(excinfo
.value
) == empty_exception_message()
545 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
548 @pytest.mark
.parametrize("table, id, data", [
549 ("test", "1", {"data": 1}),
550 ("test", "1", {"data": 2}),
551 ("test", "2", {"data": 1}),
552 ("test", "2", {"data": 2}),
553 ("test_table", "1", {"data": 1}),
554 ("test_table", "1", {"data": 2}),
555 ("test_table", "2", {"data": 1}),
556 ("test_table", "2", {"data": 2}),
557 ("test", "1", {"data_1": 1, "data_2": 2}),
558 ("test", "1", {"data_1": 2, "data_2": 1}),
559 ("test", "2", {"data_1": 1, "data_2": 2}),
560 ("test", "2", {"data_1": 2, "data_2": 1}),
561 ("test_table", "1", {"data_1": 1, "data_2": 2}),
562 ("test_table", "1", {"data_1": 2, "data_2": 1}),
563 ("test_table", "2", {"data_1": 1, "data_2": 2}),
564 ("test_table", "2", {"data_1": 2, "data_2": 1})])
565 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
566 data_to_insert
= data
567 data_to_insert
['_id'] = id
568 returned_id
= db_memory
.create(table
, data_to_insert
)
569 assert returned_id
== id
570 assert len(db_memory
.db
) == 1
571 assert table
in db_memory
.db
572 assert len(db_memory
.db
[table
]) == 1
573 assert data_to_insert
in db_memory
.db
[table
]
576 @pytest.mark
.parametrize("table, id, data", [
577 ("test", "4", {"data": 1}),
578 ("test", "5", {"data": 2}),
579 ("test", "4", {"data": 1}),
580 ("test", "5", {"data": 2}),
581 ("test_table", "4", {"data": 1}),
582 ("test_table", "5", {"data": 2}),
583 ("test_table", "4", {"data": 1}),
584 ("test_table", "5", {"data": 2}),
585 ("test", "4", {"data_1": 1, "data_2": 2}),
586 ("test", "5", {"data_1": 2, "data_2": 1}),
587 ("test", "4", {"data_1": 1, "data_2": 2}),
588 ("test", "5", {"data_1": 2, "data_2": 1}),
589 ("test_table", "4", {"data_1": 1, "data_2": 2}),
590 ("test_table", "5", {"data_1": 2, "data_2": 1}),
591 ("test_table", "4", {"data_1": 1, "data_2": 2}),
592 ("test_table", "5", {"data_1": 2, "data_2": 1})])
593 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
594 data_to_insert
= data
595 data_to_insert
['_id'] = id
596 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
597 assert returned_id
== id
598 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
599 assert table
in db_memory_with_data
.db
600 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
601 assert data_to_insert
in db_memory_with_data
.db
[table
]
604 @pytest.mark
.parametrize("table, data", [
605 ("test", {"data": 1}),
606 ("test", {"data": 2}),
607 ("test", {"data": 1}),
608 ("test", {"data": 2}),
609 ("test_table", {"data": 1}),
610 ("test_table", {"data": 2}),
611 ("test_table", {"data": 1}),
612 ("test_table", {"data": 2}),
613 ("test", {"data_1": 1, "data_2": 2}),
614 ("test", {"data_1": 2, "data_2": 1}),
615 ("test", {"data_1": 1, "data_2": 2}),
616 ("test", {"data_1": 2, "data_2": 1}),
617 ("test_table", {"data_1": 1, "data_2": 2}),
618 ("test_table", {"data_1": 2, "data_2": 1}),
619 ("test_table", {"data_1": 1, "data_2": 2}),
620 ("test_table", {"data_1": 2, "data_2": 1})])
621 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
622 returned_id
= db_memory
.create(table
, data
)
623 assert len(db_memory
.db
) == 1
624 assert table
in db_memory
.db
625 assert len(db_memory
.db
[table
]) == 1
627 data_inserted
['_id'] = returned_id
628 assert data_inserted
in db_memory
.db
[table
]
631 @pytest.mark
.parametrize("table, data", [
632 ("test", {"data": 1}),
633 ("test", {"data": 2}),
634 ("test", {"data": 1}),
635 ("test", {"data": 2}),
636 ("test_table", {"data": 1}),
637 ("test_table", {"data": 2}),
638 ("test_table", {"data": 1}),
639 ("test_table", {"data": 2}),
640 ("test", {"data_1": 1, "data_2": 2}),
641 ("test", {"data_1": 2, "data_2": 1}),
642 ("test", {"data_1": 1, "data_2": 2}),
643 ("test", {"data_1": 2, "data_2": 1}),
644 ("test_table", {"data_1": 1, "data_2": 2}),
645 ("test_table", {"data_1": 2, "data_2": 1}),
646 ("test_table", {"data_1": 1, "data_2": 2}),
647 ("test_table", {"data_1": 2, "data_2": 1})])
648 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
649 returned_id
= db_memory_with_data
.create(table
, data
)
650 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
651 assert table
in db_memory_with_data
.db
652 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
654 data_inserted
['_id'] = returned_id
655 assert data_inserted
in db_memory_with_data
.db
[table
]
658 def test_create_with_exception(db_memory
):
660 data
= {"_id": 1, "data": 1}
661 db_memory
.db
= MagicMock()
662 db_memory
.db
.__contains
__.side_effect
= Exception()
663 with pytest
.raises(DbException
) as excinfo
:
664 db_memory
.create(table
, data
)
665 assert str(excinfo
.value
) == empty_exception_message()
666 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
669 @pytest.mark
.parametrize("db_content, update_dict, expected, message", [
670 ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
671 ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
672 ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
673 ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
674 ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
675 ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
676 ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
677 ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
678 ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
679 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
680 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
681 ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
682 ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key")])
683 def test_set_one(db_memory
, db_content
, update_dict
, expected
, message
):
684 db_memory
._find
= Mock(return_value
=((0, db_content
), ))
686 with pytest
.raises(DbException
) as excinfo
:
687 db_memory
.set_one("table", {}, update_dict
)
688 assert (excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
), message
690 db_memory
.set_one("table", {}, update_dict
)
691 assert (db_content
== expected
), message
694 class TestDbMemory(unittest
.TestCase
):
695 # TODO to delete. This is cover with pytest test_set_one.
696 def test_set_one(self
):
698 # (database content, set-content, expected database content (None=fails), message)
699 ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
700 ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
701 ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
702 ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
703 ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
704 ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
706 ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
707 ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
708 ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
709 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
710 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
711 ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
712 ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key"),
715 db_men
._find
= Mock()
716 for db_content
, update_dict
, expected
, message
in test_set
:
717 db_men
._find
.return_value
= ((0, db_content
), )
719 self
.assertRaises(DbException
, db_men
.set_one
, "table", {}, update_dict
)
721 db_men
.set_one("table", {}, update_dict
)
722 self
.assertEqual(db_content
, expected
, message
)