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}]},
  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.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}, []),
 212 ])
 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,
 218                                                                                                  result_ids)
 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"]
 222
 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)
 228
 229
 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]
 239
 240
 241 def test_get_one_with_multiple_results_exception(db_memory_with_data):
 242     table = "test"
 243     db_filter = {}
 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
 248
 249
 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
 262
 263
 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
 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_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
 288
 289
 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
 300
 301
 302 def test_get_one_generic_exception(db_memory_with_data):
 303     table = 'test'
 304     db_filter = {}
 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
 310
 311
 312 @pytest.mark.parametrize("table, db_filter, expected_data", [
 313     ("test", {}, []),
 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]
 326
 327
 328 @pytest.mark.parametrize("table, db_filter", [
 329     ("test", {}),
 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
 339
 340
 341 def test_del_list_generic_exception(db_memory_with_data):
 342     table = 'test'
 343     db_filter = {}
 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
 349
 350
 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]
 366
 367
 368 @pytest.mark.parametrize("table, db_filter", [
 369     ("test", {}),
 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}),
 376     ("test_table", {}),
 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
 388
 389
 390 @pytest.mark.parametrize("table, db_filter", [
 391     ("test", {}),
 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}),
 398     ("test_table", {}),
 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
 408
 409
 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}),
 417     ("test_table", {}),
 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
 429
 430
 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}),
 438     ("test_table", {}),
 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
 448
 449
 450 @pytest.mark.parametrize("fail_on_empty", [
 451     (True),
 452     (False)])
 453 def test_del_one_generic_exception(db_memory_with_data, fail_on_empty):
 454     table = 'test'
 455     db_filter = {}
 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
 461
 462
 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]
 480
 481
 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
 491
 492
 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
 500
 501
 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
 511
 512
 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
 520
 521
 522 @pytest.mark.parametrize("fail_on_empty", [
 523     True,
 524     False])
 525 def test_replace_generic_exception(db_memory_with_data, fail_on_empty):
 526     table = 'test'
 527     _id = {}
 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
 534
 535
 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]
 562
 563
 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]
 590
 591
 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
 614     data_inserted = data
 615     data_inserted['_id'] = returned_id
 616     assert data_inserted in db_memory.db[table]
 617
 618
 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)
 641     data_inserted = data
 642     data_inserted['_id'] = returned_id
 643     assert data_inserted in db_memory_with_data.db[table]
 644
 645
 646 def test_create_with_exception(db_memory):
 647     table = "test"
 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
 655
 656
 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), ))
 673     if expected is None:
 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
 677     else:
 678         db_memory.set_one("table", {}, update_dict)
 679         assert (db_content == expected), message
 680
 681
 682 class TestDbMemory(unittest.TestCase):
 683     # TODO to delete. This is cover with pytest test_set_one.
 684     def test_set_one(self):
 685         test_set = (
 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"),
 693
 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"),
 701         )
 702         db_men = DbMemory()
 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), )
 706             if expected is None:
 707                 self.assertRaises(DbException, db_men.set_one, "table", {}, update_dict)
 708             else:
 709                 db_men.set_one("table", {}, update_dict)
 710                 self.assertEqual(db_content, expected, message)