diff --git a/features/environment.py b/features/environment.py index 851c41a1..ab65768d 100644 --- a/features/environment.py +++ b/features/environment.py @@ -422,7 +422,7 @@ class KubernetesController(AbstractDcsController): os.environ['PATRONI_KUBERNETES_LABELS'] = json.dumps(self._labels) os.environ['PATRONI_KUBERNETES_USE_ENDPOINTS'] = 'true' - from kubernetes import client as k8s_client, config as k8s_config + from patroni.dcs.kubernetes import k8s_client, k8s_config k8s_config.load_kube_config(context='local') self._client = k8s_client self._api = self._client.CoreV1Api() diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index a4630578..7a27376a 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -1,13 +1,15 @@ -from __future__ import absolute_import import datetime import functools import json import logging +import os import socket +import six import sys import time +import urllib3 +import yaml -from kubernetes import client as k8s_client, config as k8s_config, watch as k8s_watch from urllib3 import Timeout from urllib3.exceptions import HTTPError from six.moves.http_client import HTTPException @@ -15,15 +17,245 @@ from threading import Condition, Lock, Thread from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState, TimelineHistory from ..exceptions import DCSError -from ..utils import deep_compare, keepalive_socket_options, Retry, RetryFailedError, tzutc, USER_AGENT +from ..utils import deep_compare, iter_response_objects, keepalive_socket_options,\ + Retry, RetryFailedError, tzutc, uri, USER_AGENT logger = logging.getLogger(__name__) +KUBE_CONFIG_DEFAULT_LOCATION = os.environ.get('KUBECONFIG', '~/.kube/config') +SERVICE_HOST_ENV_NAME = 'KUBERNETES_SERVICE_HOST' +SERVICE_PORT_ENV_NAME = 'KUBERNETES_SERVICE_PORT' +SERVICE_TOKEN_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/token' +SERVICE_CERT_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt' + class KubernetesError(DCSError): pass +# this function does the same mapping of snake_case => camelCase for > 97% of cases as autogenerated swagger code +def to_camel_case(value): + reserved = {'api', 'apiv3', 'cidr', 'cpu', 'csi', 'id', 'io', 'ip', 'ipc', 'pid', 'tls', 'uri', 'url', 'uuid'} + words = value.split('_') + return words[0] + ''.join(w.upper() if w in reserved else w.title() for w in words[1:]) + + +class K8sConfig(object): + + class ConfigException(Exception): + pass + + def __init__(self): + self.pool_config = {'maxsize': 10, 'num_pools': 10} # configuration for urllib3.PoolManager + self._make_headers() + + def _make_headers(self, token=None, **kwargs): + self._headers = urllib3.make_headers(user_agent=USER_AGENT, **kwargs) + if token: + self._headers['authorization'] = 'Bearer ' + token + + def load_incluster_config(self): + if SERVICE_HOST_ENV_NAME not in os.environ or SERVICE_PORT_ENV_NAME not in os.environ: + raise self.ConfigException('Service host/port is not set.') + if not os.environ[SERVICE_HOST_ENV_NAME] or not os.environ[SERVICE_PORT_ENV_NAME]: + raise self.ConfigException('Service host/port is set but empty.') + if not os.path.isfile(SERVICE_CERT_FILENAME): + raise self.ConfigException('Service certificate file does not exists.') + with open(SERVICE_CERT_FILENAME) as f: + if not f.read(): + raise self.ConfigException('Cert file exists but empty.') + if not os.path.isfile(SERVICE_TOKEN_FILENAME): + raise self.ConfigException('Service token file does not exists.') + with open(SERVICE_TOKEN_FILENAME) as f: + token = f.read() + if not token: + raise self.ConfigException('Token file exists but empty.') + self._make_headers(token=token) + self.pool_config['ca_certs'] = SERVICE_CERT_FILENAME + self._server = uri('https', (os.environ[SERVICE_HOST_ENV_NAME], os.environ[SERVICE_PORT_ENV_NAME])) + + @staticmethod + def _get_by_name(config, section, name): + for c in config[section + 's']: + if c['name'] == name: + return c[section] + + def load_kube_config(self, context=None): + with open(os.path.expanduser(KUBE_CONFIG_DEFAULT_LOCATION)) as f: + config = yaml.safe_load(f) + + context = self._get_by_name(config, 'context', context or config['current-context']) + cluster = self._get_by_name(config, 'cluster', context['cluster']) + user = self._get_by_name(config, 'user', context['user']) + + self._server = cluster['server'].rstrip('/') + if self._server.startswith('https'): + self.pool_config.update({v: user[k] for k, v in {'client-certificate': 'cert_file', + 'client-key': 'key_file'}.items() if k in user}) + if 'certificate-authority' in cluster: + self.pool_config['ca_certs'] = cluster['certificate-authority'] + self.pool_config['cert_reqs'] = 'CERT_NONE' if cluster.get('insecure-skip-tls-verify') else 'CERT_REQUIRED' + 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']))) + + @property + def server(self): + return self._server + + @property + def headers(self): + return self._headers.copy() + + +class K8sObject(object): + + def __init__(self, kwargs): + self._dict = {k: self._wrap(k, v) for k, v in kwargs.items()} + + def get(self, name, default=None): + return self._dict.get(name, default) + + def __getattr__(self, name): + return self.get(to_camel_case(name)) + + @classmethod + def _wrap(cls, parent, value): + 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, six.string_types) for v in value.values()) else cls(value) + elif isinstance(value, list): + return [cls._wrap(None, v) for v in value] + else: + return value + + def to_dict(self): + return self._dict + + def __repr__(self): + return json.dumps(self, indent=4, default=lambda o: o.to_dict()) + + +class K8sClient(object): + + class rest(object): + + class ApiException(Exception): + def __init__(self, status=None, reason=None, http_resp=None): + self.status = http_resp.status if http_resp else status + self.reason = http_resp.reason if http_resp else reason + self.body = http_resp.data if http_resp else None + self.headers = http_resp.getheaders() if http_resp else None + + def __str__(self): + error_message = "({0})\nReason: {1}\n".format(self.status, self.reason) + if self.headers: + error_message += "HTTP response headers: {0}\n".format(self.headers) + if self.body: + error_message += "HTTP response body: {0}\n".format(self.body) + return error_message + + class ApiClient(object): + + _API_URL_PREFIX = '/api/v1/namespaces/' + + def __init__(self): + self.pool_manager = urllib3.PoolManager(**k8s_config.pool_config) + self.set_read_timeout(10) + + def set_read_timeout(self, timeout): + self._read_timeout = timeout + + @staticmethod + def _handle_server_response(response, _preload_content): + if response.status not in range(200, 206): + raise k8s_client.rest.ApiException(http_resp=response) + return K8sObject(json.loads(response.data.decode('utf-8'))) if _preload_content else response + + @staticmethod + def _make_headers(headers): + ret = k8s_config.headers + ret.update(headers or {}) + return ret + + def request(self, method, path, timeout=None, **kwargs): + retries = 0 if timeout else 1 + if timeout: + if isinstance(timeout, six.integer_types + (float,)): + timeout = urllib3.Timeout(total=timeout) + elif isinstance(timeout, tuple) and len(timeout) == 2: + timeout = urllib3.Timeout(connect=timeout[0], read=timeout[1]) + else: + timeout = self._read_timeout / 2.0 + timeout = urllib3.Timeout(connect=max(1, timeout/2.0), total=timeout) + kwargs.update(retries=retries, timeout=timeout) + return self.pool_manager.request(method, k8s_config.server + path, **kwargs) + + def call_api(self, method, path, headers=None, body=None, + _preload_content=True, _request_timeout=None, **kwargs): + headers = self._make_headers(headers) + fields = {to_camel_case(k): v for k, v in kwargs.items()} # resource_version => resourceVersion + body = json.dumps(body, default=lambda o: o.to_dict()) if body is not None else None + + response = self.request(method, self._API_URL_PREFIX + path, headers=headers, fields=fields, + body=body, preload_content=_preload_content, timeout=_request_timeout) + + return self._handle_server_response(response, _preload_content) + + class CoreV1Api(object): + + def __init__(self, api_client=None): + self._api_client = api_client or k8s_client.ApiClient() + + def __getattr__(self, func): # `func` name pattern: (action)_namespaced_(kind) + action, kind = func.split('_namespaced_') # (read|list|create|patch|replace|delete|delete_collection) + kind = kind.replace('_', '') + ('s' * int(kind[-1] != 's')) # plural, single word + + def wrapper(*args, **kwargs): + method = {'read': 'GET', 'list': 'GET', 'create': 'POST', + 'replace': 'PUT'}.get(action, action.split('_')[0]).upper() + + if action == 'create' or len(args) == 1: # namespace is a first argument and name in not in arguments + path = '/'.join([args[0], kind]) + else: # name, namespace followed by optional body + path = '/'.join([args[1], kind, args[0]]) + + headers = {'Content-Type': 'application/strategic-merge-patch+json'} if action == 'patch' else {} + + if len(args) == 3: # name, namespace, body + body = args[2] + elif action == 'create': # namespace, body + body = args[1] + elif action == 'delete': # name, namespace + body = kwargs.pop('body', None) + else: + body = None + + return self._api_client.call_api(method, path, headers, body, **kwargs) + return wrapper + + class _K8sObjectTemplate(K8sObject): + """The template for objects which we create locally, e.g. k8s_client.V1ObjectMeta & co""" + def __init__(self, **kwargs): + self._dict = {to_camel_case(k): v for k, v in kwargs.items()} + + def __init__(self): + self.__cls_cache = {} + self.__cls_lock = Lock() + + def __getattr__(self, name): + with self.__cls_lock: + if name not in self.__cls_cache: + self.__cls_cache[name] = type(name, (self._K8sObjectTemplate,), {}) + return self.__cls_cache[name] + + +k8s_client = K8sClient() +k8s_config = K8sConfig() + + class KubernetesRetriableException(k8s_client.rest.ApiException): def __init__(self, orig): @@ -42,10 +274,8 @@ class KubernetesRetriableException(k8s_client.rest.ApiException): class CoreV1ApiProxy(object): def __init__(self, use_endpoints=False): - self._api = k8s_client.CoreV1Api() - self._api.api_client.user_agent = USER_AGENT - self._api.api_client.rest_client.pool_manager.connection_pool_kw['maxsize'] = 10 - self._request_timeout = None + self._api_client = k8s_client.ApiClient() + self._core_v1_api = k8s_client.CoreV1Api(self._api_client) self._use_endpoints = bool(use_endpoints) def configure_timeouts(self, loop_wait, retry_timeout, ttl): @@ -53,19 +283,17 @@ class CoreV1ApiProxy(object): # If we didn't received anything after the loop_wait + retry_timeout it is a time # 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.api_client.rest_client.pool_manager.connection_pool_kw['socket_options'] = \ + self._api_client.pool_manager.connection_pool_kw['socket_options'] = \ list(keepalive_socket_options(ttl, int(loop_wait + retry_timeout))) - self._request_timeout = (1, retry_timeout / 3.0) + self._api_client.set_read_timeout(retry_timeout) def __getattr__(self, func): if func.endswith('_kind'): func = func[:-4] + ('endpoints' if self._use_endpoints else 'config_map') def wrapper(*args, **kwargs): - if '_request_timeout' not in kwargs: - kwargs['_request_timeout'] = self._request_timeout try: - return getattr(self._api, func)(*args, **kwargs) + return getattr(self._core_v1_api, func)(*args, **kwargs) except k8s_client.rest.ApiException as e: if e.status in (500, 503, 504) or e.headers and 'retry-after' in e.headers: # XXX raise KubernetesRetriableException(e) @@ -97,7 +325,6 @@ class ObjectCache(Thread): def __init__(self, dcs, func, retry, condition, name=None): Thread.__init__(self) self.daemon = True - self._api_client = k8s_client.ApiClient() self._dcs = dcs self._func = func self._retry = retry @@ -155,8 +382,7 @@ class ObjectCache(Thread): response = self._watch(objects.metadata.resource_version) try: - for line in k8s_watch.watch.iter_resp_lines(response): - event = json.loads(line) + for event in iter_response_objects(response): obj = event['object'] if obj.get('code') == 410: break @@ -165,8 +391,7 @@ class ObjectCache(Thread): name = obj['metadata']['name'] if ev_type in ('ADDED', 'MODIFIED'): - obj = k8s_watch.watch.SimpleNamespace(data=json.dumps(obj)) - obj = self._api_client.deserialize(obj, return_type) + obj = K8sObject(obj) success, old_value = self.set(name, obj) if success: new_value = (obj.metadata.annotations or {}).get(self._annotations_map.get(name)) diff --git a/patroni/utils.py b/patroni/utils.py index 34c70034..c97d63a3 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -1,3 +1,4 @@ +import json.decoder as json_decoder import logging import os import platform @@ -375,6 +376,27 @@ def uri(proto, netloc, path='', user=None): return '{0}://{1}{2}{3}{4}'.format(proto, user, host, port, path) +def iter_response_objects(response): + prev = '' + decoder = json_decoder.JSONDecoder() + for chunk in response.read_chunked(decode_content=False): + if isinstance(chunk, bytes): + chunk = chunk.decode('utf-8') + chunk = prev + chunk + + length = len(chunk) + idx = json_decoder.WHITESPACE.match(chunk, 0).end() + while idx < length: + try: + message, idx = decoder.raw_decode(chunk, idx) + except ValueError: # malformed or incomplete JSON, unlikely to happen + break + else: + yield message + idx = json_decoder.WHITESPACE.match(chunk, idx).end() + prev = chunk[idx:] + + def is_standby_cluster(config): # Check whether or not provided configuration describes a standby cluster return isinstance(config, dict) and (config.get('host') or config.get('port') or config.get('restore_command')) diff --git a/requirements.txt b/requirements.txt index b4e07f7e..debdeaa0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,4 +10,3 @@ prettytable>=0.7 python-dateutil psutil>=2.0.0 cdiff -kubernetes>=2.0.0,<=10.0.1,!=4.0.*,!=5.0.* diff --git a/setup.py b/setup.py index 45a4b034..f536fa49 100644 --- a/setup.py +++ b/setup.py @@ -23,7 +23,7 @@ KEYWORDS = 'etcd governor patroni postgresql postgres ha haproxy confd' +\ ' zookeeper exhibitor consul streaming replication kubernetes k8s' EXTRAS_REQUIRE = {'aws': ['boto'], 'etcd': ['python-etcd'], 'consul': ['python-consul'], - 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': ['kubernetes']} + 'exhibitor': ['kazoo'], 'zookeeper': ['kazoo'], 'kubernetes': []} COVERAGE_XML = True COVERAGE_HTML = False @@ -165,7 +165,7 @@ def setup_package(version): continue extra = False for e, v in EXTRAS_REQUIRE.items(): - if r.startswith(v[0]): + if v and r.startswith(v[0]): EXTRAS_REQUIRE[e] = [r] extra = True if not extra: diff --git a/tests/__init__.py b/tests/__init__.py index 3cff8183..4958f152 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -22,6 +22,7 @@ class MockResponse(object): def __init__(self, status_code=200): self.status_code = status_code self.content = '{}' + self.reason = 'Not Found' @property def data(self): @@ -35,6 +36,10 @@ class MockResponse(object): def getheader(*args): return '' + @staticmethod + def getheaders(): + return {'content-type': 'json'} + def requests_get(url, **kwargs): members = '[{"id":14855829450254237642,"peerURLs":["http://localhost:2380","http://localhost:7001"],' +\ diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 03185348..7e8f0d1b 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -2,10 +2,12 @@ import json import time import unittest -from mock import Mock, patch -from patroni.dcs.kubernetes import Kubernetes, KubernetesError, k8s_client, RetryFailedError +from mock import Mock, mock_open, patch +from patroni.dcs.kubernetes import Kubernetes, KubernetesError, K8sConfig, K8sObject, RetryFailedError,\ + k8s_client, k8s_config, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME +from six.moves import builtins from threading import Thread -from . import SleepException +from . import MockResponse, SleepException def mock_list_namespaced_config_map(*args, **kwargs): @@ -51,28 +53,99 @@ def mock_namespaced_kind(*args, **kwargs): return mock +def mock_load_k8s_config(self, *args, **kwargs): + self._server = '' + + +class TestK8sConfig(unittest.TestCase): + + def test_load_incluster_config(self): + for env in ({}, {SERVICE_HOST_ENV_NAME: '', SERVICE_PORT_ENV_NAME: ''}): + with patch('os.environ', env): + self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config) + + with patch('os.environ', {SERVICE_HOST_ENV_NAME: 'a', SERVICE_PORT_ENV_NAME: '1'}),\ + patch('os.path.isfile', Mock(side_effect=[False, True, True, False, True, True, True, True])),\ + patch.object(builtins, 'open', Mock(side_effect=[ + mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')(), + mock_open()(), mock_open(read_data='a')(), mock_open(read_data='a')()])): + for _ in range(0, 4): + self.assertRaises(k8s_config.ConfigException, k8s_config.load_incluster_config) + k8s_config.load_incluster_config() + self.assertEqual(k8s_config.server, 'https://a:1') + + def test_load_kube_config(self): + config = { + "current-context": "local", + "contexts": [{"name": "local", "context": {"user": "local", "cluster": "local"}}], + "clusters": [{"name": "local", "cluster": {"server": "https://a:1/", "certificate-authority": "a"}}], + "users": [{"name": "local", "user": {"username": "a", "password": "b", "client-certificate": "c"}}] + } + with patch.object(builtins, 'open', mock_open(read_data=json.dumps(config))): + k8s_config.load_kube_config() + self.assertEqual(k8s_config.server, 'https://a:1') + self.assertEqual(k8s_config.pool_config, {'ca_certs': 'a', 'cert_file': 'c', 'cert_reqs': 'CERT_REQUIRED', + 'maxsize': 10, 'num_pools': 10}) + + config["users"][0]["user"]["token"] = "token" + with patch.object(builtins, 'open', mock_open(read_data=json.dumps(config))): + k8s_config.load_kube_config() + self.assertEqual(k8s_config.headers.get('authorization'), 'Bearer token') + + +class TestCoreV1Api(unittest.TestCase): + + @patch.object(K8sConfig, '_server', '', create=True) + def setUp(self): + self.a = k8s_client.CoreV1Api() + self.a._api_client.set_read_timeout(10) + self.a._api_client.pool_manager.request = Mock(return_value=MockResponse()) + + def test_create_namespaced_service(self): + self.assertEqual(str(self.a.create_namespaced_service('default', {}, _request_timeout=2)), '{}') + + def test_list_namespaced_endpoints(self): + self.a._api_client.pool_manager.request.return_value.content = '{"items": [1,2,3]}' + self.assertIsInstance(self.a.list_namespaced_endpoints('default'), K8sObject) + + def test_patch_namespaced_config_map(self): + self.assertEqual(str(self.a.patch_namespaced_config_map('foo', 'default', {}, _request_timeout=(1, 2))), '{}') + + def test_list_namespaced_pod(self): + self.a._api_client.pool_manager.request.return_value.status_code = 409 + self.a._api_client.pool_manager.request.return_value.content = 'foo' + try: + self.a.list_namespaced_pod('default', label_selector='foo=bar') + self.assertFail() + except k8s_client.rest.ApiException as e: + self.assertTrue('Reason: ' in str(e)) + + def test_delete_namespaced_pod(self): + self.assertEqual(str(self.a.delete_namespaced_pod('foo', 'default', _request_timeout=(1, 2), body={})), '{}') + + class BaseTestKubernetes(unittest.TestCase): @patch('socket.TCP_KEEPIDLE', 4, create=True) @patch('socket.TCP_KEEPINTVL', 5, create=True) @patch('socket.TCP_KEEPCNT', 6, create=True) - @patch('kubernetes.config.load_kube_config', Mock()) - @patch('kubernetes.client.api_client.ThreadPool', Mock(), create=True) @patch.object(Thread, 'start', Mock()) - @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod) - @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map) + @patch.object(K8sConfig, 'load_kube_config', mock_load_k8s_config) + @patch.object(K8sConfig, 'load_incluster_config', Mock(side_effect=k8s_config.ConfigException)) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod, create=True) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True) def setUp(self, config=None): config = config or {} config.update(ttl=30, scope='test', name='p-0', loop_wait=10, retry_timeout=10, labels={'f': 'b'}) self.k = Kubernetes(config) self.assertRaises(AttributeError, self.k._pods._build_cache) self.k._pods._is_ready = True - self.assertRaises(AttributeError, self.k._kinds._build_cache) + self.assertRaises(TypeError, self.k._kinds._build_cache) self.k._kinds._is_ready = True self.k.get_cluster() -@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind) +@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', mock_namespaced_kind, create=True) class TestKubernetesConfigMaps(BaseTestKubernetes): @patch('time.time', Mock(side_effect=[1, 10.9, 100])) @@ -95,13 +168,14 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.take_leader() def test_manual_failover(self): - with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', Mock(side_effect=RetryFailedError(''))): + with patch.object(k8s_client.CoreV1Api, 'patch_namespaced_config_map', + Mock(side_effect=RetryFailedError('')), create=True): self.k.manual_failover('foo', 'bar') def test_set_config_value(self): self.k.set_config_value('{}') - @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod') + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', create=True) def test_touch_member(self, mock_patch_namespaced_pod): mock_patch_namespaced_pod.return_value.metadata.resource_version = '10' self.k.touch_member({'role': 'replica'}) @@ -119,7 +193,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.cancel_initialization() @patch.object(k8s_client.CoreV1Api, 'delete_collection_namespaced_config_map', - Mock(side_effect=k8s_client.rest.ApiException(403, ''))) + Mock(side_effect=k8s_client.rest.ApiException(403, '')), create=True) def test_delete_cluster(self): self.k.delete_cluster() @@ -134,11 +208,11 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): class TestKubernetesEndpoints(BaseTestKubernetes): - @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_endpoints', mock_list_namespaced_endpoints, create=True) def setUp(self, config=None): super(TestKubernetesEndpoints, self).setUp({'use_endpoints': True, 'pod_ip': '10.0.0.0'}) - @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints') + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True) def test_update_leader(self, mock_patch_namespaced_endpoints): self.assertIsNotNone(self.k.update_leader('123')) args = mock_patch_namespaced_endpoints.call_args[0] @@ -148,11 +222,11 @@ class TestKubernetesEndpoints(BaseTestKubernetes): self.k._kinds._object_cache['test'].metadata.annotations['leader'] = 'p-1' self.assertFalse(self.k.update_leader('123')) - @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind) + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True) def test_update_leader_with_restricted_access(self): self.assertIsNotNone(self.k.update_leader('123', True)) - @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints') + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', create=True) def test__update_leader_with_retry(self, mock_patch): mock_patch.side_effect = k8s_client.rest.ApiException(502, '') self.assertFalse(self.k.update_leader('123')) @@ -168,14 +242,15 @@ class TestKubernetesEndpoints(BaseTestKubernetes): self.assertIsNotNone(self.k._update_leader_with_retry({}, '1', [])) @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', - Mock(side_effect=[k8s_client.rest.ApiException(500, ''), k8s_client.rest.ApiException(502, '')])) + Mock(side_effect=[k8s_client.rest.ApiException(500, ''), + k8s_client.rest.ApiException(502, '')]), create=True) def test_delete_sync_state(self): self.assertFalse(self.k.delete_sync_state()) - @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', mock_namespaced_kind) - @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', mock_namespaced_kind) + @patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', mock_namespaced_kind, create=True) + @patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', mock_namespaced_kind, create=True) @patch.object(k8s_client.CoreV1Api, 'create_namespaced_service', - Mock(side_effect=[True, False, k8s_client.rest.ApiException(500, '')])) + Mock(side_effect=[True, False, k8s_client.rest.ApiException(500, '')]), create=True) def test__create_config_service(self): self.assertIsNotNone(self.k.patch_or_create_config({'foo': 'bar'})) self.assertIsNotNone(self.k.patch_or_create_config({'foo': 'bar'})) @@ -184,7 +259,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes): class TestCacheBuilder(BaseTestKubernetes): - @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map) + @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True) @patch('patroni.dcs.kubernetes.ObjectCache._watch') def test__build_cache(self, mock_response): mock_response.return_value.read_chunked.return_value = [json.dumps( @@ -195,7 +270,7 @@ class TestCacheBuilder(BaseTestKubernetes): 'name': self.k.config_path, 'resourceVersion': '3'}}} ) + '\n' + json.dumps( {'type': 'MDIFIED', 'object': {'metadata': {'name': self.k.config_path}}} - ) + '\n' + json.dumps({'object': {'code': 410}}) + '\n').encode('utf-8')] + ) + '\n').encode('utf-8'), b'{"object":{', b'"code":410}}\n'] self.k._kinds._build_cache() @patch('patroni.dcs.kubernetes.logger.error', Mock(side_effect=SleepException))