From 3fe2a7868a0b0a5f7ed72f196878159b914db0bc Mon Sep 17 00:00:00 2001 From: Polina Bungina <27892524+hughcapet@users.noreply.github.com> Date: Mon, 3 Apr 2023 09:52:22 +0200 Subject: [PATCH] Ignore D401 in flake8-docstrings (#2627) * Ignore D401 in flake8-docstrings * Fix newly reported flake8 issues, ignore the old W503 rule * rely on concatenation of adjecent strings * Format behave scripts * Reformat ha.py according to new rules Co-authored-by: Alexander Kukushkin --- features/environment.py | 38 +++---- features/steps/basic_replication.py | 2 +- features/steps/citus.py | 6 +- features/steps/standby_cluster.py | 4 +- patroni/api.py | 6 +- patroni/config.py | 4 +- patroni/ctl.py | 10 +- patroni/dcs/__init__.py | 16 +-- patroni/dcs/consul.py | 10 +- patroni/dcs/etcd.py | 12 +- patroni/dcs/etcd3.py | 10 +- patroni/dcs/kubernetes.py | 19 ++-- patroni/dcs/raft.py | 6 +- patroni/dcs/zookeeper.py | 21 ++-- patroni/ha.py | 34 +++--- patroni/postgresql/__init__.py | 36 +++--- patroni/postgresql/cancellable.py | 2 +- patroni/postgresql/citus.py | 2 +- patroni/postgresql/config.py | 18 +-- patroni/postgresql/rewind.py | 15 +-- patroni/postgresql/slots.py | 24 ++-- patroni/postgresql/sync.py | 4 +- patroni/validator.py | 166 ++++++++++++++-------------- tests/test_api.py | 4 +- tests/test_bootstrap.py | 2 +- tests/test_consul.py | 10 +- tests/test_ctl.py | 2 +- tests/test_etcd.py | 8 +- tests/test_etcd3.py | 2 +- tests/test_kubernetes.py | 4 +- tests/test_postgresql.py | 32 +++--- tests/test_raft.py | 2 +- tests/test_rewind.py | 12 +- tests/test_validator.py | 12 +- tests/test_watchdog.py | 2 +- tests/test_zookeeper.py | 2 +- tox.ini | 1 + 37 files changed, 282 insertions(+), 278 deletions(-) diff --git a/features/environment.py b/features/environment.py index 2e03b310..bb677026 100644 --- a/features/environment.py +++ b/features/environment.py @@ -191,7 +191,7 @@ class PatroniController(AbstractController): config['raft'] = {'data_dir': self._output_dir, 'self_addr': 'localhost:' + os.environ['RAFT_PORT']} host = config['restapi']['listen'].rsplit(':', 1)[0] - config['restapi']['listen'] = config['restapi']['connect_address'] = '{0}:{1}'.format(host, 8008+int(name[-1])) + config['restapi']['listen'] = config['restapi']['connect_address'] = '{}:{}'.format(host, 8008 + int(name[-1])) host = config['postgresql']['listen'].rsplit(':', 1)[0] config['postgresql']['listen'] = config['postgresql']['connect_address'] = '{0}:{1}'.format(host, self.__PORT) @@ -251,9 +251,9 @@ class PatroniController(AbstractController): 'parameters': { 'wal_keep_segments': 100, 'archive_mode': 'on', - 'archive_command': (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT + - ' --mode archive ' + - '--dirname {} --filename %f --pathname %p').format( + 'archive_command': (PatroniPoolController.ARCHIVE_RESTORE_SCRIPT + + ' --mode archive ' + + '--dirname {} --filename %f --pathname %p').format( os.path.join(self._work_directory, 'data', 'wal_archive')) } } @@ -801,7 +801,7 @@ class PatroniPoolController(object): raise Exception # this one should never happen because the previous line will always raise and exception except Exception as e: self._context.postgres_supports_ssl = isinstance(e, subprocess.CalledProcessError)\ - and 'SSL is not supported by this build' not in e.output.decode() + and 'SSL is not supported by this build' not in e.output.decode() @property def patroni_path(self): @@ -853,8 +853,8 @@ class PatroniPoolController(object): 'bootstrap': { 'method': 'pg_basebackup', 'pg_basebackup': { - 'command': " ".join(self.BACKUP_SCRIPT + - ['--walmethod=stream', '--dbname="{0}"'.format(f.backup_source)]) + 'command': " ".join(self.BACKUP_SCRIPT + + ['--walmethod=stream', '--dbname="{0}"'.format(f.backup_source)]) }, 'dcs': { 'postgresql': { @@ -867,9 +867,9 @@ class PatroniPoolController(object): 'postgresql': { 'parameters': { 'archive_mode': 'on', - 'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive ' + - '--dirname {} --filename %f --pathname %p').format( - os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/')) + 'archive_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode archive ' + + '--dirname {} --filename %f --pathname %p') + .format(os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/')) }, 'authentication': { 'superuser': {'password': 'zalando1'}, @@ -885,13 +885,13 @@ class PatroniPoolController(object): 'bootstrap': { 'method': 'backup_restore', 'backup_restore': { - 'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' + - os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')), + 'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' + + os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')), 'recovery_conf': { 'recovery_target_action': 'promote', 'recovery_target_timeline': 'latest', - 'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' + - '--dirname {} --filename %f --pathname %p').format( + 'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' + + '--dirname {} --filename %f --pathname %p').format( os.path.join(self.patroni_path, 'data', 'wal_archive_clone').replace('\\', '/')) } } @@ -910,14 +910,14 @@ class PatroniPoolController(object): 'scope': cluster_name, 'postgresql': { 'recovery_conf': { - 'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' + - '--dirname {} --filename %f --pathname %p').format( - os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/')) + 'restore_command': (self.ARCHIVE_RESTORE_SCRIPT + ' --mode restore ' + + '--dirname {} --filename %f --pathname %p') + .format(os.path.join(self.patroni_path, 'data', 'wal_archive').replace('\\', '/')) }, 'create_replica_methods': ['no_leader_bootstrap'], 'no_leader_bootstrap': { - 'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' + - os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')), + 'command': (self.BACKUP_RESTORE_SCRIPT + ' --sourcedir=' + + os.path.join(self.patroni_path, 'data', 'basebackup').replace('\\', '/')), 'no_leader': '1' } } diff --git a/features/steps/basic_replication.py b/features/steps/basic_replication.py index 664288ac..9b51a859 100644 --- a/features/steps/basic_replication.py +++ b/features/steps/basic_replication.py @@ -38,7 +38,7 @@ def toggle_wal_replay(context, action, pg_name): # pause or resume the wal replay process try: version = context.pctl.query(pg_name, "SHOW server_version_num").fetchone()[0] - wal_name = 'xlog' if int(version)/10000 < 10 else 'wal' + wal_name = 'xlog' if int(version) / 10000 < 10 else 'wal' context.pctl.query(pg_name, "SELECT pg_{0}_replay_{1}()".format(wal_name, action)) except pg.Error as e: assert False, "Error during {0} wal recovery on {1}: {2}".format(action, pg_name, e) diff --git a/features/steps/citus.py b/features/steps/citus.py index 01f5ef2f..d645a504 100644 --- a/features/steps/citus.py +++ b/features/steps/citus.py @@ -35,7 +35,7 @@ def check_group_member(context, name, group, key, value, time_limit): except Exception: pass time.sleep(1) - assert False, ("{0} in a group {1} does not have {2}={3} (found {4}) in dcs" + + assert False, ("{0} in a group {1} does not have {2}={3} (found {4}) in dcs" " after {5} seconds").format(name, group, key, value, response, time_limit) @@ -93,7 +93,7 @@ def thread_is_alive(context): @step("I stop a thread") def stop_insert_thread(context): context.thread_stop_event.set() - context.thread.join(1*context.timeout_multiplier) + context.thread.join(1 * context.timeout_multiplier) assert not context.thread.is_alive(), "Thread is still alive" @@ -114,4 +114,4 @@ def check_transaction(context, name): @step("a transaction finishes in {timeout:d} seconds") def check_transaction_timeout(context, timeout): assert (datetime.now(tzutc) - context.xact_start).seconds > timeout,\ - "a transaction finished earlier than in {0} seconds".format(timeout) + "a transaction finished earlier than in {0} seconds".format(timeout) diff --git a/features/steps/standby_cluster.py b/features/steps/standby_cluster.py index 19256c26..17d635b8 100644 --- a/features/steps/standby_cluster.py +++ b/features/steps/standby_cluster.py @@ -16,8 +16,8 @@ def start_patroni(context, name, cluster_name): "postgresql": { "callbacks": callbacks(context, name), "backup_restore": { - "command": (context.pctl.PYTHON + " features/backup_restore.py --sourcedir=" + - os.path.join(context.pctl.patroni_path, 'data', 'basebackup').replace('\\', '/'))} + "command": (context.pctl.PYTHON + " features/backup_restore.py --sourcedir=" + + os.path.join(context.pctl.patroni_path, 'data', 'basebackup').replace('\\', '/'))} } }) diff --git a/patroni/api.py b/patroni/api.py index c614dcb9..e5b8e6fb 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -338,7 +338,7 @@ class RestApiHandler(BaseHTTPRequestHandler): metrics.append("# TYPE patroni_is_paused gauge") metrics.append("patroni_is_paused{0} {1}".format(scope_label, int(postgres.get('pause', 0)))) - self._write_response(200, '\n'.join(metrics)+'\n', content_type='text/plain') + self._write_response(200, '\n'.join(metrics) + '\n', content_type='text/plain') def _read_json_content(self, body_is_optional=False): if 'content-length' not in self.headers: @@ -537,7 +537,7 @@ class RestApiHandler(BaseHTTPRequestHandler): def poll_failover_result(self, leader, candidate, action): timeout = max(10, self.server.patroni.dcs.loop_wait) - for _ in range(0, timeout*2): + for _ in range(0, timeout * 2): time.sleep(1) try: cluster = self.server.patroni.dcs.get_cluster() @@ -861,7 +861,7 @@ class RestApiServer(ThreadingMixIn, HTTPServer, Thread): HTTPServer.__init__(self, info[0][-1][:2], RestApiHandler) except socket.error: logger.error( - "Couldn't start a service on '%s:%s', please check your `restapi.listen` configuration", host, port) + "Couldn't start a service on '%s:%s', please check your `restapi.listen` configuration", host, port) raise def __initialize(self, listen, ssl_options): diff --git a/patroni/config.py b/patroni/config.py index 2ec129ad..1fca41b9 100644 --- a/patroni/config.py +++ b/patroni/config.py @@ -321,8 +321,8 @@ class Config(object): @staticmethod def _process_postgresql_parameters(parameters, is_local=False): return {name: value for name, value in (parameters or {}).items() - if name not in ConfigHandler.CMDLINE_OPTIONS or - not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)} + if name not in ConfigHandler.CMDLINE_OPTIONS + or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)} def _safe_copy_dynamic_configuration(self, dynamic_configuration): config = deepcopy(self.__DEFAULT_CONFIG) diff --git a/patroni/ctl.py b/patroni/ctl.py index 526ac560..fb6aa2b4 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -249,9 +249,9 @@ def get_all_members(obj, cluster, group, role='leader'): role = {'primary': 'master', 'standby-leader': 'standby_leader'}.get(role, role) for cluster in clusters.values(): if cluster.leader is not None and cluster.leader.name and\ - (role == 'leader' or - cluster.leader.data.get('role') != 'master' and role == 'standby_leader' or - cluster.leader.data.get('role') != 'standby_leader' and role == 'master'): + (role == 'leader' + or cluster.leader.data.get('role') != 'master' and role == 'standby_leader' + or cluster.leader.data.get('role') != 'standby_leader' and role == 'master'): yield cluster.leader.member return @@ -879,7 +879,7 @@ def output_members(obj, cluster, name, extended=False, fmt='pretty', group=None) member.update(cluster=name, member=member['name'], group=g, host=member.get('host', ''), tl=member.get('timeline', ''), role=member['role'].replace('_', ' ').title(), - lag_in_mb=round(lag/1024/1024) if isinstance(lag, int) else lag, + lag_in_mb=round(lag / 1024 / 1024) if isinstance(lag, int) else lag, pending_restart='*' if member.get('pending_restart') else '') if append_port and member['host'] and member.get('port'): @@ -1254,7 +1254,7 @@ def edit_config(obj, cluster_name, group, force, quiet, kvpairs, pgkvpairs, appl after_editing, changed_data = apply_yaml_file(changed_data, apply_filename) if kvpairs or pgkvpairs: - all_pairs = list(kvpairs) + ['postgresql.parameters.'+v.lstrip() for v in pgkvpairs] + all_pairs = list(kvpairs) + ['postgresql.parameters.' + v.lstrip() for v in pgkvpairs] after_editing, changed_data = apply_config_changes(before_editing, changed_data, all_pairs) # If no changes were specified on the command line invoke editor diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index aa448efd..f5a5b89c 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -353,8 +353,8 @@ class ClusterConfig(namedtuple('ClusterConfig', 'index,data,modify_index')): @property def permanent_slots(self): return isinstance(self.data, dict) and ( - self.data.get('permanent_replication_slots') or - self.data.get('permanent_slots') or self.data.get('slots') + self.data.get('permanent_replication_slots') + or self.data.get('permanent_slots') or self.data.get('slots') ) or {} @property @@ -546,15 +546,15 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,' # primary), or if replicatefrom destination member happens to be the current primary use_slots = self.use_slots if role in ('master', 'primary', 'standby_leader'): - slot_members = [m.name for m in self.members if use_slots and m.name != my_name and - (m.replicatefrom is None or m.replicatefrom == my_name or - not self.has_member(m.replicatefrom))] + slot_members = [m.name for m in self.members if use_slots and m.name != my_name + and (m.replicatefrom is None or m.replicatefrom == my_name + or not self.has_member(m.replicatefrom))] permanent_slots = self.__permanent_slots if use_slots and \ role in ('master', 'primary') else self.__permanent_physical_slots else: # only manage slots for replicas that replicate from this one, except for the leader among them - slot_members = [m.name for m in self.members if use_slots and - m.replicatefrom == my_name and m.name != self.leader_name] + slot_members = [m.name for m in self.members if use_slots + and m.replicatefrom == my_name and m.name != self.leader_name] permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {} slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members} @@ -590,7 +590,7 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,' if major_version < 110000: disabled_permanent_logical_slots.append(name) elif name in slots: - logger.error("Permanent logical replication slot {'%s': %s} is conflicting with" + + logger.error("Permanent logical replication slot {'%s': %s} is conflicting with" " physical replication slot for cluster member", name, value) else: slots[name] = value diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 5c988833..fbf4c63f 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -15,7 +15,7 @@ from urllib3.exceptions import HTTPError from urllib.parse import urlencode, urlparse, quote from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re from ..exceptions import DCSError from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT @@ -63,7 +63,7 @@ class HTTPClient(object): self._ttl = None def set_read_timeout(self, timeout): - self._read_timeout = timeout/3.0 + self._read_timeout = timeout / 3.0 @property def ttl(self): @@ -113,7 +113,7 @@ class HTTPClient(object): # supplied maximum wait time to spread out the wake up time of any concurrent requests. This adds # up to wait / 16 additional time to the maximum duration. Since our goal is actually getting a # response rather read timeout we will add to the timeout a slightly bigger value. - kwargs['timeout'] = timeout + max(timeout/15.0, 1) + kwargs['timeout'] = timeout + max(timeout / 15.0, 1) else: kwargs['timeout'] = self._read_timeout kwargs['headers'] = (headers or {}).copy() @@ -267,7 +267,7 @@ class Consul(AbstractDCS): self._register_service = should_register_service def set_ttl(self, ttl): - if self._client.http.set_ttl(ttl/2.0): # Consul multiplies the TTL by 2x + if self._client.http.set_ttl(ttl / 2.0): # Consul multiplies the TTL by 2x self._session = None self.__do_not_watch = True @@ -282,7 +282,7 @@ class Consul(AbstractDCS): def adjust_ttl(self): try: settings = self._client.agent.self() - min_ttl = (settings['Config']['SessionTTLMin'] or 10000000000)/1000000000.0 + min_ttl = (settings['Config']['SessionTTLMin'] or 10000000000) / 1000000000.0 logger.warning('Changing Session TTL from %s to %s', self._client.http.ttl, min_ttl) self._client.http.set_ttl(min_ttl) except Exception: diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 154a7de0..c736c1c6 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -22,7 +22,7 @@ from urllib3 import Timeout from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re from ..exceptions import DCSError from ..request import get as requests_get from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT @@ -148,7 +148,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client): kwargs.update(retries=0, timeout=timeout) else: _, per_node_timeout, per_node_retries = self._calculate_timeouts(etcd_nodes) - connect_timeout = max(1, per_node_timeout/2) + connect_timeout = max(1, per_node_timeout / 2) kwargs.update(timeout=Timeout(connect=connect_timeout, total=per_node_timeout), retries=per_node_retries) return kwargs @@ -233,8 +233,8 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client): # whether the key didn't received an update or there is a network problem. elif i + 1 < len(machines_cache): self.set_base_uri(machines_cache[i + 1]) - if (isinstance(fields, dict) and fields.get("wait") == "true" and - isinstance(e, (ReadTimeoutError, ProtocolError))): + if (isinstance(fields, dict) and fields.get("wait") == "true" + and isinstance(e, (ReadTimeoutError, ProtocolError))): logger.debug("Watch timed out.") raise etcd.EtcdWatchTimedOut("Watch timed out: {0}".format(e), cause=e) logger.error("Request to server %s failed: %r", base_uri, e) @@ -287,7 +287,7 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client): retry.sleep_func(sleeptime) retry.update_delay() # We still have some time left. Partially reduce `machines_cache` and retry request - kwargs.update(timeout=Timeout(connect=max(1, timeout/2), total=timeout), retries=retries) + kwargs.update(timeout=Timeout(connect=max(1, timeout / 2), total=timeout), retries=retries) machines_cache = machines_cache[:nodes] @staticmethod @@ -584,7 +584,7 @@ class AbstractEtcd(AbstractDCS): ttl = int(ttl) ret = self._ttl != ttl self._ttl = ttl - self._client.set_machines_cache_ttl(ttl*10) + self._client.set_machines_cache_ttl(ttl * 10) return ret @property diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index b89684dd..b17e16fc 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -14,7 +14,7 @@ from threading import Condition, Lock, Thread from urllib3.exceptions import ReadTimeoutError, ProtocolError from . import ClusterConfig, Cluster, Failover, Leader, Member, SyncState,\ - TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re + TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors from ..exceptions import DCSError, PatroniException from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT @@ -411,15 +411,15 @@ class KVCache(Thread): new_value = kv.get('value') value_changed = old_value != new_value and \ - (key == self._leader_key or key in (self._optime_key, self._status_key) and new_value is not None or - key == self._config_key and old_value is not None and new_value is not None) + (key == self._leader_key or key in (self._optime_key, self._status_key) and new_value is not None + or key == self._config_key and old_value is not None and new_value is not None) if value_changed: logger.debug('%s changed from %s to %s', key, old_value, new_value) # We also want to wake up HA loop on replicas if leader optime (or status key) was updated - if value_changed and (key not in (self._optime_key, self._status_key) or - (self.get(self._leader_key) or {}).get('value') != self._name): + if value_changed and (key not in (self._optime_key, self._status_key) + or (self.get(self._leader_key) or {}).get('value') != self._name): self._dcs.event.set() def _process_message(self, message): diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index dbda2371..5c1db1b5 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -20,10 +20,10 @@ from typing import Any, Dict, List, Optional from urllib3.exceptions import HTTPError from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\ - TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re + TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re from ..exceptions import DCSError from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\ - Retry, RetryFailedError, tzutc, uri, USER_AGENT + Retry, RetryFailedError, tzutc, uri, USER_AGENT logger = logging.getLogger(__name__) @@ -143,7 +143,7 @@ class K8sConfig(object): if user.get('token'): self._make_headers(token=user['token']) elif 'username' in user and 'password' in user: - self._headers = self._make_headers(basic_auth=':'.join((user['username'], user['password']))) + self._headers = self._make_headers(basic_auth=':'.join((user['username'], user['password']))) @property def server(self): @@ -175,7 +175,7 @@ class K8sObject(object): if isinstance(value, dict): # we know that `annotations` and `labels` are dicts and therefore don't want to convert them into K8sObject return value if parent in {'annotations', 'labels'} and \ - all(isinstance(v, str) for v in value.values()) else cls(value) + all(isinstance(v, str) for v in value.values()) else cls(value) elif isinstance(value, list): return [cls._wrap(None, v) for v in value] else: @@ -264,7 +264,7 @@ class K8sClient(object): def _get_api_servers(self, api_servers_cache): _, per_node_timeout, per_node_retries = self._calculate_timeouts(len(api_servers_cache)) kwargs = {'headers': self._make_headers({}), 'preload_content': True, 'retries': per_node_retries, - 'timeout': urllib3.Timeout(connect=max(1, per_node_timeout/2.0), total=per_node_timeout)} + 'timeout': urllib3.Timeout(connect=max(1, per_node_timeout / 2.0), total=per_node_timeout)} path = self._API_URL_PREFIX + 'default/endpoints/kubernetes' for base_uri in api_servers_cache: try: @@ -382,7 +382,7 @@ class K8sClient(object): retries = 0 else: _, timeout, retries = self._calculate_timeouts(api_servers) - timeout = urllib3.Timeout(connect=max(1, timeout/2.0), total=timeout) + timeout = urllib3.Timeout(connect=max(1, timeout / 2.0), total=timeout) kwargs.update(retries=retries, timeout=timeout) while True: @@ -405,7 +405,8 @@ class K8sClient(object): retry.sleep_func(sleeptime) retry.update_delay() # We still have some time left. Partially reduce `api_servers_cache` and retry request - kwargs.update(timeout=urllib3.Timeout(connect=max(1, timeout/2.0), total=timeout), retries=retries) + kwargs.update(timeout=urllib3.Timeout(connect=max(1, timeout / 2.0), total=timeout), + retries=retries) api_servers_cache = api_servers_cache[:nodes] def call_api(self, method, path, headers=None, body=None, _retry=None, @@ -503,7 +504,7 @@ class CoreV1ApiProxy(object): # to start worrying (send keepalive messages). Finally, the connection should be # considered as dead if we received nothing from the socket after the ttl seconds. self._api_client.pool_manager.connection_pool_kw['socket_options'] = \ - list(keepalive_socket_options(ttl, int(loop_wait + retry_timeout))) + list(keepalive_socket_options(ttl, int(loop_wait + retry_timeout))) self._api_client.set_read_timeout(retry_timeout) self._api_client.set_api_servers_cache_ttl(loop_wait) @@ -905,7 +906,7 @@ class Kubernetes(AbstractDCS): # get synchronization state sync = nodes.get(path + self._SYNC) metadata = sync and sync.metadata - sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations) + sync = SyncState.from_node(metadata and metadata.resource_version, metadata and metadata.annotations) return Cluster(initialize, config, leader, last_lsn, members, failover, sync, history, slots, failsafe) diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index e09b3290..5d0dbc50 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -47,7 +47,7 @@ class SyncObjUtility(object): def __init__(self, otherNodes, conf, retry_timeout=10): self._nodes = otherNodes - self._utility = TcpUtility(conf.password, retry_timeout/max(1, len(otherNodes))) + self._utility = TcpUtility(conf.password, retry_timeout / max(1, len(otherNodes))) def executeCommand(self, command): try: @@ -83,8 +83,8 @@ class DynMemberSyncObj(SyncObj): def getMembers(self, args, callback): callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED - if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] + - [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None) + if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] + + [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None) def _onTick(self, timeToWait=0.0): super(DynMemberSyncObj, self)._onTick(timeToWait) diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index ac192bae..3f102f8d 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -28,7 +28,7 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler): self.set_connect_timeout(connect_timeout) def set_connect_timeout(self, connect_timeout): - self._connect_timeout = max(1.0, connect_timeout/2.0) # try to connect to zookeeper node during loop_wait/2 + self._connect_timeout = max(1.0, connect_timeout / 2.0) # try to connect to zookeeper node during loop_wait/2 def create_connection(self, *args, **kwargs): """This method is trying to establish connection with one of the zookeeper nodes. @@ -43,11 +43,11 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler): args = list(args) if len(args) == 0: # kazoo 2.6.0 slightly changed the way how it calls create_connection method - kwargs['timeout'] = max(self._connect_timeout, kwargs.get('timeout', self._connect_timeout*10)/10.0) + kwargs['timeout'] = max(self._connect_timeout, kwargs.get('timeout', self._connect_timeout * 10) / 10.0) elif len(args) == 1: args.append(self._connect_timeout) else: - args[1] = max(self._connect_timeout, args[1]/10.0) + args[1] = max(self._connect_timeout, args[1] / 10.0) return super(PatroniSequentialThreadingHandler, self).create_connection(*args, **kwargs) def select(self, *args, **kwargs): @@ -134,7 +134,7 @@ class ZooKeeper(AbstractDCS): `write_leader_optime()` methods, which also may hang...""" ret = self._orig_kazoo_connect(*args) - return max(self.loop_wait - 2, 2)*1000, ret[1] + return max(self.loop_wait - 2, 2) * 1000, ret[1] def session_listener(self, state): if state in [KazooState.SUSPENDED, KazooState.LOST]: @@ -385,12 +385,13 @@ class ZooKeeper(AbstractDCS): member = cluster and cluster.get_member(self._name, fallback_to_leader=False) member_data = self.__last_member_data or member and member.data # We want to notify leader if some important fields in the member key changed by removing ZNode - if member and (self._client.client_id is not None and member.session != self._client.client_id[0] or - not (deep_compare(member_data.get('tags', {}), data.get('tags', {})) and - (member_data.get('state') == data.get('state') or - 'running' not in (member_data.get('state'), data.get('state'))) and - member_data.get('version') == data.get('version') and - member_data.get('checkpoint_after_promote') == data.get('checkpoint_after_promote'))): + if member and (self._client.client_id is not None and member.session != self._client.client_id[0] + or not (deep_compare(member_data.get('tags', {}), data.get('tags', {})) + and (member_data.get('state') == data.get('state') + or 'running' not in (member_data.get('state'), data.get('state'))) + and member_data.get('version') == data.get('version') + and member_data.get('checkpoint_after_promote') + == data.get('checkpoint_after_promote'))): try: self._client.delete_async(self.member_path).get(timeout=1) except NoNodeError: diff --git a/patroni/ha.py b/patroni/ha.py index 402f9ba4..ebb69d03 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -377,14 +377,13 @@ class Ha(object): return 'failed to acquire initialize lock' else: create_replica_methods = self.global_config.get_standby_cluster_config().get('create_replica_methods', []) \ - if self.is_standby_cluster() else None + if self.is_standby_cluster() else None can_bootstrap = self.state_handler.can_create_replica_without_replication_connection(create_replica_methods) concurrent_bootstrap = self.cluster.initialize == "" if can_bootstrap and not concurrent_bootstrap: msg = 'bootstrap (without leader)' return self._async_executor.try_run_async(msg, self.clone) or 'trying to ' + msg - return 'waiting for {0}leader to bootstrap'.format( - 'standby_' if self.is_standby_cluster() else '') + return 'waiting for {0}leader to bootstrap'.format('standby_' if self.is_standby_cluster() else '') def bootstrap_standby_leader(self): """ If we found 'standby' key in the configuration, we need to bootstrap @@ -651,7 +650,7 @@ class Ha(object): if self.touch_member(): # Primary should notice the updated value during the next cycle. We will wait double that, if primary # hasn't noticed the value by then not disabling sync replication is not likely to matter. - for _ in polling_loop(timeout=self.dcs.loop_wait*2, interval=2): + for _ in polling_loop(timeout=self.dcs.loop_wait * 2, interval=2): try: if not self.is_sync_standby(self.dcs.get_cluster()): break @@ -970,7 +969,7 @@ class Ha(object): if self.state_handler.is_leader(): # in pause leader is the healthiest only when no initialize or sysid matches with initialize! return not self.is_paused() or not self.cluster.initialize\ - or self.state_handler.sysid == self.cluster.initialize + or self.state_handler.sysid == self.cluster.initialize if self.is_paused(): return False @@ -1035,10 +1034,11 @@ class Ha(object): PostgreSQL as quickly as possible without regard for data durability. May only be called synchronously. """ mode_control = { - 'offline': dict(stop='fast', checkpoint=False, release=False, offline=True, async_req=False), - 'graceful': dict(stop='fast', checkpoint=True, release=True, offline=False, async_req=False), - 'immediate': dict(stop='immediate', checkpoint=False, release=True, offline=False, async_req=True), - 'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True), + 'offline': dict(stop='fast', checkpoint=False, release=False, offline=True, async_req=False), # noqa: E241,E501 + 'graceful': dict(stop='fast', checkpoint=True, release=True, offline=False, async_req=False), # noqa: E241,E501 + 'immediate': dict(stop='immediate', checkpoint=False, release=True, offline=False, async_req=True), # noqa: E241,E501 + 'immediate-nolock': dict(stop='immediate', checkpoint=False, release=False, offline=False, async_req=True), # noqa: E241,E501 + }[mode] logger.info('Demoting self (%s)', mode) @@ -1267,10 +1267,10 @@ class Ha(object): if self.is_standby_cluster(): return self.follow('cannot be a real primary in a standby cluster', 'no action. I am ({0}), a secondary, and following a standby leader ({1})'.format( - self.state_handler.name, lock_owner), refresh=False) + self.state_handler.name, lock_owner), refresh=False) return self.follow('demoting self because I do not have the lock and I was a leader', 'no action. I am ({0}), a secondary, and following a leader ({1})'.format( - self.state_handler.name, lock_owner), refresh=False) + self.state_handler.name, lock_owner), refresh=False) def evaluate_scheduled_restart(self): if self._async_executor.busy: # Restart already in progress @@ -1287,8 +1287,8 @@ class Ha(object): self.delete_future_restart() return None - if (restart_data and - self.should_run_scheduled_action('restart', restart_data['schedule'], self.delete_future_restart)): + if restart_data\ + and self.should_run_scheduled_action('restart', restart_data['schedule'], self.delete_future_restart): try: ret, message = self.restart(restart_data, run_async=True) if not ret: @@ -1336,8 +1336,8 @@ class Ha(object): return ret def future_restart_scheduled(self): - return self.patroni.scheduled_restart.copy() if (self.patroni.scheduled_restart and - isinstance(self.patroni.scheduled_restart, dict)) else None + return self.patroni.scheduled_restart.copy()\ + if (self.patroni.scheduled_restart and isinstance(self.patroni.scheduled_restart, dict)) else None def restart_scheduled(self): return self._async_executor.scheduled_action == 'restart' @@ -1637,7 +1637,7 @@ class Ha(object): if self.has_lock(): self.release_leader_key_voluntarily() return 'released leader key voluntarily as data dir {0} and currently leader'.format( - 'empty' if data_directory_is_accessible else 'not accessible') + 'empty' if data_directory_is_accessible else 'not accessible') if not data_directory_is_accessible: return 'data directory is not accessible: {0}'.format(data_directory_error) @@ -1827,7 +1827,7 @@ class Ha(object): # XXX: what about when Patroni is started as the wrong user that has access to the watchdog device # but cannot shut down PostgreSQL. Root would be the obvious example. Would be nice to not kill the # system due to a bad config. - logger.error("PostgreSQL shutdown failed, leader key not removed." + + logger.error("PostgreSQL shutdown failed, leader key not removed.%s", (" Leaving watchdog running." if self.watchdog.is_running else "")) def watch(self, timeout): diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py index 0b1dba5a..472d0adb 100644 --- a/patroni/postgresql/__init__.py +++ b/patroni/postgresql/__init__.py @@ -93,11 +93,11 @@ class Postgresql(object): self.cancellable = CancellableSubprocess() self._sysid = None - self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1, + self.retry = Retry(max_tries=-1, deadline=config['retry_timeout'] / 2.0, max_delay=1, retry_exceptions=PostgresConnectionException) # Retry 'pg_is_in_recovery()' only once - self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout']/2.0, max_delay=1, + self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout'] / 2.0, max_delay=1, retry_exceptions=PostgresConnectionException) self._role_lock = Lock() @@ -173,21 +173,21 @@ class Postgresql(object): If some conditions are not satisfied we simply put static values instead. E.g., NULL, 0, '', and so on.""" - extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), " + + extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), " "pg_catalog.current_setting('synchronous_standby_names'), " - "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT w.pid as pid, application_name, sync_state," + - " pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn," + - " pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn," + - " pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn " + - "FROM pg_catalog.pg_stat_get_wal_senders() w," + - " pg_catalog.pg_stat_get_activity(w.pid)" + + "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT w.pid as pid, application_name, sync_state," + " pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn," + " pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn," + " pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn " + "FROM pg_catalog.pg_stat_get_wal_senders() w," + " pg_catalog.pg_stat_get_activity(w.pid)" " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name) if (not self._global_config or self._global_config.is_synchronous_mode) and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL") if self._major_version >= 90600: - extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " + - "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" + + extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, " + "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)" if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END," @@ -253,7 +253,7 @@ class Postgresql(object): def reload_config(self, config, sighup=False): self.config.reload_config(config, sighup) - self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout']/2.0 + self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout'] / 2.0 @property def pending_restart(self): @@ -331,8 +331,8 @@ class Postgresql(object): return deepcopy(self.config.get(method, {})) def replica_method_can_work_without_replication_connection(self, method): - return method != 'basebackup' and (self.replica_method_options(method).get('no_master') or - self.replica_method_options(method).get('no_leader')) + return method != 'basebackup' and (self.replica_method_options(method).get('no_master') + or self.replica_method_options(method).get('no_leader')) def can_create_replica_without_replication_connection(self, replica_methods=None): """ go through the replication methods to see if there are ones @@ -374,8 +374,8 @@ class Postgresql(object): # We want to enable hot_standby_feedback if the replica is supposed # to have a logical slot or in case if it is the cascading replica. self.set_enforce_hot_standby_feedback( - self._has_permanent_logical_slots or - cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) + self._has_permanent_logical_slots + or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version)) self._global_config = global_config @@ -971,8 +971,8 @@ class Postgresql(object): # and we know for sure that postgres was already running before, we will only execute on_role_change # callback and prevent execution of on_restart/on_start callback. # If the role remains the same (replica or standby_leader), we will execute on_start or on_restart - change_role = self.cb_called and (self.role in ('master', 'primary', 'demoted') or - not {'standby_leader', 'replica'} - {self.role, role}) + change_role = self.cb_called and (self.role in ('master', 'primary', 'demoted') + or not {'standby_leader', 'replica'} - {self.role, role}) if change_role: self.__cb_pending = CallbackAction.NOOP diff --git a/patroni/postgresql/cancellable.py b/patroni/postgresql/cancellable.py index 09057e8e..64a9e6d7 100644 --- a/patroni/postgresql/cancellable.py +++ b/patroni/postgresql/cancellable.py @@ -29,7 +29,7 @@ class CancellableExecutor(object): self._process_cmd = cmd self._process = psutil.Popen(cmd, *args, **kwargs) except Exception: - return logger.exception('Failed to execute %s', cmd) + return logger.exception('Failed to execute %s', cmd) return True def _kill_process(self): diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/citus.py index 2c7b1964..c4bfac15 100644 --- a/patroni/postgresql/citus.py +++ b/patroni/postgresql/citus.py @@ -326,7 +326,7 @@ class CitusHandler(Thread): task = self.add_task(event['type'], event['group'], cluster.leader.conn_url, - event['timeout'], event['cooldown']*1000) + event['timeout'], event['cooldown'] * 1000) if task and event['type'] == 'before_demote': task.wait() diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py index 87544558..d2a097bc 100644 --- a/patroni/postgresql/config.py +++ b/patroni/postgresql/config.py @@ -9,11 +9,11 @@ import time from urllib.parse import urlparse, parse_qsl, unquote from .validator import CaseInsensitiveDict, recovery_parameters,\ - transform_postgresql_parameter_value, transform_recovery_parameter_value + transform_postgresql_parameter_value, transform_recovery_parameter_value from ..dcs import RemoteMember, slot_name_from_member_name from ..exceptions import PatroniFatalException from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \ - validate_directory, is_subpath + validate_directory, is_subpath logger = logging.getLogger(__name__) @@ -520,8 +520,8 @@ class ConfigHandler(object): def build_recovery_params(self, member): recovery_params = CaseInsensitiveDict({p: v for p, v in self.get('recovery_conf', {}).items() - if not p.lower().startswith('recovery_target') and - p.lower() not in ('primary_conninfo', 'primary_slot_name')}) + if not p.lower().startswith('recovery_target') + and p.lower() not in ('primary_conninfo', 'primary_slot_name')}) recovery_params.update({'standby_mode': 'on', 'recovery_target_timeline': 'latest'}) if self._postgresql.major_version >= 120000: # on pg12 we want to protect from following params being set in one of included files @@ -864,8 +864,8 @@ class ConfigHandler(object): self._postgresql.citus_handler.adjust_postgres_gucs(parameters) - ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._postgresql.major_version or - self._postgresql.major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]}) + ret = CaseInsensitiveDict({k: v for k, v in parameters.items() if not self._postgresql.major_version + or self._postgresql.major_version >= self.CMDLINE_OPTIONS.get(k, (0, 1, 90100))[2]}) ret.update({k: os.path.join(self._config_dir, ret[k]) for k in ('hba_file', 'ident_file') if k in ret}) return ret @@ -924,8 +924,8 @@ class ConfigHandler(object): def _get_pg_settings(self, names): return {r[0]: r for r in self._postgresql.query(('SELECT name, setting, unit, vartype, context, sourcefile' - + ' FROM pg_catalog.pg_settings ' + - ' WHERE pg_catalog.lower(name) = ANY(%s)'), + + ' FROM pg_catalog.pg_settings ' + + ' WHERE pg_catalog.lower(name) = ANY(%s)'), [n.lower() for n in names])} @staticmethod @@ -1123,7 +1123,7 @@ class ConfigHandler(object): @property def rewind_credentials(self): return self._config['authentication'].get('rewind', self._superuser) \ - if self._postgresql.major_version >= 110000 else self._superuser + if self._postgresql.major_version >= 110000 else self._superuser @property def ident_file(self): diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 3dc5fcd1..ba323eb5 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -78,9 +78,9 @@ class Rewind(object): def check_leader_has_run_checkpoint(conn_kwargs): try: with get_connection_cursor(connect_timeout=3, options='-c statement_timeout=2000', **conn_kwargs) as cur: - cur.execute("SELECT NOT pg_catalog.pg_is_in_recovery()" + - " AND ('x' || pg_catalog.substr(pg_catalog.pg_walfile_name(" + - " pg_catalog.pg_current_wal_lsn()), 1, 8))::bit(32)::int = timeline_id" + + cur.execute("SELECT NOT pg_catalog.pg_is_in_recovery()" + " AND ('x' || pg_catalog.substr(pg_catalog.pg_walfile_name(" + " pg_catalog.pg_current_wal_lsn()), 1, 8))::bit(32)::int = timeline_id" " FROM pg_catalog.pg_control_checkpoint()") if not cur.fetchone()[0]: return 'leader has not run a checkpoint yet' @@ -384,9 +384,10 @@ class Rewind(object): if self._postgresql.major_version < 120000 else self._postgresql.get_guc_value('restore_command') # Until v15 pg_rewind expected postgresql.conf to be inside $PGDATA, which is not the case on e.g. Debian - pg_rewind_can_restore = restore_command and (self._postgresql.major_version >= 150000 or - (self._postgresql.major_version >= 130000 and - self._postgresql.config._config_dir == self._postgresql.data_dir)) + pg_rewind_can_restore = restore_command and (self._postgresql.major_version >= 150000 + or (self._postgresql.major_version >= 130000 + and self._postgresql.config._config_dir + == self._postgresql.data_dir)) cmd = [self._postgresql.pgcommand('pg_rewind')] if pg_rewind_can_restore: @@ -439,7 +440,7 @@ class Rewind(object): # superuser credentials match rewind_credentials if the latter are not provided or we run 10 or older if self._postgresql.config.superuser == self._postgresql.config.rewind_credentials: leader_status = self._postgresql.checkpoint( - self._conn_kwargs(leader, self._postgresql.config.superuser)) + self._conn_kwargs(leader, self._postgresql.config.superuser)) else: # we run 11+ and have a dedicated pg_rewind user leader_status = self.check_leader_has_run_checkpoint(r) if leader_status: # we tried to run/check for a checkpoint on the remote leader, but it failed diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py index c304b1f3..aa5efd8e 100644 --- a/patroni/postgresql/slots.py +++ b/patroni/postgresql/slots.py @@ -14,8 +14,8 @@ logger = logging.getLogger(__name__) def compare_slots(s1, s2, dbid='database'): - return s1['type'] == s2['type'] and (s1['type'] == 'physical' or - s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin']) + return s1['type'] == s2['type'] and (s1['type'] == 'physical' + or s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin']) class SlotsAdvanceThread(Thread): @@ -180,11 +180,11 @@ class SlotsHandler(object): def drop_replication_slot(self, name): """Returns a tuple(active, dropped)""" - cursor = self._query(('WITH slots AS (SELECT slot_name, active' + - ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' + - ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' + - ' true AS dropped FROM slots WHERE not active) ' + - 'SELECT active, COALESCE(dropped, false) FROM slots' + + cursor = self._query(('WITH slots AS (SELECT slot_name, active' + ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' + ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' + ' true AS dropped FROM slots WHERE not active) ' + 'SELECT active, COALESCE(dropped, false) FROM slots' ' FULL OUTER JOIN dropped ON true'), name) return cursor.fetchone() if cursor.rowcount == 1 else (False, False) @@ -216,8 +216,8 @@ class SlotsHandler(object): for name, value in slots.items(): if name not in self._replication_slots and value['type'] == 'physical': try: - self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" + - " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + + self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})" + " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" " WHERE slot_type = 'physical' AND slot_name = %s)").format( immediately_reserve), name, name) except Exception: @@ -247,8 +247,8 @@ class SlotsHandler(object): with self.get_local_connection_cursor(dbname=database) as cur: for name, value in values.items(): try: - cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" + - " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" + + cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" + " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots" " WHERE slot_type = 'logical' AND slot_name = %s)", (name, value['plugin'], name)) except Exception as e: @@ -378,7 +378,7 @@ class SlotsHandler(object): if compare_slots(slot, slots[r[0]]): create_slots[r[0]] = slot else: - logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: ' + + logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: ' 'configuration=%s, slot on the primary=%s', r[0], slots[r[0]], slot) except Exception as e: logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e) diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 550248f7..dd75c293 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -243,8 +243,8 @@ class SyncHandler(object): else: sync_param = next(iter(value), None) - if not (self._postgresql.config.set_synchronous_standby_names(sync_param) and - self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']: + if not (self._postgresql.config.set_synchronous_standby_names(sync_param) + and self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']: return time.sleep(0.1) # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms diff --git a/patroni/validator.py b/patroni/validator.py index d755a0bc..890b3d56 100644 --- a/patroni/validator.py +++ b/patroni/validator.py @@ -666,7 +666,7 @@ def _get_type_name(python_type: Any) -> str: """ return {str: 'a string', int: 'an integer', float: 'a number', bool: 'a boolean', list: 'an array', dict: 'a dictionary'}.get( - python_type, getattr(python_type, __name__, "unknown type")) + python_type, getattr(python_type, __name__, "unknown type")) def assert_(condition: bool, message: OptionalType[str] = "Wrong value") -> None: @@ -699,89 +699,89 @@ validate_etcd = { } schema = Schema({ - "name": str, - "scope": str, - "restapi": { - "listen": validate_host_port_listen, - "connect_address": validate_connect_address - }, - Optional("bootstrap"): { - "dcs": { - Optional("ttl"): int, - Optional("loop_wait"): int, - Optional("retry_timeout"): int, - Optional("maximum_lag_on_failover"): int + "name": str, + "scope": str, + "restapi": { + "listen": validate_host_port_listen, + "connect_address": validate_connect_address + }, + Optional("bootstrap"): { + "dcs": { + Optional("ttl"): int, + Optional("loop_wait"): int, + Optional("retry_timeout"): int, + Optional("maximum_lag_on_failover"): int }, - "pg_hba": [str], - "initdb": [Or(str, dict)] - }, - Or(*available_dcs): Case({ - "consul": { - Or("host", "url"): Case({ - "host": validate_host_port, - "url": str}) - }, - "etcd": validate_etcd, - "etcd3": validate_etcd, - "exhibitor": { - "hosts": [str], - "port": lambda i: assert_(int(i) <= 65535), - Optional("pool_interval"): int - }, - "raft": { - "self_addr": validate_connect_address, - Optional("bind_addr"): validate_host_port_listen, - "partner_addrs": validate_host_port_list, - Optional("data_dir"): str, - Optional("password"): str - }, - "zookeeper": { - "hosts": Or(comma_separated_host_port, [validate_host_port]), - }, - "kubernetes": { - "labels": {}, - Optional("namespace"): str, - Optional("scope_label"): str, - Optional("role_label"): str, - Optional("use_endpoints"): bool, - Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address), - Optional("ports"): [{"name": str, "port": int}], - Optional("retriable_http_codes"): Or(int, [int]), - }, - }), - Optional("citus"): { - "database": str, - "group": int - }, - "postgresql": { - "listen": validate_host_port_listen_multiple_hosts, - "connect_address": validate_connect_address, - Optional("proxy_address"): validate_connect_address, - "authentication": { - "replication": userattributes, - "superuser": userattributes, - "rewind": userattributes + "pg_hba": [str], + "initdb": [Or(str, dict)] }, - "data_dir": validate_data_dir, - Optional("bin_dir"): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup", - "postgres", "pg_isready"]), - Optional("parameters"): { - Optional("unix_socket_directories"): lambda s: assert_(all([isinstance(s, str), len(s)])) + Or(*available_dcs): Case({ + "consul": { + Or("host", "url"): Case({ + "host": validate_host_port, + "url": str}) + }, + "etcd": validate_etcd, + "etcd3": validate_etcd, + "exhibitor": { + "hosts": [str], + "port": lambda i: assert_(int(i) <= 65535), + Optional("pool_interval"): int + }, + "raft": { + "self_addr": validate_connect_address, + Optional("bind_addr"): validate_host_port_listen, + "partner_addrs": validate_host_port_list, + Optional("data_dir"): str, + Optional("password"): str + }, + "zookeeper": { + "hosts": Or(comma_separated_host_port, [validate_host_port]), + }, + "kubernetes": { + "labels": {}, + Optional("namespace"): str, + Optional("scope_label"): str, + Optional("role_label"): str, + Optional("use_endpoints"): bool, + Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address), + Optional("ports"): [{"name": str, "port": int}], + Optional("retriable_http_codes"): Or(int, [int]), + }, + }), + Optional("citus"): { + "database": str, + "group": int }, - Optional("pg_hba"): [str], - Optional("pg_ident"): [str], - Optional("pg_ctl_timeout"): int, - Optional("use_pg_rewind"): bool - }, - Optional("watchdog"): { - Optional("mode"): lambda m: assert_(m in ["off", "automatic", "required"]), - Optional("device"): str - }, - Optional("tags"): { - Optional("nofailover"): bool, - Optional("clonefrom"): bool, - Optional("noloadbalance"): bool, - Optional("replicatefrom"): str, - Optional("nosync"): bool - } + "postgresql": { + "listen": validate_host_port_listen_multiple_hosts, + "connect_address": validate_connect_address, + Optional("proxy_address"): validate_connect_address, + "authentication": { + "replication": userattributes, + "superuser": userattributes, + "rewind": userattributes + }, + "data_dir": validate_data_dir, + Optional("bin_dir"): Directory(contains_executable=["pg_ctl", "initdb", "pg_controldata", "pg_basebackup", + "postgres", "pg_isready"]), + Optional("parameters"): { + Optional("unix_socket_directories"): lambda s: assert_(all([isinstance(s, str), len(s)])) + }, + Optional("pg_hba"): [str], + Optional("pg_ident"): [str], + Optional("pg_ctl_timeout"): int, + Optional("use_pg_rewind"): bool + }, + Optional("watchdog"): { + Optional("mode"): lambda m: assert_(m in ["off", "automatic", "required"]), + Optional("device"): str + }, + Optional("tags"): { + Optional("nofailover"): bool, + Optional("clonefrom"): bool, + Optional("noloadbalance"): bool, + Optional("replicatefrom"): str, + Optional("nosync"): bool + } }) diff --git a/tests/test_api.py b/tests/test_api.py index f167735e..85c01b49 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -398,8 +398,8 @@ class TestRestApiHandler(unittest.TestCase): with patch.object(MockHa, 'is_failsafe_mode', Mock(return_value=False), create=True): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization)) with patch.object(MockHa, 'is_failsafe_mode', Mock(return_value=True), create=True): - self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization + - '\nContent-Length: 9\n\n{"a":"b"}')) + self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /failsafe HTTP/1.0' + self._authorization + + '\nContent-Length: 9\n\n{"a":"b"}')) @patch.object(MockPatroni, 'sighup_handler', Mock()) def test_do_POST_reload(self): diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py index 1c16306f..1cce58b9 100644 --- a/tests/test_bootstrap.py +++ b/tests/test_bootstrap.py @@ -132,7 +132,7 @@ class TestBootstrap(BaseTestPostgresql): @patch.object(CancellableSubprocess, 'call') @patch.object(Postgresql, 'get_major_version', Mock(return_value=90600)) - @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'in production'})) + @patch.object(Postgresql, 'controldata', Mock(return_value={'Database cluster state': 'in production'})) def test_custom_bootstrap(self, mock_cancellable_subprocess_call): self.p.config._config.pop('pg_hba') config = {'method': 'foo', 'foo': {'command': 'bar'}} diff --git a/tests/test_consul.py b/tests/test_consul.py index 7934551c..7bbdbc02 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -4,7 +4,7 @@ import unittest from consul import ConsulException, NotFound from mock import Mock, PropertyMock, patch from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \ - ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError + ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError from . import SleepException @@ -26,12 +26,12 @@ def kv_get(self, key, **kwargs): 'ModifyIndex': 2621, 'Session': 'fd4f44fe-2cac-bba5-a60b-304b51ff39b7', 'Value': b'postgresql1'}, {'CreateIndex': 6156, 'Flags': 0, 'Key': key + 'members/postgresql0', 'LockIndex': 1, 'ModifyIndex': 6156, 'Session': '782e6da4-ed02-3aef-7963-99a90ed94b53', - 'Value': ('postgres://replicator:rep-pass@127.0.0.1:5432/postgres' + - '?application_name=http://127.0.0.1:8008/patroni').encode('utf-8')}, + 'Value': ('postgres://replicator:rep-pass@127.0.0.1:5432/postgres' + + '?application_name=http://127.0.0.1:8008/patroni').encode('utf-8')}, {'CreateIndex': 2630, 'Flags': 0, 'Key': key + 'members/postgresql1', 'LockIndex': 1, 'ModifyIndex': 2630, 'Session': 'fd4f44fe-2cac-bba5-a60b-304b51ff39b7', - 'Value': ('postgres://replicator:rep-pass@127.0.0.1:5433/postgres' + - '?application_name=http://127.0.0.1:8009/patroni').encode('utf-8')}, + 'Value': ('postgres://replicator:rep-pass@127.0.0.1:5433/postgres' + + '?application_name=http://127.0.0.1:8009/patroni').encode('utf-8')}, {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'optime/leader', 'LockIndex': 0, 'ModifyIndex': 6429, 'Value': b'4496294792'}, {'CreateIndex': 1085, 'Flags': 0, 'Key': key + 'sync', 'LockIndex': 0, diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 7f300deb..65f9d444 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -470,7 +470,7 @@ class TestCtl(unittest.TestCase): scheduled_at = datetime.now(tzutc) + timedelta(seconds=600) mock_get_dcs.return_value.get_cluster = Mock( - return_value=get_cluster_initialized_with_leader(Failover(1, 'a', 'b', scheduled_at))) + return_value=get_cluster_initialized_with_leader(Failover(1, 'a', 'b', scheduled_at))) result = self.runner.invoke(ctl, ['flush', 'dummy', 'switchover']) assert result.output.startswith('Success: ') diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 817358a2..8f780342 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -61,13 +61,13 @@ def etcd_read(self, key, **kwargs): "modifiedIndex": 1582, "createdIndex": 1582}, {"key": "/service/batman5/members", "dir": True, "nodes": [ {"key": "/service/batman5/members/postgresql1", - "value": "postgres://replicator:rep-pass@127.0.0.1:5434/postgres" + - "?application_name=http://127.0.0.1:8009/patroni", + "value": "postgres://replicator:rep-pass@127.0.0.1:5434/postgres" + + "?application_name=http://127.0.0.1:8009/patroni", "expiration": "2015-05-15T09:10:59.949384522Z", "ttl": 21, "modifiedIndex": 20727, "createdIndex": 20727}, {"key": "/service/batman5/members/postgresql0", - "value": "postgres://replicator:rep-pass@127.0.0.1:5433/postgres" + - "?application_name=http://127.0.0.1:8008/patroni", + "value": "postgres://replicator:rep-pass@127.0.0.1:5433/postgres" + + "?application_name=http://127.0.0.1:8008/patroni", "expiration": "2015-05-15T09:11:09.611860899Z", "ttl": 30, "modifiedIndex": 20730, "createdIndex": 20730}], "modifiedIndex": 1581, "createdIndex": 1581}, diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 4fad7765..58433bbb 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -6,7 +6,7 @@ import urllib3 from mock import Mock, patch from patroni.dcs.etcd import DnsCachingResolver from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3Client, Etcd3Error, Etcd3ClientError, RetryFailedError,\ - InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3 + InvalidAuthToken, Unavailable, Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, base64_encode, Etcd3 from threading import Thread from . import SleepException, MockResponse diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index f92f1f40..86669a08 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -8,8 +8,8 @@ import unittest from mock import Mock, PropertyMock, mock_open, patch from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed,\ - K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\ - Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME + K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException,\ + Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME from threading import Thread from . import MockResponse, SleepException diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index f8fa13c1..bab65159 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -357,27 +357,27 @@ class TestPostgresql(BaseTestPostgresql): self.assertEqual(self.p.latest_checkpoint_location(), 28163096) # 9.3 and 9.4 format mock_popen.return_value.communicate.side_effect = [ - (b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' + - b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' + - b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None), - (b'rmgr: Transaction len (rec/tot): 64/ 96, tx: 726, lsn: 0/01ADBBB8, prev 0/01ADBB70, ' + - b'bkp: 0000, desc: commit: 2021-02-26 11:19:37.900918 CET; inval msgs: catcache 11 catcache 10', None)] + (b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' + + b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' + + b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None), + (b'rmgr: Transaction len (rec/tot): 64/ 96, tx: 726, lsn: 0/01ADBBB8, prev 0/01ADBB70, ' + + b'bkp: 0000, desc: commit: 2021-02-26 11:19:37.900918 CET; inval msgs: catcache 11 catcache 10', None)] self.assertEqual(self.p.latest_checkpoint_location(), 28163096) mock_popen.return_value.communicate.side_effect = [ - (b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' + - b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' + - b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None), - (b'rmgr: XLOG len (rec/tot): 0/ 32, tx: 0, lsn: 0/01ADBBB8, prev 0/01ADBBA0, ' + - b'bkp: 0000, desc: xlog switch ', None)] + (b'rmgr: XLOG len (rec/tot): 72/ 104, tx: 0, lsn: 0/01ADBC18, prev 0/01ADBBB8, ' + + b'bkp: 0000, desc: checkpoint: redo 0/1ADBC18; tli 1; prev tli 1; fpw true; xid 0/727; oid 16386; multi' + + b' 1; offset 0; oldest xid 715 in DB 1; oldest multi 1 in DB 1; oldest running xid 0; shutdown', None), + (b'rmgr: XLOG len (rec/tot): 0/ 32, tx: 0, lsn: 0/01ADBBB8, prev 0/01ADBBA0, ' + + b'bkp: 0000, desc: xlog switch ', None)] self.assertEqual(self.p.latest_checkpoint_location(), 28163000) # 9.5+ format mock_popen.return_value.communicate.side_effect = [ - (b'rmgr: XLOG len (rec/tot): 114/ 114, tx: 0, lsn: 0/01ADBC18, prev 0/018260F8, ' + - b'desc: CHECKPOINT_SHUTDOWN redo 0/1825ED8; tli 1; prev tli 1; fpw true; xid 0:494; oid 16387; multi 1' + - b'; offset 0; oldest xid 479 in DB 1; oldest multi 1 in DB 1; oldest/newest commit timestamp xid: 0/0;' + - b' oldest running xid 0; shutdown', None), - (b'rmgr: XLOG len (rec/tot): 24/ 24, tx: 0, lsn: 0/018260F8, prev 0/01826080, ' + - b'desc: SWITCH ', None)] + (b'rmgr: XLOG len (rec/tot): 114/ 114, tx: 0, lsn: 0/01ADBC18, prev 0/018260F8, ' + + b'desc: CHECKPOINT_SHUTDOWN redo 0/1825ED8; tli 1; prev tli 1; fpw true; xid 0:494; oid 16387; multi 1' + + b'; offset 0; oldest xid 479 in DB 1; oldest multi 1 in DB 1; oldest/newest commit timestamp xid: 0/0;' + + b' oldest running xid 0; shutdown', None), + (b'rmgr: XLOG len (rec/tot): 24/ 24, tx: 0, lsn: 0/018260F8, prev 0/01826080, ' + + b'desc: SWITCH ', None)] self.assertEqual(self.p.latest_checkpoint_location(), 25321720) def test_reload(self): diff --git a/tests/test_raft.py b/tests/test_raft.py index 21306e20..4f8ddb2c 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -5,7 +5,7 @@ import time from mock import Mock, PropertyMock, patch from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL,\ - Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport + Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport from pysyncobj import SyncObjConf, FAIL_REASON diff --git a/tests/test_rewind.py b/tests/test_rewind.py index e6a79026..3d921621 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -20,8 +20,8 @@ class MockThread(object): def mock_cancellable_call(*args, **kwargs): communicate = kwargs.pop('communicate', None) if isinstance(communicate, dict): - communicate.update(stdout=b'', stderr=b'pg_rewind: error: could not open file ' + - b'"data/postgresql0/pg_xlog/000000010000000000000003": No such file') + communicate.update(stdout=b'', stderr=b'pg_rewind: error: could not open file ' + + b'"data/postgresql0/pg_xlog/000000010000000000000003": No such file') return 1 @@ -222,7 +222,7 @@ class TestRewind(BaseTestPostgresql): @patch('patroni.postgresql.rewind.logger.info') def test_archive_ready_wals(self, mock_logger_info): with patch('os.listdir', Mock(side_effect=OSError)), \ - patch.object(Postgresql, 'get_guc_value', Mock(side_effect=['on', 'command %f'])): + patch.object(Postgresql, 'get_guc_value', Mock(side_effect=['on', 'command %f'])): self.r._archive_ready_wals() mock_logger_info.assert_not_called() @@ -232,19 +232,19 @@ class TestRewind(BaseTestPostgresql): 'on', '', ] with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=get_guc_value_res)): - for _ in range(len(get_guc_value_res)//2): + for _ in range(len(get_guc_value_res) // 2): self.r._archive_ready_wals() mock_logger_info.assert_not_called() with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])): # successful archive_command call - with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)): + with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)): get_guc_value_res = [ 'on', 'command %f', 'always', 'command %f', ] with patch.object(Postgresql, 'get_guc_value', Mock(side_effect=get_guc_value_res)): - for _ in range(len(get_guc_value_res)//2): + for _ in range(len(get_guc_value_res) // 2): self.r._archive_ready_wals() mock_logger_info.assert_called_once() self.assertEqual(('Trying to archive %s: %s', diff --git a/tests/test_validator.py b/tests/test_validator.py index 1558a17c..df017500 100644 --- a/tests/test_validator.py +++ b/tests/test_validator.py @@ -23,7 +23,7 @@ config = { "loop_wait": 1000, "retry_timeout": 1000, "maximum_lag_on_failover": 1000 - }, + }, "pg_hba": ["string"], "initdb": ["string", {"key": "value"}] }, @@ -49,7 +49,7 @@ config = { "password": "12345" }, "zookeeper": { - "hosts": "127.0.0.1:3379,127.0.0.1:3380" + "hosts": "127.0.0.1:3379,127.0.0.1:3380" }, "kubernetes": { "namespace": "string", @@ -85,10 +85,10 @@ config = { "device": "string" }, "tags": { - "nofailover": False, - "clonefrom": False, - "noloadbalance": False, - "nosync": False + "nofailover": False, + "clonefrom": False, + "noloadbalance": False, + "nosync": False } } diff --git a/tests/test_watchdog.py b/tests/test_watchdog.py index 6d68e1ea..78752719 100644 --- a/tests/test_watchdog.py +++ b/tests/test_watchdog.py @@ -37,7 +37,7 @@ def mock_ioctl(fd, op, arg=None, mutate_flag=False): sys.stderr.write("Get support\n") assert (mutate_flag is True) arg.options = sum(map(linuxwd.WDIOF.get, ['SETTIMEOUT', 'KEEPALIVEPING'])) - arg.identity = (ctypes.c_ubyte*32)(*map(ord, 'Mock Watchdog')) + arg.identity = (ctypes.c_ubyte * 32)(*map(ord, 'Mock Watchdog')) elif op == linuxwd.WDIOC_GETTIMEOUT: arg.value = dev.timeout elif op == linuxwd.WDIOC_SETTIMEOUT: diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 0fa0817b..067877b5 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -8,7 +8,7 @@ from kazoo.protocol.states import KeeperState, ZnodeStat from kazoo.retry import RetryFailedError from mock import Mock, PropertyMock, patch from patroni.dcs.zookeeper import Cluster, Leader, PatroniKazooClient,\ - PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError + PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError class MockKazooClient(Mock): diff --git a/tox.ini b/tox.ini index aa079ec5..67616ade 100644 --- a/tox.ini +++ b/tox.ini @@ -1,2 +1,3 @@ [flake8] max-line-length=120 +ignore=D401,W503