test API calls

Pipfile

@@ -13,6 +13,7 @@ coveralls = "*"
 tox = "*"
 coverage = "*"
 twine = "*"
+responses = "*"
 
 [requires]
 python_version = "3.7"

pykube/http.py

@@ -30,9 +30,14 @@
 
 class KubernetesHTTPAdapter(requests.adapters.HTTPAdapter):
 
+    # _do_send: the actual send method of HTTPAdapter
+    # it can be overwritten in unit tests to mock the actual HTTP calls
+    _do_send = requests.adapters.HTTPAdapter.send
+
     def __init__(self, kube_config, **kwargs):
         self.kube_config = kube_config
-        super(KubernetesHTTPAdapter, self).__init__(**kwargs)
+
+        super().__init__(**kwargs)
 
     def _persist_credentials(self, config, token, expiry):
         user_name = config.contexts[config.current_context]["user"]
@@ -124,8 +129,7 @@ def send(self, request, **kwargs):
         elif "insecure-skip-tls-verify" in config.cluster:
             kwargs["verify"] = not config.cluster["insecure-skip-tls-verify"]
 
-        send = super(KubernetesHTTPAdapter, self).send
-        response = send(request, **kwargs)
+        response = self._do_send(request, **kwargs)
 
         _retry_status_codes = {http_client.UNAUTHORIZED}
 

tests/test_api.py

@@ -0,0 +1,71 @@
+import json
+import pytest
+import responses
+
+from pykube import KubeConfig, HTTPClient, Deployment
+
+
+@pytest.fixture
+def kubeconfig(tmpdir):
+    kubeconfig = tmpdir.join('kubeconfig')
+    kubeconfig.write('''
+apiVersion: v1
+clusters:
+- cluster: {server: 'https://localhost:9443'}
+  name: test
+contexts:
+- context: {cluster: test, user: test}
+  name: test
+current-context: test
+kind: Config
+preferences: {}
+users:
+- name: test
+  user: {token: testtoken}
+    ''')
+    return kubeconfig
+
+
+@pytest.fixture
+def requests_mock():
+    return responses.RequestsMock(target='pykube.http.KubernetesHTTPAdapter._do_send')
+
+
+def test_list_deployments(kubeconfig, monkeypatch, requests_mock):
+    config = KubeConfig.from_file(str(kubeconfig))
+    api = HTTPClient(config)
+
+    with requests_mock as rsps:
+        rsps.add(responses.GET, 'https://localhost:9443/apis/apps/v1/namespaces/default/deployments',
+                 json={'items': []})
+
+        assert list(Deployment.objects(api)) == []
+        assert len(rsps.calls) == 1
+        # ensure that we passed the token specified in kubeconfig..
+        assert rsps.calls[0].request.headers['Authorization'] == 'Bearer testtoken'
+
+
+def test_list_and_update_deployments(kubeconfig, monkeypatch, requests_mock):
+    config = KubeConfig.from_file(str(kubeconfig))
+    api = HTTPClient(config)
+
+    with requests_mock as rsps:
+        rsps.add(responses.GET, 'https://localhost:9443/apis/apps/v1/namespaces/default/deployments',
+                 json={'items': [{'metadata': {'name': 'deploy-1'}, 'spec': {'replicas': 3}}]})
+
+        deployments = list(Deployment.objects(api))
+        assert len(deployments) == 1
+        deploy = deployments[0]
+        assert deploy.name == 'deploy-1'
+        assert deploy.namespace == 'default'
+        assert deploy.replicas == 3
+
+        deploy.replicas = 2
+
+        rsps.add(responses.PATCH, 'https://localhost:9443/apis/apps/v1/namespaces/default/deployments/deploy-1',
+                 json={'items': [{'metadata': {'name': 'deploy-1'}, 'spec': {'replicas': 2}}]})
+
+        deploy.update()
+        assert len(rsps.calls) == 2
+
+        assert json.loads(rsps.calls[-1].request.body) == {"metadata": {"name": "deploy-1"}, "spec": {"replicas": 2}}

tests/test_http.py

@@ -19,7 +19,7 @@ def test_http(monkeypatch):
 
     mock_send = MagicMock()
     mock_send.side_effect = Exception('MOCK HTTP')
-    monkeypatch.setattr('requests.adapters.HTTPAdapter.send', mock_send)
+    monkeypatch.setattr('pykube.http.KubernetesHTTPAdapter._do_send', mock_send)
 
     with pytest.raises(Exception):
         session.get('http://localhost:9090/test')