More tests

Makefile

@@ -10,6 +10,7 @@ clean:
 test:
 	pipenv run flake8
 	pipenv run coverage run --source=pykube -m py.test
+	pipenv run coverage html
 	pipenv run coverage report
 
 apidocs:

pykube/config.py

@@ -12,7 +12,7 @@
 from pykube import exceptions
 
 
-class KubeConfig(object):
+class KubeConfig:
     """
     Main configuration class.
     """
@@ -216,7 +216,7 @@ def reload(self):
             delattr(self, "_clusters")
 
 
-class BytesOrFile(object):
+class BytesOrFile:
     """
     Implements the same interface for files and byte input.
     """

pykube/http.py

@@ -145,7 +145,7 @@ def send(self, request, **kwargs):
         return response
 
 
-class HTTPClient(object):
+class HTTPClient:
     """
     Client for interfacing with the Kubernetes API.
     """
@@ -187,7 +187,7 @@ def version(self):
         return (data["major"], data["minor"])
 
     def resource_list(self, api_version):
-        cached_attr = "_cached_resource_list"
+        cached_attr = f'_cached_resource_list_{api_version}'
         if not hasattr(self, cached_attr):
             r = self.get(version=api_version)
             r.raise_for_status()

pykube/mixins.py

@@ -1,7 +1,7 @@
 import time
 
 
-class ReplicatedMixin(object):
+class ReplicatedMixin:
 
     scalable_attr = "replicas"
 
@@ -14,7 +14,7 @@ def replicas(self, value):
         self.obj["spec"]["replicas"] = value
 
 
-class ScalableMixin(object):
+class ScalableMixin:
 
     @property
     def scalable(self):

pykube/objects.py

@@ -10,7 +10,7 @@
 from .utils import obj_merge
 
 
-class ObjectManager(object):
+class ObjectManager:
     def __call__(self, api, namespace=None):
         if namespace is None and NamespacedAPIObject in getmro(self.api_obj_class):
             namespace = api.config.namespace
@@ -22,7 +22,7 @@ def __get__(self, obj, api_obj_class):
         return self
 
 
-class APIObject(object):
+class APIObject:
     '''
     Baseclass for all Kubernetes API objects
     '''
