Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_do_PATCH_config(self, mock_dcs):
config = {'postgresql': {'use_slots': False, 'use_pg_rewind': True, 'parameters': {'wal_level': 'logical'}}}
mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, json.dumps(config))
request = 'PATCH /config HTTP/1.0' + self._authorization
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request))
request += '\nContent-Length: '
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '34\n\n{"postgresql":{"use_slots":false}}'))
config['ttl'] = 5
config['postgresql'].update({'use_slots': {'foo': True}, "parameters": None})
config = json.dumps(config)
request += str(len(config)) + '\n\n' + config
MockRestApiServer(RestApiHandler, request)
mock_dcs.set_config_value.return_value = False
MockRestApiServer(RestApiHandler, request)
def test_do_PUT_config(self, mock_dcs):
mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, '{}')
request = 'PUT /config HTTP/1.0' + self._authorization + '\nContent-Length: '
self.assertIsNotNone(MockRestApiServer(RestApiHandler, request + '2\n\n{}'))
config = '{"foo": "bar"}'
request += str(len(config)) + '\n\n' + config
MockRestApiServer(RestApiHandler, request)
mock_dcs.set_config_value.return_value = False
MockRestApiServer(RestApiHandler, request)
mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, config)
MockRestApiServer(RestApiHandler, request)
def _load_cluster(self):
prefix = self.client_path('')
response = self._sync_obj.get(prefix, recursive=True)
if not response:
return Cluster(None, None, None, None, [], None, None, None)
nodes = {os.path.relpath(key, prefix).replace('\\', '/'): value for key, value in response.items()}
# get initialize flag
initialize = nodes.get(self._INITIALIZE)
initialize = initialize and initialize['value']
# get global dynamic configuration
config = nodes.get(self._CONFIG)
config = config and ClusterConfig.from_node(config['index'], config['value'])
# get timeline history
history = nodes.get(self._HISTORY)
history = history and TimelineHistory.from_node(history['index'], history['value'])
# get last leader operation
last_leader_operation = nodes.get(self._LEADER_OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value'])
# get list of members
members = [self.member(k, n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
# get leader
leader = nodes.get(self._LEADER)
if leader:
member = Member(-1, leader['value'], None, {})
def _load_cluster(self):
cluster = None
try:
result = self.retry(self._client.read, self.client_path(''), recursive=True)
nodes = {node.key[len(result.key):].lstrip('/'): node for node in result.leaves}
# get initialize flag
initialize = nodes.get(self._INITIALIZE)
initialize = initialize and initialize.value
# get global dynamic configuration
config = nodes.get(self._CONFIG)
config = config and ClusterConfig.from_node(config.modifiedIndex, config.value)
# get timeline history
history = nodes.get(self._HISTORY)
history = history and TimelineHistory.from_node(history.modifiedIndex, history.value)
# get last leader operation
last_leader_operation = nodes.get(self._LEADER_OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation.value)
# get list of members
members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
# get leader
leader = nodes.get(self._LEADER)
if leader:
member = Member(-1, leader.value, None, {})
with self._condition:
self._wait_caches()
members = [self.member(pod) for pod in self._pods.copy().values()]
nodes = self._kinds.copy()
config = nodes.get(self.config_path)
metadata = config and config.metadata
self._config_resource_version = metadata.resource_version if metadata else None
annotations = metadata and metadata.annotations or {}
# get initialize flag
initialize = annotations.get(self._INITIALIZE)
# get global dynamic configuration
config = ClusterConfig.from_node(metadata and metadata.resource_version,
annotations.get(self._CONFIG) or '{}',
metadata.resource_version if self._CONFIG in annotations else 0)
# get timeline history
history = TimelineHistory.from_node(metadata and metadata.resource_version,
annotations.get(self._HISTORY) or '[]')
leader = nodes.get(self.leader_path)
metadata = leader and leader.metadata
self._leader_resource_version = metadata.resource_version if metadata else None
self._leader_observed_subsets = leader.subsets if self.__subsets and leader and leader.subsets else []
annotations = metadata and metadata.annotations or {}
# get last leader operation
last_leader_operation = annotations.get(self._OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation)
def _inner_load_cluster(self):
self._fetch_cluster = False
self.event.clear()
nodes = set(self.get_children(self.client_path(''), self.cluster_watcher))
if not nodes:
self._fetch_cluster = True
# get initialize flag
initialize = (self.get_node(self.initialize_path) or [None])[0] if self._INITIALIZE in nodes else None
# get global dynamic configuration
config = self.get_node(self.config_path, watch=self.cluster_watcher) if self._CONFIG in nodes else None
config = config and ClusterConfig.from_node(config[1].version, config[0], config[1].mzxid)
# get timeline history
history = self.get_node(self.history_path, watch=self.cluster_watcher) if self._HISTORY in nodes else None
history = history and TimelineHistory.from_node(history[1].mzxid, history[0])
# get last leader operation
last_leader_operation = self._OPTIME in nodes and self._fetch_cluster and self.get_node(self.leader_optime_path)
last_leader_operation = last_leader_operation and int(last_leader_operation[0]) or 0
# get synchronization state
sync = self.get_node(self.sync_path, watch=self.cluster_watcher) if self._SYNC in nodes else None
sync = SyncState.from_node(sync and sync[1].version, sync and sync[0])
# get list of members
sync_standby = sync.leader == self._name and sync.sync_standby or None
members = self.load_members(sync_standby) if self._MEMBERS[:-1] in nodes else []
path = self.client_path('')
result = self.retry(self._client.prefix, path)
nodes = {}
for node in result.get('kvs', []):
node['key'] = base64_decode(node['key'])
node['value'] = base64_decode(node.get('value', ''))
node['lease'] = node.get('lease')
nodes[node['key'][len(path):].lstrip('/')] = node
# get initialize flag
initialize = nodes.get(self._INITIALIZE)
initialize = initialize and initialize['value']
# get global dynamic configuration
config = nodes.get(self._CONFIG)
config = config and ClusterConfig.from_node(config['mod_revision'], config['value'])
# get timeline history
history = nodes.get(self._HISTORY)
history = history and TimelineHistory.from_node(history['mod_revision'], history['value'])
# get last leader operation
last_leader_operation = nodes.get(self._LEADER_OPTIME)
last_leader_operation = 0 if last_leader_operation is None else int(last_leader_operation['value'])
# get list of members
members = [self.member(n) for k, n in nodes.items() if k.startswith(self._MEMBERS) and k.count('/') == 1]
# get leader
leader = nodes.get(self._LEADER)
if not self._ctl and leader and leader['value'] == self._name and self._lease != leader.get('lease'):
logger.warning('I am the leader but not owner of the lease')