552e8f12c8a36381e14a178e031b555a41df6514
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}]},
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.neq": 2}, [1, 3, 4, 5, 6, 7, 8]),
180 ({"data.data2.data3": [2, 3]}, [2, 3]),
181 ({"data.data2.data3.gt": 4}, [5, 6, 7]),
182 ({"data.data2.data3.gte": 4}, [4, 5, 6, 7]),
183 ({"data.data2.data3.lt": 4}, [1, 2, 3]),
184 ({"data.data2.data3.lte": 4}, [1, 2, 3, 4]),
185 ({"data.data2.data3.lte": 4.5}, [1, 2, 3, 4]),
186 ({"data.data2.data3.gt": "text"}, []),
187 ({"text.eq": "sometext"}, [1]),
188 ({"text.neq": "sometext"}, [2, 3, 4, 5, 6, 7, 8]),
189 ({"text.eq": "somet"}, []),
190 ({"text.gte": "a"}, [1]),
191 ({"text.gte": "somet"}, [1]),
192 ({"text.gte": "sometext"}, [1]),
193 ({"text.lt": "somet"}, []),
194 ({"data.data2.data3": 2, "data.data2.data4": None}, [2]),
195 ({"data.data2.data3": 2, "data.data2.data4": 5}, []),
196 ({"data.data2.data3": 4}, [4]),
197 ({"data.data2.data3": [3, 4, "e"]}, [3, 4]),
198 ({"data.data2.data3": None}, [8]),
199 ({"data.data2": "4"}, []),
200 ({"list.0.a": 1}, [1, 6]),
201 ({"list.ANYINDEX.a": 1}, [1]),
202 ({"list.a": 3, "list.b": 1}, [8]),
203 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 1}, []),
204 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.a": 3}, [8]),
205 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.b": 0}, [8]),
206 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.ANYINDEX.a": 0, "list.ANYINDEX.c.ANYINDEX.b": "v"}, [8]),
207 ({"list.ANYINDEX.a": 3, "list.ANYINDEX.c.ANYINDEX.a": 0, "list.ANYINDEX.c.ANYINDEX.b": 1}, []),
208 ({"list.c.b": 1}, [8]),
209 ({"list.c.b": None}, [1, 2, 3, 4, 5, 6, 7]),
210 # ({"data.data2.data3": 4}, []),
211 # ({"data.data2.data3": 4}, []),
213 def test_get_list(db_memory_with_many_data
, db_filter
, expected_ids
):
214 result
= db_memory_with_many_data
.get_list("test", db_filter
)
215 assert isinstance(result
, list)
216 result_ids
= [item
["_id"] for item
in result
]
217 assert len(result
) == len(expected_ids
), "for db_filter={} result={} expected_ids={}".format(db_filter
, result
,
219 assert result_ids
== expected_ids
220 for i
in range(len(result
)):
221 assert result
[i
] in db_memory_with_many_data
.db
["test"]
223 assert len(db_memory_with_many_data
.db
) == 1
224 assert "test" in db_memory_with_many_data
.db
225 assert len(db_memory_with_many_data
.db
["test"]) == 8
226 result
= db_memory_with_many_data
.count("test", db_filter
)
227 assert result
== len(expected_ids
)
230 @pytest.mark
.parametrize("table, db_filter, expected_data", [
231 ("test", {}, {"_id": 1, "data": 1})])
232 def test_get_one_with_multiple_results(db_memory_with_data
, table
, db_filter
, expected_data
):
233 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_more
=False)
234 assert result
== expected_data
235 assert len(db_memory_with_data
.db
) == 1
236 assert table
in db_memory_with_data
.db
237 assert len(db_memory_with_data
.db
[table
]) == 3
238 assert result
in db_memory_with_data
.db
[table
]
241 def test_get_one_with_multiple_results_exception(db_memory_with_data
):
244 with pytest
.raises(DbException
) as excinfo
:
245 db_memory_with_data
.get_one(table
, db_filter
)
246 assert str(excinfo
.value
) == (empty_exception_message() + get_one_multiple_exception_message(db_filter
))
247 # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
250 @pytest.mark
.parametrize("table, db_filter", [
251 ("test", {"_id": 4}),
252 ("test", {"data": 4}),
253 ("test", {"_id": 4, "data": 4}),
254 ("test_table", {"_id": 4}),
255 ("test_table", {"data": 4}),
256 ("test_table", {"_id": 4, "data": 4})])
257 def test_get_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
258 with pytest
.raises(DbException
) as excinfo
:
259 db_memory_with_data
.get_one(table
, db_filter
)
260 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(db_filter
))
261 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
264 @pytest.mark
.parametrize("table, db_filter", [
265 ("test", {"_id": 4}),
266 ("test", {"data": 4}),
267 ("test", {"_id": 4, "data": 4}),
268 ("test_table", {"_id": 4}),
269 ("test_table", {"data": 4}),
270 ("test_table", {"_id": 4, "data": 4})])
271 def test_get_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
272 result
= db_memory_with_data
.get_one(table
, db_filter
, fail_on_empty
=False)
273 assert result
is None
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_empty_db_exception(db_memory
, table
, db_filter
):
284 with pytest
.raises(DbException
) as excinfo
:
285 db_memory
.get_one(table
, db_filter
)
286 assert str(excinfo
.value
) == (empty_exception_message() + get_one_exception_message(db_filter
))
287 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
290 @pytest.mark
.parametrize("table, db_filter", [
291 ("test", {"_id": 4}),
292 ("test", {"data": 4}),
293 ("test", {"_id": 4, "data": 4}),
294 ("test_table", {"_id": 4}),
295 ("test_table", {"data": 4}),
296 ("test_table", {"_id": 4, "data": 4})])
297 def test_get_one_with_empty_db_none(db_memory
, table
, db_filter
):
298 result
= db_memory
.get_one(table
, db_filter
, fail_on_empty
=False)
299 assert result
is None
302 def test_get_one_generic_exception(db_memory_with_data
):
305 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
306 with pytest
.raises(DbException
) as excinfo
:
307 db_memory_with_data
.get_one(table
, db_filter
)
308 assert str(excinfo
.value
) == empty_exception_message()
309 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
312 @pytest.mark
.parametrize("table, db_filter, expected_data", [
314 ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
315 ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
316 ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
317 ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
318 def test_del_list_with_non_empty_db(db_memory_with_data
, table
, db_filter
, expected_data
):
319 result
= db_memory_with_data
.del_list(table
, db_filter
)
320 assert result
["deleted"] == (3 - len(expected_data
))
321 assert len(db_memory_with_data
.db
) == 1
322 assert table
in db_memory_with_data
.db
323 assert len(db_memory_with_data
.db
[table
]) == len(expected_data
)
324 for data
in expected_data
:
325 assert data
in db_memory_with_data
.db
[table
]
328 @pytest.mark
.parametrize("table, db_filter", [
330 ("test", {"_id": 1}),
331 ("test", {"_id": 2}),
332 ("test", {"data": 1}),
333 ("test", {"data": 2}),
334 ("test", {"_id": 1, "data": 1}),
335 ("test", {"_id": 2, "data": 2})])
336 def test_del_list_with_empty_db(db_memory
, table
, db_filter
):
337 result
= db_memory
.del_list(table
, db_filter
)
338 assert result
['deleted'] == 0
341 def test_del_list_generic_exception(db_memory_with_data
):
344 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
345 with pytest
.raises(DbException
) as excinfo
:
346 db_memory_with_data
.del_list(table
, db_filter
)
347 assert str(excinfo
.value
) == empty_exception_message()
348 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
351 @pytest.mark
.parametrize("table, db_filter, data", [
352 ("test", {}, {"_id": 1, "data": 1}),
353 ("test", {"_id": 1}, {"_id": 1, "data": 1}),
354 ("test", {"data": 1}, {"_id": 1, "data": 1}),
355 ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
356 ("test", {"_id": 2}, {"_id": 2, "data": 2}),
357 ("test", {"data": 2}, {"_id": 2, "data": 2}),
358 ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
359 def test_del_one(db_memory_with_data
, table
, db_filter
, data
):
360 result
= db_memory_with_data
.del_one(table
, db_filter
)
361 assert result
== {"deleted": 1}
362 assert len(db_memory_with_data
.db
) == 1
363 assert table
in db_memory_with_data
.db
364 assert len(db_memory_with_data
.db
[table
]) == 2
365 assert data
not in db_memory_with_data
.db
[table
]
368 @pytest.mark
.parametrize("table, db_filter", [
370 ("test", {"_id": 1}),
371 ("test", {"_id": 2}),
372 ("test", {"data": 1}),
373 ("test", {"data": 2}),
374 ("test", {"_id": 1, "data": 1}),
375 ("test", {"_id": 2, "data": 2}),
377 ("test_table", {"_id": 1}),
378 ("test_table", {"_id": 2}),
379 ("test_table", {"data": 1}),
380 ("test_table", {"data": 2}),
381 ("test_table", {"_id": 1, "data": 1}),
382 ("test_table", {"_id": 2, "data": 2})])
383 def test_del_one_with_empty_db_exception(db_memory
, table
, db_filter
):
384 with pytest
.raises(DbException
) as excinfo
:
385 db_memory
.del_one(table
, db_filter
)
386 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(db_filter
))
387 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
390 @pytest.mark
.parametrize("table, db_filter", [
392 ("test", {"_id": 1}),
393 ("test", {"_id": 2}),
394 ("test", {"data": 1}),
395 ("test", {"data": 2}),
396 ("test", {"_id": 1, "data": 1}),
397 ("test", {"_id": 2, "data": 2}),
399 ("test_table", {"_id": 1}),
400 ("test_table", {"_id": 2}),
401 ("test_table", {"data": 1}),
402 ("test_table", {"data": 2}),
403 ("test_table", {"_id": 1, "data": 1}),
404 ("test_table", {"_id": 2, "data": 2})])
405 def test_del_one_with_empty_db_none(db_memory
, table
, db_filter
):
406 result
= db_memory
.del_one(table
, db_filter
, fail_on_empty
=False)
407 assert result
is None
410 @pytest.mark
.parametrize("table, db_filter", [
411 ("test", {"_id": 4}),
412 ("test", {"_id": 5}),
413 ("test", {"data": 4}),
414 ("test", {"data": 5}),
415 ("test", {"_id": 1, "data": 2}),
416 ("test", {"_id": 2, "data": 3}),
418 ("test_table", {"_id": 1}),
419 ("test_table", {"_id": 2}),
420 ("test_table", {"data": 1}),
421 ("test_table", {"data": 2}),
422 ("test_table", {"_id": 1, "data": 1}),
423 ("test_table", {"_id": 2, "data": 2})])
424 def test_del_one_with_non_empty_db_exception(db_memory_with_data
, table
, db_filter
):
425 with pytest
.raises(DbException
) as excinfo
:
426 db_memory_with_data
.del_one(table
, db_filter
)
427 assert str(excinfo
.value
) == (empty_exception_message() + del_one_exception_message(db_filter
))
428 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
431 @pytest.mark
.parametrize("table, db_filter", [
432 ("test", {"_id": 4}),
433 ("test", {"_id": 5}),
434 ("test", {"data": 4}),
435 ("test", {"data": 5}),
436 ("test", {"_id": 1, "data": 2}),
437 ("test", {"_id": 2, "data": 3}),
439 ("test_table", {"_id": 1}),
440 ("test_table", {"_id": 2}),
441 ("test_table", {"data": 1}),
442 ("test_table", {"data": 2}),
443 ("test_table", {"_id": 1, "data": 1}),
444 ("test_table", {"_id": 2, "data": 2})])
445 def test_del_one_with_non_empty_db_none(db_memory_with_data
, table
, db_filter
):
446 result
= db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=False)
447 assert result
is None
450 @pytest.mark
.parametrize("fail_on_empty", [
453 def test_del_one_generic_exception(db_memory_with_data
, fail_on_empty
):
456 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
457 with pytest
.raises(DbException
) as excinfo
:
458 db_memory_with_data
.del_one(table
, db_filter
, fail_on_empty
=fail_on_empty
)
459 assert str(excinfo
.value
) == empty_exception_message()
460 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
463 @pytest.mark
.parametrize("table, _id, indata", [
464 ("test", 1, {"_id": 1, "data": 42}),
465 ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
466 ("test", 1, {"_id": 1}),
467 ("test", 2, {"_id": 2, "data": 42}),
468 ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
469 ("test", 2, {"_id": 2}),
470 ("test", 3, {"_id": 3, "data": 42}),
471 ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
472 ("test", 3, {"_id": 3})])
473 def test_replace(db_memory_with_data
, table
, _id
, indata
):
474 result
= db_memory_with_data
.replace(table
, _id
, indata
)
475 assert result
== {"updated": 1}
476 assert len(db_memory_with_data
.db
) == 1
477 assert table
in db_memory_with_data
.db
478 assert len(db_memory_with_data
.db
[table
]) == 3
479 assert indata
in db_memory_with_data
.db
[table
]
482 @pytest.mark
.parametrize("table, _id, indata", [
483 ("test", 1, {"_id": 1, "data": 42}),
484 ("test", 2, {"_id": 2}),
485 ("test", 3, {"_id": 3})])
486 def test_replace_without_data_exception(db_memory
, table
, _id
, indata
):
487 with pytest
.raises(DbException
) as excinfo
:
488 db_memory
.replace(table
, _id
, indata
, fail_on_empty
=True)
489 assert str(excinfo
.value
) == (replace_exception_message(_id
))
490 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
493 @pytest.mark
.parametrize("table, _id, indata", [
494 ("test", 1, {"_id": 1, "data": 42}),
495 ("test", 2, {"_id": 2}),
496 ("test", 3, {"_id": 3})])
497 def test_replace_without_data_none(db_memory
, table
, _id
, indata
):
498 result
= db_memory
.replace(table
, _id
, indata
, fail_on_empty
=False)
499 assert result
is None
502 @pytest.mark
.parametrize("table, _id, indata", [
503 ("test", 11, {"_id": 11, "data": 42}),
504 ("test", 12, {"_id": 12}),
505 ("test", 33, {"_id": 33})])
506 def test_replace_with_data_exception(db_memory_with_data
, table
, _id
, indata
):
507 with pytest
.raises(DbException
) as excinfo
:
508 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=True)
509 assert str(excinfo
.value
) == (replace_exception_message(_id
))
510 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
513 @pytest.mark
.parametrize("table, _id, indata", [
514 ("test", 11, {"_id": 11, "data": 42}),
515 ("test", 12, {"_id": 12}),
516 ("test", 33, {"_id": 33})])
517 def test_replace_with_data_none(db_memory_with_data
, table
, _id
, indata
):
518 result
= db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=False)
519 assert result
is None
522 @pytest.mark
.parametrize("fail_on_empty", [
525 def test_replace_generic_exception(db_memory_with_data
, fail_on_empty
):
528 indata
= {'_id': 1, 'data': 1}
529 db_memory_with_data
._find
= MagicMock(side_effect
=Exception())
530 with pytest
.raises(DbException
) as excinfo
:
531 db_memory_with_data
.replace(table
, _id
, indata
, fail_on_empty
=fail_on_empty
)
532 assert str(excinfo
.value
) == empty_exception_message()
533 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
536 @pytest.mark
.parametrize("table, id, data", [
537 ("test", "1", {"data": 1}),
538 ("test", "1", {"data": 2}),
539 ("test", "2", {"data": 1}),
540 ("test", "2", {"data": 2}),
541 ("test_table", "1", {"data": 1}),
542 ("test_table", "1", {"data": 2}),
543 ("test_table", "2", {"data": 1}),
544 ("test_table", "2", {"data": 2}),
545 ("test", "1", {"data_1": 1, "data_2": 2}),
546 ("test", "1", {"data_1": 2, "data_2": 1}),
547 ("test", "2", {"data_1": 1, "data_2": 2}),
548 ("test", "2", {"data_1": 2, "data_2": 1}),
549 ("test_table", "1", {"data_1": 1, "data_2": 2}),
550 ("test_table", "1", {"data_1": 2, "data_2": 1}),
551 ("test_table", "2", {"data_1": 1, "data_2": 2}),
552 ("test_table", "2", {"data_1": 2, "data_2": 1})])
553 def test_create_with_empty_db_with_id(db_memory
, table
, id, data
):
554 data_to_insert
= data
555 data_to_insert
['_id'] = id
556 returned_id
= db_memory
.create(table
, data_to_insert
)
557 assert returned_id
== id
558 assert len(db_memory
.db
) == 1
559 assert table
in db_memory
.db
560 assert len(db_memory
.db
[table
]) == 1
561 assert data_to_insert
in db_memory
.db
[table
]
564 @pytest.mark
.parametrize("table, id, data", [
565 ("test", "4", {"data": 1}),
566 ("test", "5", {"data": 2}),
567 ("test", "4", {"data": 1}),
568 ("test", "5", {"data": 2}),
569 ("test_table", "4", {"data": 1}),
570 ("test_table", "5", {"data": 2}),
571 ("test_table", "4", {"data": 1}),
572 ("test_table", "5", {"data": 2}),
573 ("test", "4", {"data_1": 1, "data_2": 2}),
574 ("test", "5", {"data_1": 2, "data_2": 1}),
575 ("test", "4", {"data_1": 1, "data_2": 2}),
576 ("test", "5", {"data_1": 2, "data_2": 1}),
577 ("test_table", "4", {"data_1": 1, "data_2": 2}),
578 ("test_table", "5", {"data_1": 2, "data_2": 1}),
579 ("test_table", "4", {"data_1": 1, "data_2": 2}),
580 ("test_table", "5", {"data_1": 2, "data_2": 1})])
581 def test_create_with_non_empty_db_with_id(db_memory_with_data
, table
, id, data
):
582 data_to_insert
= data
583 data_to_insert
['_id'] = id
584 returned_id
= db_memory_with_data
.create(table
, data_to_insert
)
585 assert returned_id
== id
586 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
587 assert table
in db_memory_with_data
.db
588 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
589 assert data_to_insert
in db_memory_with_data
.db
[table
]
592 @pytest.mark
.parametrize("table, data", [
593 ("test", {"data": 1}),
594 ("test", {"data": 2}),
595 ("test", {"data": 1}),
596 ("test", {"data": 2}),
597 ("test_table", {"data": 1}),
598 ("test_table", {"data": 2}),
599 ("test_table", {"data": 1}),
600 ("test_table", {"data": 2}),
601 ("test", {"data_1": 1, "data_2": 2}),
602 ("test", {"data_1": 2, "data_2": 1}),
603 ("test", {"data_1": 1, "data_2": 2}),
604 ("test", {"data_1": 2, "data_2": 1}),
605 ("test_table", {"data_1": 1, "data_2": 2}),
606 ("test_table", {"data_1": 2, "data_2": 1}),
607 ("test_table", {"data_1": 1, "data_2": 2}),
608 ("test_table", {"data_1": 2, "data_2": 1})])
609 def test_create_with_empty_db_without_id(db_memory
, table
, data
):
610 returned_id
= db_memory
.create(table
, data
)
611 assert len(db_memory
.db
) == 1
612 assert table
in db_memory
.db
613 assert len(db_memory
.db
[table
]) == 1
615 data_inserted
['_id'] = returned_id
616 assert data_inserted
in db_memory
.db
[table
]
619 @pytest.mark
.parametrize("table, data", [
620 ("test", {"data": 1}),
621 ("test", {"data": 2}),
622 ("test", {"data": 1}),
623 ("test", {"data": 2}),
624 ("test_table", {"data": 1}),
625 ("test_table", {"data": 2}),
626 ("test_table", {"data": 1}),
627 ("test_table", {"data": 2}),
628 ("test", {"data_1": 1, "data_2": 2}),
629 ("test", {"data_1": 2, "data_2": 1}),
630 ("test", {"data_1": 1, "data_2": 2}),
631 ("test", {"data_1": 2, "data_2": 1}),
632 ("test_table", {"data_1": 1, "data_2": 2}),
633 ("test_table", {"data_1": 2, "data_2": 1}),
634 ("test_table", {"data_1": 1, "data_2": 2}),
635 ("test_table", {"data_1": 2, "data_2": 1})])
636 def test_create_with_non_empty_db_without_id(db_memory_with_data
, table
, data
):
637 returned_id
= db_memory_with_data
.create(table
, data
)
638 assert len(db_memory_with_data
.db
) == (1 if table
== 'test' else 2)
639 assert table
in db_memory_with_data
.db
640 assert len(db_memory_with_data
.db
[table
]) == (4 if table
== 'test' else 1)
642 data_inserted
['_id'] = returned_id
643 assert data_inserted
in db_memory_with_data
.db
[table
]
646 def test_create_with_exception(db_memory
):
648 data
= {"_id": 1, "data": 1}
649 db_memory
.db
= MagicMock()
650 db_memory
.db
.__contains
__.side_effect
= Exception()
651 with pytest
.raises(DbException
) as excinfo
:
652 db_memory
.create(table
, data
)
653 assert str(excinfo
.value
) == empty_exception_message()
654 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
657 @pytest.mark
.parametrize("db_content, update_dict, expected, message", [
658 ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
659 ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
660 ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
661 ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
662 ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
663 ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
664 ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
665 ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
666 ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
667 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
668 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
669 ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
670 ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key")])
671 def test_set_one(db_memory
, db_content
, update_dict
, expected
, message
):
672 db_memory
._find
= Mock(return_value
=((0, db_content
), ))
674 with pytest
.raises(DbException
) as excinfo
:
675 db_memory
.set_one("table", {}, update_dict
)
676 assert (excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
), message
678 db_memory
.set_one("table", {}, update_dict
)
679 assert (db_content
== expected
), message
682 class TestDbMemory(unittest
.TestCase
):
683 # TODO to delete. This is cover with pytest test_set_one.
684 def test_set_one(self
):
686 # (database content, set-content, expected database content (None=fails), message)
687 ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
688 ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
689 ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
690 ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
691 ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
692 ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
694 ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
695 ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
696 ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
697 {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
698 ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
699 ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
700 ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key"),
703 db_men
._find
= Mock()
704 for db_content
, update_dict
, expected
, message
in test_set
:
705 db_men
._find
.return_value
= ((0, db_content
), )
707 self
.assertRaises(DbException
, db_men
.set_one
, "table", {}, update_dict
)
709 db_men
.set_one("table", {}, update_dict
)
710 self
.assertEqual(db_content
, expected
, message
)