@@ -180,6 +180,7 @@ def object_factory(api, api_version, kind):
     It is planned to fix this, but in the mean time pass it as you would normally.
     """
     resource_list = api.resource_list(api_version)
+    print(resource_list)
     resource = next((resource for resource in resource_list["resources"] if resource["kind"] == kind), None)
     base = NamespacedAPIObject if resource["namespaced"] else APIObject
     return type(kind, (base,), {

pykube/query.py

@@ -12,7 +12,7 @@
 now = object()
 
 
-class BaseQuery(object):
+class BaseQuery:
 
     def __init__(self, api, api_obj_class, namespace=None):
         self.api = api

tests/test_api.py

@@ -4,7 +4,7 @@
 import responses
 
 import pykube
-from pykube import KubeConfig, HTTPClient, Deployment
+from pykube import KubeConfig, HTTPClient, Deployment, ObjectDoesNotExist
 
 
 @pytest.fixture
@@ -191,3 +191,57 @@ def test_list_and_update_deployments(api, requests_mock):
         assert len(rsps.calls) == 2
 
         assert json.loads(rsps.calls[-1].request.body) == {"metadata": {"name": "deploy-1"}, "spec": {"replicas": 2}}
+
+
+def test_pod_exists(api, requests_mock):
+    with requests_mock as rsps:
+        obj = {
+            "apiVersion": "v1",
+            "kind": "Pod",
+            "metadata": {
+                "name": "my-pod"
+            },
+            "spec": {
+                "containers": []
+            }
+        }
+        pod = pykube.Pod(api, obj)
+
+        rsps.add(responses.GET, 'https://localhost:9443/api/v1/namespaces/default/pods/my-pod',
+                 status=404)
+        assert not pod.exists()
+
+        with pytest.raises(ObjectDoesNotExist):
+            pod.exists(ensure=True)
+
+        rsps.replace(responses.GET, 'https://localhost:9443/api/v1/namespaces/default/pods/my-pod',
+                     json={})
+        assert pod.exists()
+
+
+def test_reload(api, requests_mock):
+    with requests_mock as rsps:
+        rsps.add(responses.GET, 'https://localhost:9443/api/v1/namespaces/default/pods/my-pod',
+                 json={'metadata': {'name': 'my-pod', 'labels': {'a': 'foo'}}})
+        pod = pykube.Pod.objects(api).get_by_name('my-pod')
+        assert pod.labels['a'] == 'foo'
+
+        rsps.replace(responses.GET, 'https://localhost:9443/api/v1/namespaces/default/pods/my-pod',
+                     json={'metadata': {'name': 'my-pod', 'labels': {'a': 'bar'}}})
+        pod.reload()
+        assert pod.labels['a'] == 'bar'
+
+
+def test_resource_list(api, requests_mock):
+    with requests_mock as rsps:
+        data1 = {'resources': [{'kind': 'Pod', 'name': 'pods'}]}
+        rsps.add(responses.GET, 'https://localhost:9443/api/v1/',
+                 json=data1)
+        resource_list = api.resource_list('v1')
+        assert resource_list == data1
+
+        data2 = {'resources': [{'kind': 'ExampleObject', 'name': 'exampleobjects'}]}
+        rsps.add(responses.GET, 'https://localhost:9443/apis/example.org/v1/',
+                 json=data2)
+        resource_list = api.resource_list('example.org/v1')
+        assert resource_list == data2

tests/test_config_with_insecure_skip_tls_verify.yaml

@@ -0,0 +1,17 @@
+current-context: thecluster
+clusters:
+    - name: thecluster
+      cluster: 
+        insecure-skip-tls-verify: true
+users:
+    - name: admin
+      user:
+        username: adm
+        password: somepassword
+contexts:
+    - name: thecluster
+      context:
+        cluster: thecluster
+        user: admin
+    - name: second
+      context: secondcontext

tests/test_http.py

@@ -12,6 +12,7 @@
 from pykube.config import KubeConfig
 
 GOOD_CONFIG_FILE_PATH = os.path.sep.join(["tests", "test_config_with_context.yaml"])
+CONFIG_WITH_INSECURE_SKIP_TLS_VERIFY = os.path.sep.join(["tests", "test_config_with_insecure_skip_tls_verify.yaml"])
 
 
 def test_http(monkeypatch):
@@ -30,3 +31,19 @@ def test_http(monkeypatch):
     assert mock_send.call_args[0][0].headers['User-Agent'] == f'pykube-ng/{__version__}'
     # check that the default HTTP timeout was set
     assert mock_send.call_args[1]['timeout'] == DEFAULT_HTTP_TIMEOUT
+
+
+def test_http_insecure_skip_tls_verify(monkeypatch):
+    cfg = KubeConfig.from_file(CONFIG_WITH_INSECURE_SKIP_TLS_VERIFY)
+    api = HTTPClient(cfg)
+
+    mock_send = MagicMock()
+    mock_send.side_effect = Exception('MOCK HTTP')
+    monkeypatch.setattr('pykube.http.KubernetesHTTPAdapter._do_send', mock_send)
+
+    with pytest.raises(Exception):
+        api.get(url='test')
+
+    mock_send.assert_called_once()
+    # check that SSL is not verified
+    assert not mock_send.call_args[1]['verify']

tests/test_objects.py

@@ -1,10 +1,24 @@
+from unittest.mock import MagicMock
+
 import pykube
 
+from pykube.objects import Pod, NamespacedAPIObject
+
 
 def test_api_object():
-    pod = pykube.Pod(None, {'metadata': {'name': 'myname'}})
+    pod = Pod(None, {'metadata': {'name': 'myname'}})
     assert repr(pod) == '<Pod myname>'
     assert str(pod) == 'myname'
     assert pod.metadata == {'name': 'myname'}
     assert pod.labels == {}
     assert pod.annotations == {}
+
+
+def test_object_factory():
+    api = MagicMock()
+    api.resource_list.return_value = {'resources': [{'kind': 'ExampleObject', 'namespaced': True, 'name': 'exampleobjects'}]}
+    ExampleObject = pykube.object_factory(api, 'example.org/v1', 'ExampleObject')
+    assert ExampleObject.kind == 'ExampleObject'
+    assert ExampleObject.endpoint == 'exampleobjects'
+    assert ExampleObject.version == 'example.org/v1'
+    assert NamespacedAPIObject in ExampleObject.mro()

tests/test_query.py

@@ -17,6 +17,22 @@ def test_get(api):
         Query(api, Pod).get(namespace='myns')
 
 
+def test_get_one_object(api):
+    response = MagicMock()
+    response.json.return_value = {'items': [{'metadata': {'name': 'pod1'}}]}
+    api.get.return_value = response
+    pod = Query(api, Pod).get(namespace='myns')
+    assert pod.name == 'pod1'
+
+
+def test_get_more_than_one_object(api):
+    response = MagicMock()
+    response.json.return_value = {'items': [{'metadata': {'name': 'pod1'}}, {'metadata': {'name': 'pod2'}}]}
+    api.get.return_value = response
+    with pytest.raises(ValueError):
+        Query(api, Pod).get(namespace='myns')
+
+
 def test_filter_by_namespace(api):
     Query(api, Pod).filter(namespace='myns').execute()
     api.get.assert_called_once_with(namespace='myns', url='pods', version='v1')
@@ -40,3 +56,13 @@ def test_filter_by_labels_in(api):
 def test_filter_by_labels_notin(api):
     Query(api, Pod).filter(selector={'application__notin': ['foo', 'bar']}).execute()
     api.get.assert_called_once_with(url='pods?labelSelector=application+notin+%28bar%2Cfoo%29', version='v1')
+
+
+def test_filter_invalid_selector(api):
+    with pytest.raises(ValueError):
+        Query(api, Pod).filter(selector={'application__x': 'foo'}).execute()
+
+
+def test_filter_selector_string(api):
+    Query(api, Pod).filter(selector='application=foo').execute()
+    api.get.assert_called_once_with(url='pods?labelSelector=application%3Dfoo', version='v1')