ricsdl-package/tests/backend/test_fake_dict_db.py

   1 # Copyright (c) 2019 AT&T Intellectual Property.
   2 # Copyright (c) 2018-2019 Nokia.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain 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,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 #
  17 # This source code is part of the near-RT RIC (RAN Intelligent Controller)
  18 # platform project (RICP).
  19 #
  20
  21
  22 from unittest.mock import Mock
  23 import pytest
  24 import ricsdl.backend
  25 from ricsdl.configuration import _Configuration
  26 from ricsdl.configuration import DbBackendType
  27
  28
  29 @pytest.fixture()
  30 def fake_dict_backend_fixture(request):
  31     request.cls.ns = 'some-ns'
  32     request.cls.dm = {'abc': b'1', 'bcd': b'2'}
  33     request.cls.new_dm = {'abc': b'3', 'bcd': b'2'}
  34     request.cls.dm2 = {'cdf': b'4'}
  35     request.cls.remove_dm = {'bcd': b'2'}
  36     request.cls.key = 'abc'
  37     request.cls.keys = ['abc', 'bcd']
  38     request.cls.key2 = ['cdf']
  39     request.cls.old_data = b'1'
  40     request.cls.new_data = b'3'
  41     request.cls.keypattern = r'*bc*'
  42     request.cls.group = 'some-group'
  43     request.cls.groupmember = b'm1'
  44     request.cls.groupmembers = set([b'm1', b'm2'])
  45     request.cls.new_groupmembers = set(b'm3')
  46     request.cls.all_groupmembers = request.cls.groupmembers | request.cls.new_groupmembers
  47
  48     request.cls.configuration = Mock()
  49     mock_conf_params = _Configuration.Params(db_host=None,
  50                                              db_port=None,
  51                                              db_sentinel_port=None,
  52                                              db_sentinel_master_name=None,
  53                                              db_type=DbBackendType.FAKE_DICT)
  54     request.cls.configuration.get_params.return_value = mock_conf_params
  55     request.cls.db = ricsdl.backend.get_backend_instance(request.cls.configuration)
  56
  57
  58 @pytest.mark.usefixtures('fake_dict_backend_fixture')
  59 class TestFakeDictBackend:
  60     def test_is_connected_function_success(self):
  61         ret = self.db.is_connected()
  62         assert ret is True
  63
  64     def test_set_function_success(self):
  65         self.db.set(self.ns, self.dm)
  66         self.db.set(self.ns, self.dm2)
  67         ret = self.db.get(self.ns, self.keys)
  68         assert ret == self.dm
  69         ret = self.db.get(self.ns, self.key2)
  70         assert ret == self.dm2
  71
  72     def test_set_if_function_success(self):
  73         self.db.set(self.ns, self.dm)
  74         ret = self.db.set_if(self.ns, self.key, self.old_data, self.new_data)
  75         assert ret is True
  76         ret = self.db.get(self.ns, self.keys)
  77         assert ret == self.new_dm
  78
  79     def test_set_if_function_returns_false_if_existing_key_value_not_expected(self):
  80         self.db.set_if(self.ns, self.key, self.old_data, self.new_data)
  81         self.db.set(self.ns, self.new_dm)
  82         ret = self.db.set_if(self.ns, self.key, self.old_data, self.new_data)
  83         assert ret is False
  84
  85     def test_set_if_not_exists_function_success(self):
  86         ret = self.db.set_if_not_exists(self.ns, self.key, self.new_data)
  87         assert ret is True
  88         ret = self.db.get(self.ns, self.keys)
  89         assert ret == {self.key: self.new_data}
  90
  91     def test_set_if_not_exists_function_returns_false_if_key_already_exists(self):
  92         self.db.set(self.ns, self.dm)
  93         ret = self.db.set_if_not_exists(self.ns, self.key, self.new_data)
  94         assert ret is False
  95
  96     def test_find_keys_function_success(self):
  97         self.db.set(self.ns, self.dm)
  98         ret = self.db.find_keys(self.ns, self.keypattern)
  99         assert ret == self.keys
 100
 101     def test_find_keys_function_returns_empty_list_when_no_matching_keys_found(self):
 102         ret = self.db.find_keys(self.ns, self.keypattern)
 103         assert ret == []
 104
 105     def test_find_and_get_function_success(self):
 106         self.db.set(self.ns, self.dm)
 107         ret = self.db.find_and_get(self.ns, self.keypattern)
 108         assert ret == self.dm
 109
 110     def test_find_and_get_function_returns_empty_dict_when_no_matching_keys_exist(self):
 111         ret = self.db.find_and_get(self.ns, self.keypattern)
 112         assert ret == dict()
 113
 114     def test_remove_function_success(self):
 115         self.db.set(self.ns, self.dm)
 116         self.db.remove(self.ns, self.keys)
 117         ret = self.db.get(self.ns, self.keys)
 118         assert ret == dict()
 119
 120     def test_remove_if_function_success(self):
 121         self.db.set(self.ns, self.dm)
 122         ret = self.db.remove_if(self.ns, self.key, self.old_data)
 123         assert ret is True
 124         ret = self.db.get(self.ns, self.keys)
 125         assert ret == self.remove_dm
 126
 127     def test_remove_if_function_returns_false_if_data_does_not_match(self):
 128         ret = self.db.remove_if(self.ns, self.key, self.old_data)
 129         assert ret is False
 130         self.db.set(self.ns, self.dm)
 131         ret = self.db.remove_if(self.ns, self.key, self.new_data)
 132         assert ret is False
 133
 134     def test_add_member_function_success(self):
 135         self.db.add_member(self.ns, self.group, self.groupmembers)
 136         ret = self.db.get_members(self.ns, self.group)
 137         assert ret == self.groupmembers
 138
 139         self.db.add_member(self.ns, self.group, self.new_groupmembers)
 140         ret = self.db.get_members(self.ns, self.group)
 141         assert ret == self.all_groupmembers
 142
 143     def test_remove_member_function_success(self):
 144         self.db.remove_member(self.ns, self.group, self.groupmembers)
 145         self.db.add_member(self.ns, self.group, self.groupmembers)
 146         self.db.remove_member(self.ns, self.group, self.groupmembers)
 147         ret = self.db.get_members(self.ns, self.group)
 148         assert ret == set()
 149
 150     def test_remove_group_function_success(self):
 151         self.db.remove_group(self.ns, self.group)
 152         ret = self.db.get_members(self.ns, self.group)
 153         assert ret == set()
 154
 155     def test_is_member_function_success(self):
 156         ret = self.db.is_member(self.ns, self.group, b'not member')
 157         assert ret is False
 158         self.db.add_member(self.ns, self.group, self.groupmembers)
 159         ret = self.db.is_member(self.ns, self.group, self.groupmember)
 160         assert ret is True
 161         ret = self.db.is_member(self.ns, self.group, b'not member')
 162         assert ret is False
 163
 164     def test_group_size_function_success(self):
 165         ret = self.db.group_size(self.ns, self.group)
 166         assert ret == 0
 167         self.db.add_member(self.ns, self.group, self.groupmembers)
 168         ret = self.db.group_size(self.ns, self.group)
 169         assert ret == len(self.groupmembers)
 170
 171     def test_fake_dict_backend_object_string_representation(self):
 172         assert str(self.db) == str({'DB type': 'FAKE DB'})
 173
 174 @pytest.fixture()
 175 def fake_dict_backend_lock_fixture(request):
 176     request.cls.ns = 'some-ns'
 177     request.cls.lockname = 'some-lock-name'
 178     request.cls.expiration = 10
 179     request.cls.retry_interval = 0.1
 180     request.cls.retry_timeout = 1
 181
 182     request.cls.configuration = Mock()
 183     mock_conf_params = _Configuration.Params(db_host=None,
 184                                              db_port=None,
 185                                              db_sentinel_port=None,
 186                                              db_sentinel_master_name=None,
 187                                              db_type=DbBackendType.FAKE_DICT)
 188     request.cls.configuration.get_params.return_value = mock_conf_params
 189     request.cls.lock = ricsdl.backend.get_backend_lock_instance(request.cls.configuration,
 190                                                                 request.cls.ns,
 191                                                                 request.cls.lockname,
 192                                                                 request.cls.expiration,
 193                                                                 Mock())
 194
 195
 196 @pytest.mark.usefixtures('fake_dict_backend_lock_fixture')
 197 class TestFakeDictBackendLock:
 198     def test_acquire_function_success(self):
 199         ret = self.lock.acquire(self.retry_interval, self.retry_timeout)
 200         assert ret is True
 201
 202     def test_acquire_function_returns_false_if_lock_is_not_acquired(self):
 203         self.lock.acquire(self.retry_interval, self.retry_timeout)
 204         ret = self.lock.acquire(self.retry_interval, self.retry_timeout)
 205         assert ret is False
 206
 207     def test_release_function_success(self):
 208         self.lock.acquire(self.retry_interval, self.retry_timeout)
 209         ret = self.lock.acquire(self.retry_interval, self.retry_timeout)
 210         assert ret is False
 211         self.lock.release()
 212         ret = self.lock.acquire(self.retry_interval, self.retry_timeout)
 213         assert ret is True
 214
 215     def test_get_validity_time_function_success(self):
 216         ret = self.lock.get_validity_time()
 217         assert ret == self.expiration
 218
 219     def test_fake_dict_backend_lock_object_string_representation(self):
 220         expected_lock_info = {'lock DB type': 'FAKE DB',
 221                               'lock namespace': 'some-ns',
 222                               'lock name': 'some-lock-name',
 223                               'lock status': 'unlocked'}
 224         assert str(self.lock) == str(expected_lock_info)