osm_common/tests/test_dbmemory.py

   1 # Copyright 2018 Whitestack, LLC
   2 # Copyright 2018 Telefonica S.A.
   3 #
   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
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
  18 ##
  19
  20 import http
  21 import logging
  22 import pytest
  23 import unittest
  24 from unittest.mock import Mock
  25
  26 from unittest.mock import MagicMock
  27 from osm_common.dbbase import DbException
  28 from osm_common.dbmemory import DbMemory
  29
  30 __author__ = 'Eduardo Sousa <eduardosousa@av.it.pt>'
  31
  32
  33 @pytest.fixture(scope="function", params=[True, False])
  34 def db_memory(request):
  35     db = DbMemory(lock=request.param)
  36     return db
  37
  38
  39 @pytest.fixture(scope="function", params=[True, False])
  40 def db_memory_with_data(request):
  41     db = DbMemory(lock=request.param)
  42
  43     db.create("test", {"_id": 1, "data": 1})
  44     db.create("test", {"_id": 2, "data": 2})
  45     db.create("test", {"_id": 3, "data": 3})
  46
  47     return db
  48
  49
  50 @pytest.fixture(scope="function")
  51 def db_memory_with_many_data(request):
  52     db = DbMemory(lock=False)
  53
  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}]},
  63     ])
  64     return db
  65
  66
  67 def empty_exception_message():
  68     return 'database exception '
  69
  70
  71 def get_one_exception_message(db_filter):
  72     return "database exception Not found entry with filter='{}'".format(db_filter)
  73
  74
  75 def get_one_multiple_exception_message(db_filter):
  76     return "database exception Found more than one entry with filter='{}'".format(db_filter)
  77
  78
  79 def del_one_exception_message(db_filter):
  80     return "database exception Not found entry with filter='{}'".format(db_filter)
  81
  82
  83 def replace_exception_message(value):
  84     return "database exception Not found entry with _id='{}'".format(value)
  85
  86
  87 def test_constructor():
  88     db = DbMemory()
  89     assert db.logger == logging.getLogger('db')
  90     assert db.db == {}
  91
  92
  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)
  97     assert db.db == {}
  98
  99
 100 def test_db_connect():
 101     logger_name = 'db_local'
 102     config = {'logger_name': logger_name}
 103     db = DbMemory()
 104     db.db_connect(config)
 105     assert db.logger == logging.getLogger(logger_name)
 106     assert db.db == {}
 107
 108
 109 def test_db_disconnect(db_memory):
 110     db_memory.db_disconnect()
 111
 112
 113 @pytest.mark.parametrize("table, db_filter", [
 114     ("test", {}),
 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
 121
 122
 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
 143
 144
 145 def test_get_list_exception(db_memory_with_data):
 146     table = 'test'
 147     db_filter = {}
 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
 153
 154
 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]
 172
 173
 174 @pytest.mark.parametrize("db_filter, expected_ids", [
 175     ({}, [1, 2, 3, 4, 5, 6, 7, 8]),
 176     ({"_id": 1}, [1]),
 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]),
 209     ({"list2": 1}, [2]),
 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}, []),
 224 ])
 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,
 230                                                                                                  result_ids)
 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"]
 234
 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)
 240
 241
 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]
 251
 252
 253 def test_get_one_with_multiple_results_exception(db_memory_with_data):
 254     table = "test"
 255     db_filter = {}
 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
 260
 261
 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
 274
 275
 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
 286
 287
 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
 300
 301
 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
 312
 313
 314 def test_get_one_generic_exception(db_memory_with_data):
 315     table = 'test'
 316     db_filter = {}
 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
 322
 323
 324 @pytest.mark.parametrize("table, db_filter, expected_data", [
 325     ("test", {}, []),
 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]
 338
 339
 340 @pytest.mark.parametrize("table, db_filter", [
 341     ("test", {}),
 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
 351
 352
 353 def test_del_list_generic_exception(db_memory_with_data):
 354     table = 'test'
 355     db_filter = {}
 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
 361
 362
 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]
 378
 379
 380 @pytest.mark.parametrize("table, db_filter", [
 381     ("test", {}),
 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}),
 388     ("test_table", {}),
 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
 400
 401
 402 @pytest.mark.parametrize("table, db_filter", [
 403     ("test", {}),
 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}),
 410     ("test_table", {}),
 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
 420
 421
 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}),
 429     ("test_table", {}),
 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
 441
 442
 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}),
 450     ("test_table", {}),
 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
 460
 461
 462 @pytest.mark.parametrize("fail_on_empty", [
 463     (True),
 464     (False)])
 465 def test_del_one_generic_exception(db_memory_with_data, fail_on_empty):
 466     table = 'test'
 467     db_filter = {}
 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
 473
 474
 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]
 492
 493
 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
 503
 504
 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
 512
 513
 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
 523
 524
 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
 532
 533
 534 @pytest.mark.parametrize("fail_on_empty", [
 535     True,
 536     False])
 537 def test_replace_generic_exception(db_memory_with_data, fail_on_empty):
 538     table = 'test'
 539     _id = {}
 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
 546
 547
 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]
 574
 575
 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]
 602
 603
 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
 626     data_inserted = data
 627     data_inserted['_id'] = returned_id
 628     assert data_inserted in db_memory.db[table]
 629
 630
 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)
 653     data_inserted = data
 654     data_inserted['_id'] = returned_id
 655     assert data_inserted in db_memory_with_data.db[table]
 656
 657
 658 def test_create_with_exception(db_memory):
 659     table = "test"
 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
 667
 668
 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), ))
 685     if expected is None:
 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
 689     else:
 690         db_memory.set_one("table", {}, update_dict)
 691         assert (db_content == expected), message
 692
 693
 694 class TestDbMemory(unittest.TestCase):
 695     # TODO to delete. This is cover with pytest test_set_one.
 696     def test_set_one(self):
 697         test_set = (
 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"),
 705
 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"),
 713         )
 714         db_men = DbMemory()
 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), )
 718             if expected is None:
 719                 self.assertRaises(DbException, db_men.set_one, "table", {}, update_dict)
 720             else:
 721                 db_men.set_one("table", {}, update_dict)
 722                 self.assertEqual(db_content, expected, message)