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
 227
 228 @pytest.mark.parametrize("table, db_filter, expected_data", [
 229     ("test", {}, {"_id": 1, "data": 1})])
 230 def test_get_one_with_multiple_results(db_memory_with_data, table, db_filter, expected_data):
 231     result = db_memory_with_data.get_one(table, db_filter, fail_on_more=False)
 232     assert result == expected_data
 233     assert len(db_memory_with_data.db) == 1
 234     assert table in db_memory_with_data.db
 235     assert len(db_memory_with_data.db[table]) == 3
 236     assert result in db_memory_with_data.db[table]
 237
 238
 239 def test_get_one_with_multiple_results_exception(db_memory_with_data):
 240     table = "test"
 241     db_filter = {}
 242     with pytest.raises(DbException) as excinfo:
 243         db_memory_with_data.get_one(table, db_filter)
 244     assert str(excinfo.value) == (empty_exception_message() + get_one_multiple_exception_message(db_filter))
 245     # assert excinfo.value.http_code == http.HTTPStatus.CONFLICT
 246
 247
 248 @pytest.mark.parametrize("table, db_filter", [
 249     ("test", {"_id": 4}),
 250     ("test", {"data": 4}),
 251     ("test", {"_id": 4, "data": 4}),
 252     ("test_table", {"_id": 4}),
 253     ("test_table", {"data": 4}),
 254     ("test_table", {"_id": 4, "data": 4})])
 255 def test_get_one_with_non_empty_db_exception(db_memory_with_data, table, 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_exception_message(db_filter))
 259     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 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_none(db_memory_with_data, table, db_filter):
 270     result = db_memory_with_data.get_one(table, db_filter, fail_on_empty=False)
 271     assert result is None
 272
 273
 274 @pytest.mark.parametrize("table, db_filter", [
 275     ("test", {"_id": 4}),
 276     ("test", {"data": 4}),
 277     ("test", {"_id": 4, "data": 4}),
 278     ("test_table", {"_id": 4}),
 279     ("test_table", {"data": 4}),
 280     ("test_table", {"_id": 4, "data": 4})])
 281 def test_get_one_with_empty_db_exception(db_memory, table, db_filter):
 282     with pytest.raises(DbException) as excinfo:
 283         db_memory.get_one(table, db_filter)
 284     assert str(excinfo.value) == (empty_exception_message() + get_one_exception_message(db_filter))
 285     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 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_none(db_memory, table, db_filter):
 296     result = db_memory.get_one(table, db_filter, fail_on_empty=False)
 297     assert result is None
 298
 299
 300 def test_get_one_generic_exception(db_memory_with_data):
 301     table = 'test'
 302     db_filter = {}
 303     db_memory_with_data._find = MagicMock(side_effect=Exception())
 304     with pytest.raises(DbException) as excinfo:
 305         db_memory_with_data.get_one(table, db_filter)
 306     assert str(excinfo.value) == empty_exception_message()
 307     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 308
 309
 310 @pytest.mark.parametrize("table, db_filter, expected_data", [
 311     ("test", {}, []),
 312     ("test", {"_id": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
 313     ("test", {"_id": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}]),
 314     ("test", {"_id": 1, "data": 1}, [{"_id": 2, "data": 2}, {"_id": 3, "data": 3}]),
 315     ("test", {"_id": 2, "data": 2}, [{"_id": 1, "data": 1}, {"_id": 3, "data": 3}])])
 316 def test_del_list_with_non_empty_db(db_memory_with_data, table, db_filter, expected_data):
 317     result = db_memory_with_data.del_list(table, db_filter)
 318     assert result["deleted"] == (3 - len(expected_data))
 319     assert len(db_memory_with_data.db) == 1
 320     assert table in db_memory_with_data.db
 321     assert len(db_memory_with_data.db[table]) == len(expected_data)
 322     for data in expected_data:
 323         assert data in db_memory_with_data.db[table]
 324
 325
 326 @pytest.mark.parametrize("table, db_filter", [
 327     ("test", {}),
 328     ("test", {"_id": 1}),
 329     ("test", {"_id": 2}),
 330     ("test", {"data": 1}),
 331     ("test", {"data": 2}),
 332     ("test", {"_id": 1, "data": 1}),
 333     ("test", {"_id": 2, "data": 2})])
 334 def test_del_list_with_empty_db(db_memory, table, db_filter):
 335     result = db_memory.del_list(table, db_filter)
 336     assert result['deleted'] == 0
 337
 338
 339 def test_del_list_generic_exception(db_memory_with_data):
 340     table = 'test'
 341     db_filter = {}
 342     db_memory_with_data._find = MagicMock(side_effect=Exception())
 343     with pytest.raises(DbException) as excinfo:
 344         db_memory_with_data.del_list(table, db_filter)
 345     assert str(excinfo.value) == empty_exception_message()
 346     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 347
 348
 349 @pytest.mark.parametrize("table, db_filter, data", [
 350     ("test", {}, {"_id": 1, "data": 1}),
 351     ("test", {"_id": 1}, {"_id": 1, "data": 1}),
 352     ("test", {"data": 1}, {"_id": 1, "data": 1}),
 353     ("test", {"_id": 1, "data": 1}, {"_id": 1, "data": 1}),
 354     ("test", {"_id": 2}, {"_id": 2, "data": 2}),
 355     ("test", {"data": 2}, {"_id": 2, "data": 2}),
 356     ("test", {"_id": 2, "data": 2}, {"_id": 2, "data": 2})])
 357 def test_del_one(db_memory_with_data, table, db_filter, data):
 358     result = db_memory_with_data.del_one(table, db_filter)
 359     assert result == {"deleted": 1}
 360     assert len(db_memory_with_data.db) == 1
 361     assert table in db_memory_with_data.db
 362     assert len(db_memory_with_data.db[table]) == 2
 363     assert data not in db_memory_with_data.db[table]
 364
 365
 366 @pytest.mark.parametrize("table, db_filter", [
 367     ("test", {}),
 368     ("test", {"_id": 1}),
 369     ("test", {"_id": 2}),
 370     ("test", {"data": 1}),
 371     ("test", {"data": 2}),
 372     ("test", {"_id": 1, "data": 1}),
 373     ("test", {"_id": 2, "data": 2}),
 374     ("test_table", {}),
 375     ("test_table", {"_id": 1}),
 376     ("test_table", {"_id": 2}),
 377     ("test_table", {"data": 1}),
 378     ("test_table", {"data": 2}),
 379     ("test_table", {"_id": 1, "data": 1}),
 380     ("test_table", {"_id": 2, "data": 2})])
 381 def test_del_one_with_empty_db_exception(db_memory, table, db_filter):
 382     with pytest.raises(DbException) as excinfo:
 383         db_memory.del_one(table, db_filter)
 384     assert str(excinfo.value) == (empty_exception_message() + del_one_exception_message(db_filter))
 385     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 386
 387
 388 @pytest.mark.parametrize("table, db_filter", [
 389     ("test", {}),
 390     ("test", {"_id": 1}),
 391     ("test", {"_id": 2}),
 392     ("test", {"data": 1}),
 393     ("test", {"data": 2}),
 394     ("test", {"_id": 1, "data": 1}),
 395     ("test", {"_id": 2, "data": 2}),
 396     ("test_table", {}),
 397     ("test_table", {"_id": 1}),
 398     ("test_table", {"_id": 2}),
 399     ("test_table", {"data": 1}),
 400     ("test_table", {"data": 2}),
 401     ("test_table", {"_id": 1, "data": 1}),
 402     ("test_table", {"_id": 2, "data": 2})])
 403 def test_del_one_with_empty_db_none(db_memory, table, db_filter):
 404     result = db_memory.del_one(table, db_filter, fail_on_empty=False)
 405     assert result is None
 406
 407
 408 @pytest.mark.parametrize("table, db_filter", [
 409     ("test", {"_id": 4}),
 410     ("test", {"_id": 5}),
 411     ("test", {"data": 4}),
 412     ("test", {"data": 5}),
 413     ("test", {"_id": 1, "data": 2}),
 414     ("test", {"_id": 2, "data": 3}),
 415     ("test_table", {}),
 416     ("test_table", {"_id": 1}),
 417     ("test_table", {"_id": 2}),
 418     ("test_table", {"data": 1}),
 419     ("test_table", {"data": 2}),
 420     ("test_table", {"_id": 1, "data": 1}),
 421     ("test_table", {"_id": 2, "data": 2})])
 422 def test_del_one_with_non_empty_db_exception(db_memory_with_data, table, db_filter):
 423     with pytest.raises(DbException) as excinfo:
 424         db_memory_with_data.del_one(table, db_filter)
 425     assert str(excinfo.value) == (empty_exception_message() + del_one_exception_message(db_filter))
 426     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 427
 428
 429 @pytest.mark.parametrize("table, db_filter", [
 430     ("test", {"_id": 4}),
 431     ("test", {"_id": 5}),
 432     ("test", {"data": 4}),
 433     ("test", {"data": 5}),
 434     ("test", {"_id": 1, "data": 2}),
 435     ("test", {"_id": 2, "data": 3}),
 436     ("test_table", {}),
 437     ("test_table", {"_id": 1}),
 438     ("test_table", {"_id": 2}),
 439     ("test_table", {"data": 1}),
 440     ("test_table", {"data": 2}),
 441     ("test_table", {"_id": 1, "data": 1}),
 442     ("test_table", {"_id": 2, "data": 2})])
 443 def test_del_one_with_non_empty_db_none(db_memory_with_data, table, db_filter):
 444     result = db_memory_with_data.del_one(table, db_filter, fail_on_empty=False)
 445     assert result is None
 446
 447
 448 @pytest.mark.parametrize("fail_on_empty", [
 449     (True),
 450     (False)])
 451 def test_del_one_generic_exception(db_memory_with_data, fail_on_empty):
 452     table = 'test'
 453     db_filter = {}
 454     db_memory_with_data._find = MagicMock(side_effect=Exception())
 455     with pytest.raises(DbException) as excinfo:
 456         db_memory_with_data.del_one(table, db_filter, fail_on_empty=fail_on_empty)
 457     assert str(excinfo.value) == empty_exception_message()
 458     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 459
 460
 461 @pytest.mark.parametrize("table, _id, indata", [
 462     ("test", 1, {"_id": 1, "data": 42}),
 463     ("test", 1, {"_id": 1, "data": 42, "kk": 34}),
 464     ("test", 1, {"_id": 1}),
 465     ("test", 2, {"_id": 2, "data": 42}),
 466     ("test", 2, {"_id": 2, "data": 42, "kk": 34}),
 467     ("test", 2, {"_id": 2}),
 468     ("test", 3, {"_id": 3, "data": 42}),
 469     ("test", 3, {"_id": 3, "data": 42, "kk": 34}),
 470     ("test", 3, {"_id": 3})])
 471 def test_replace(db_memory_with_data, table, _id, indata):
 472     result = db_memory_with_data.replace(table, _id, indata)
 473     assert result == {"updated": 1}
 474     assert len(db_memory_with_data.db) == 1
 475     assert table in db_memory_with_data.db
 476     assert len(db_memory_with_data.db[table]) == 3
 477     assert indata in db_memory_with_data.db[table]
 478
 479
 480 @pytest.mark.parametrize("table, _id, indata", [
 481     ("test", 1, {"_id": 1, "data": 42}),
 482     ("test", 2, {"_id": 2}),
 483     ("test", 3, {"_id": 3})])
 484 def test_replace_without_data_exception(db_memory, table, _id, indata):
 485     with pytest.raises(DbException) as excinfo:
 486         db_memory.replace(table, _id, indata, fail_on_empty=True)
 487     assert str(excinfo.value) == (replace_exception_message(_id))
 488     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 489
 490
 491 @pytest.mark.parametrize("table, _id, indata", [
 492     ("test", 1, {"_id": 1, "data": 42}),
 493     ("test", 2, {"_id": 2}),
 494     ("test", 3, {"_id": 3})])
 495 def test_replace_without_data_none(db_memory, table, _id, indata):
 496     result = db_memory.replace(table, _id, indata, fail_on_empty=False)
 497     assert result is None
 498
 499
 500 @pytest.mark.parametrize("table, _id, indata", [
 501     ("test", 11, {"_id": 11, "data": 42}),
 502     ("test", 12, {"_id": 12}),
 503     ("test", 33, {"_id": 33})])
 504 def test_replace_with_data_exception(db_memory_with_data, table, _id, indata):
 505     with pytest.raises(DbException) as excinfo:
 506         db_memory_with_data.replace(table, _id, indata, fail_on_empty=True)
 507     assert str(excinfo.value) == (replace_exception_message(_id))
 508     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 509
 510
 511 @pytest.mark.parametrize("table, _id, indata", [
 512     ("test", 11, {"_id": 11, "data": 42}),
 513     ("test", 12, {"_id": 12}),
 514     ("test", 33, {"_id": 33})])
 515 def test_replace_with_data_none(db_memory_with_data, table, _id, indata):
 516     result = db_memory_with_data.replace(table, _id, indata, fail_on_empty=False)
 517     assert result is None
 518
 519
 520 @pytest.mark.parametrize("fail_on_empty", [
 521     True,
 522     False])
 523 def test_replace_generic_exception(db_memory_with_data, fail_on_empty):
 524     table = 'test'
 525     _id = {}
 526     indata = {'_id': 1, 'data': 1}
 527     db_memory_with_data._find = MagicMock(side_effect=Exception())
 528     with pytest.raises(DbException) as excinfo:
 529         db_memory_with_data.replace(table, _id, indata, fail_on_empty=fail_on_empty)
 530     assert str(excinfo.value) == empty_exception_message()
 531     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 532
 533
 534 @pytest.mark.parametrize("table, id, data", [
 535     ("test", "1", {"data": 1}),
 536     ("test", "1", {"data": 2}),
 537     ("test", "2", {"data": 1}),
 538     ("test", "2", {"data": 2}),
 539     ("test_table", "1", {"data": 1}),
 540     ("test_table", "1", {"data": 2}),
 541     ("test_table", "2", {"data": 1}),
 542     ("test_table", "2", {"data": 2}),
 543     ("test", "1", {"data_1": 1, "data_2": 2}),
 544     ("test", "1", {"data_1": 2, "data_2": 1}),
 545     ("test", "2", {"data_1": 1, "data_2": 2}),
 546     ("test", "2", {"data_1": 2, "data_2": 1}),
 547     ("test_table", "1", {"data_1": 1, "data_2": 2}),
 548     ("test_table", "1", {"data_1": 2, "data_2": 1}),
 549     ("test_table", "2", {"data_1": 1, "data_2": 2}),
 550     ("test_table", "2", {"data_1": 2, "data_2": 1})])
 551 def test_create_with_empty_db_with_id(db_memory, table, id, data):
 552     data_to_insert = data
 553     data_to_insert['_id'] = id
 554     returned_id = db_memory.create(table, data_to_insert)
 555     assert returned_id == id
 556     assert len(db_memory.db) == 1
 557     assert table in db_memory.db
 558     assert len(db_memory.db[table]) == 1
 559     assert data_to_insert in db_memory.db[table]
 560
 561
 562 @pytest.mark.parametrize("table, id, data", [
 563     ("test", "4", {"data": 1}),
 564     ("test", "5", {"data": 2}),
 565     ("test", "4", {"data": 1}),
 566     ("test", "5", {"data": 2}),
 567     ("test_table", "4", {"data": 1}),
 568     ("test_table", "5", {"data": 2}),
 569     ("test_table", "4", {"data": 1}),
 570     ("test_table", "5", {"data": 2}),
 571     ("test", "4", {"data_1": 1, "data_2": 2}),
 572     ("test", "5", {"data_1": 2, "data_2": 1}),
 573     ("test", "4", {"data_1": 1, "data_2": 2}),
 574     ("test", "5", {"data_1": 2, "data_2": 1}),
 575     ("test_table", "4", {"data_1": 1, "data_2": 2}),
 576     ("test_table", "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 def test_create_with_non_empty_db_with_id(db_memory_with_data, table, id, data):
 580     data_to_insert = data
 581     data_to_insert['_id'] = id
 582     returned_id = db_memory_with_data.create(table, data_to_insert)
 583     assert returned_id == id
 584     assert len(db_memory_with_data.db) == (1 if table == 'test' else 2)
 585     assert table in db_memory_with_data.db
 586     assert len(db_memory_with_data.db[table]) == (4 if table == 'test' else 1)
 587     assert data_to_insert in db_memory_with_data.db[table]
 588
 589
 590 @pytest.mark.parametrize("table, data", [
 591     ("test", {"data": 1}),
 592     ("test", {"data": 2}),
 593     ("test", {"data": 1}),
 594     ("test", {"data": 2}),
 595     ("test_table", {"data": 1}),
 596     ("test_table", {"data": 2}),
 597     ("test_table", {"data": 1}),
 598     ("test_table", {"data": 2}),
 599     ("test", {"data_1": 1, "data_2": 2}),
 600     ("test", {"data_1": 2, "data_2": 1}),
 601     ("test", {"data_1": 1, "data_2": 2}),
 602     ("test", {"data_1": 2, "data_2": 1}),
 603     ("test_table", {"data_1": 1, "data_2": 2}),
 604     ("test_table", {"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 def test_create_with_empty_db_without_id(db_memory, table, data):
 608     returned_id = db_memory.create(table, data)
 609     assert len(db_memory.db) == 1
 610     assert table in db_memory.db
 611     assert len(db_memory.db[table]) == 1
 612     data_inserted = data
 613     data_inserted['_id'] = returned_id
 614     assert data_inserted in db_memory.db[table]
 615
 616
 617 @pytest.mark.parametrize("table, data", [
 618     ("test", {"data": 1}),
 619     ("test", {"data": 2}),
 620     ("test", {"data": 1}),
 621     ("test", {"data": 2}),
 622     ("test_table", {"data": 1}),
 623     ("test_table", {"data": 2}),
 624     ("test_table", {"data": 1}),
 625     ("test_table", {"data": 2}),
 626     ("test", {"data_1": 1, "data_2": 2}),
 627     ("test", {"data_1": 2, "data_2": 1}),
 628     ("test", {"data_1": 1, "data_2": 2}),
 629     ("test", {"data_1": 2, "data_2": 1}),
 630     ("test_table", {"data_1": 1, "data_2": 2}),
 631     ("test_table", {"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 def test_create_with_non_empty_db_without_id(db_memory_with_data, table, data):
 635     returned_id = db_memory_with_data.create(table, data)
 636     assert len(db_memory_with_data.db) == (1 if table == 'test' else 2)
 637     assert table in db_memory_with_data.db
 638     assert len(db_memory_with_data.db[table]) == (4 if table == 'test' else 1)
 639     data_inserted = data
 640     data_inserted['_id'] = returned_id
 641     assert data_inserted in db_memory_with_data.db[table]
 642
 643
 644 def test_create_with_exception(db_memory):
 645     table = "test"
 646     data = {"_id": 1, "data": 1}
 647     db_memory.db = MagicMock()
 648     db_memory.db.__contains__.side_effect = Exception()
 649     with pytest.raises(DbException) as excinfo:
 650         db_memory.create(table, data)
 651     assert str(excinfo.value) == empty_exception_message()
 652     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 653
 654
 655 @pytest.mark.parametrize("db_content, update_dict, expected, message", [
 656     ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
 657     ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
 658     ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
 659     ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
 660     ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
 661     ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
 662     ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
 663     ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
 664     ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
 665      {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
 666     ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
 667     ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
 668     ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key")])
 669 def test_set_one(db_memory, db_content, update_dict, expected, message):
 670     db_memory._find = Mock(return_value=((0, db_content), ))
 671     if expected is None:
 672         with pytest.raises(DbException) as excinfo:
 673             db_memory.set_one("table", {}, update_dict)
 674         assert (excinfo.value.http_code == http.HTTPStatus.NOT_FOUND), message
 675     else:
 676         db_memory.set_one("table", {}, update_dict)
 677         assert (db_content == expected), message
 678
 679
 680 class TestDbMemory(unittest.TestCase):
 681     # TODO to delete. This is cover with pytest test_set_one.
 682     def test_set_one(self):
 683         test_set = (
 684             # (database content, set-content, expected database content (None=fails), message)
 685             ({"a": {"none": None}}, {"a.b.num": "v"}, {"a": {"none": None, "b": {"num": "v"}}}, "create dict"),
 686             ({"a": {"none": None}}, {"a.none.num": "v"}, {"a": {"none": {"num": "v"}}}, "create dict over none"),
 687             ({"a": {"b": {"num": 4}}}, {"a.b.num": "v"}, {"a": {"b": {"num": "v"}}}, "replace_number"),
 688             ({"a": {"b": {"num": 4}}}, {"a.b.num.c.d": "v"}, None, "create dict over number should fail"),
 689             ({"a": {"b": {"num": 4}}}, {"a.b": "v"}, {"a": {"b": "v"}}, "replace dict with a string"),
 690             ({"a": {"b": {"num": 4}}}, {"a.b": None}, {"a": {"b": None}}, "replace dict with None"),
 691
 692             ({"a": [{"b": {"num": 4}}]}, {"a.b.num": "v"}, None, "create dict over list should fail"),
 693             ({"a": [{"b": {"num": 4}}]}, {"a.0.b.num": "v"}, {"a": [{"b": {"num": "v"}}]}, "set list"),
 694             ({"a": [{"b": {"num": 4}}]}, {"a.3.b.num": "v"},
 695              {"a": [{"b": {"num": 4}}, None, None, {"b": {"num": "v"}}]}, "expand list"),
 696             ({"a": [[4]]}, {"a.0.0": "v"}, {"a": [["v"]]}, "set nested list"),
 697             ({"a": [[4]]}, {"a.0.2": "v"}, {"a": [[4, None, "v"]]}, "expand nested list"),
 698             ({"a": [[4]]}, {"a.2.2": "v"}, {"a": [[4], None, {"2": "v"}]}, "expand list and add number key"),
 699         )
 700         db_men = DbMemory()
 701         db_men._find = Mock()
 702         for db_content, update_dict, expected, message in test_set:
 703             db_men._find.return_value = ((0, db_content), )
 704             if expected is None:
 705                 self.assertRaises(DbException, db_men.set_one, "table", {}, update_dict)
 706             else:
 707                 db_men.set_one("table", {}, update_dict)
 708                 self.assertEqual(db_content, expected, message)