ricsdl-package/ricsdl/backend/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 """The module provides fake implementation of Shared Data Layer (SDL) database backend interface."""
  23 import fnmatch
  24 from typing import (Dict, Set, List, Union)
  25 from ricsdl.configuration import _Configuration
  26 from .dbbackend_abc import DbBackendAbc
  27 from .dbbackend_abc import DbBackendLockAbc
  28
  29
  30 class FakeDictBackend(DbBackendAbc):
  31     """
  32     A class providing fake implementation of database backend of Shared Data Layer (SDL).
  33     This class does not provide working database solution, this class can be used in testing
  34     purposes only. Implementation does not provide shared database resource, SDL client sees
  35     only its local local 'fake' database, which is a simple Python dictionary. Also keys are
  36     stored in database under the same namespace.
  37
  38     Args:
  39         configuration (_Configuration): SDL configuration, containing credentials to connect to
  40                                         Redis database backend.
  41     """
  42     def __init__(self, configuration: _Configuration) -> None:
  43         super().__init__()
  44         self._db = {}
  45         self._configuration = configuration
  46
  47     def __str__(self):
  48         return str(
  49             {
  50                 "DB type": "FAKE DB",
  51             }
  52         )
  53
  54     def is_connected(self):
  55         return True
  56
  57     def close(self):
  58         pass
  59
  60     def set(self, ns: str, data_map: Dict[str, bytes]) -> None:
  61         self._db.update(data_map.copy())
  62
  63     def set_if(self, ns: str, key: str, old_data: bytes, new_data: bytes) -> bool:
  64         if key not in self._db:
  65             return False
  66         db_data = self._db[key]
  67         if db_data == old_data:
  68             self._db[key] = new_data
  69             return True
  70         return False
  71
  72     def set_if_not_exists(self, ns: str, key: str, data: bytes) -> bool:
  73         if key not in self._db:
  74             self._db[key] = data
  75             return True
  76         return False
  77
  78     def get(self, ns: str, keys: List[str]) -> Dict[str, bytes]:
  79         ret = {}
  80         for k in keys:
  81             if k in self._db:
  82                 ret[k] = self._db[k]
  83         return ret
  84
  85     def find_keys(self, ns: str, key_pattern: str) -> List[str]:
  86         ret = []
  87         for k in self._db:
  88             if fnmatch.fnmatch(k, key_pattern):
  89                 ret.append(k)
  90         return ret
  91
  92     def find_and_get(self, ns: str, key_pattern: str) -> Dict[str, bytes]:
  93         ret = {}
  94         for key, val in self._db.items():
  95             if fnmatch.fnmatch(key, key_pattern):
  96                 ret[key] = val
  97         return ret
  98
  99     def remove(self, ns: str, keys: List[str]) -> None:
 100         for key in keys:
 101             self._db.pop(key, None)
 102
 103     def remove_if(self, ns: str, key: str, data: bytes) -> bool:
 104         if key in self._db:
 105             db_data = self._db[key]
 106             if db_data == data:
 107                 self._db.pop(key)
 108                 return True
 109         return False
 110
 111     def add_member(self, ns: str, group: str, members: Set[bytes]) -> None:
 112         if group in self._db:
 113             self._db[group] = self._db[group] | members.copy()
 114         else:
 115             self._db[group] = members.copy()
 116
 117     def remove_member(self, ns: str, group: str, members: Set[bytes]) -> None:
 118         if group not in self._db:
 119             return
 120         for member in members:
 121             self._db[group].discard(member)
 122
 123     def remove_group(self, ns: str, group: str) -> None:
 124         self._db.pop(group, None)
 125
 126     def get_members(self, ns: str, group: str) -> Set[bytes]:
 127         return self._db.get(group, set())
 128
 129     def is_member(self, ns: str, group: str, member: bytes) -> bool:
 130         if group not in self._db:
 131             return False
 132         if member in self._db[group]:
 133             return True
 134         return False
 135
 136     def group_size(self, ns: str, group: str) -> int:
 137         if group not in self._db:
 138             return 0
 139         return len(self._db[group])
 140
 141
 142 class FakeDictBackendLock(DbBackendLockAbc):
 143     """
 144     A class providing fake implementation of database backend lock of Shared Data Layer (SDL).
 145     This class does not provide working database solution, this class can be used in testing
 146     purposes only. Implementation does not provide shared database resource, SDL client sees
 147     only its local local 'fake' database, which is a simple Python dictionary. Also keys are
 148     stored in database under the same namespace.
 149     Args:
 150         ns (str): Namespace under which this lock is targeted.
 151         name (str): Lock name, identifies the lock key in a Redis database backend.
 152         expiration (int, float): Lock expiration time after which the lock is removed if it hasn't
 153                                  been released earlier by a 'release' method.
 154         redis_backend (FakeBackend): Database backend object containing fake databese connection.
 155     """
 156
 157     def __init__(self, ns: str, name: str, expiration: Union[int, float],
 158                  redis_backend: FakeDictBackend) -> None:
 159         super().__init__(ns, name)
 160         self._locked = False
 161         self._ns = ns
 162         self._lock_name = name
 163         self._lock_expiration = expiration
 164         self.redis_backend = redis_backend
 165
 166     def __str__(self):
 167         return str(
 168             {
 169                 "lock DB type": "FAKE DB",
 170                 "lock namespace": self._ns,
 171                 "lock name": self._lock_name,
 172                 "lock status": self._lock_status_to_string()
 173             }
 174         )
 175
 176     def acquire(self, retry_interval: Union[int, float] = 0.1,
 177                 retry_timeout: Union[int, float] = 10) -> bool:
 178         if self._locked:
 179             return False
 180         self._locked = True
 181         return self._locked
 182
 183     def release(self) -> None:
 184         self._locked = False
 185
 186     def refresh(self) -> None:
 187         pass
 188
 189     def get_validity_time(self) -> Union[int, float]:
 190         return self._lock_expiration
 191
 192     def _lock_status_to_string(self) -> str:
 193         if self._locked:
 194             return 'locked'
 195         return 'unlocked'