kubernetes-client/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add e2e for deployment and daemonsets

Browse Source
This commit is contained in:
Davanum Srinivas 2017-01-25 16:44:55 -05:00
parent cf5d796cf4
commit 03731f1003
1 changed files with 86 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
kubernetes/e2e_test/test_client.py
View File

@ -25,9 +25,11 @@ and then run this test
import unittest
import urllib3
import uuid
import yaml
from kubernetes.client import api_client
from kubernetes.client.apis import core_v1_api
from kubernetes.client.apis import extensions_v1beta1_api
def _is_k8s_running():
@ -39,6 +41,26 @@ def _is_k8s_running():
class TestClient(unittest.TestCase):
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_read_namespaces(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = core_v1_api.CoreV1Api(client)
expected_namespaces = ('default', 'kube-system')
for ns in expected_namespaces:
api.read_namespace(name=ns)
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_read_services(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = core_v1_api.CoreV1Api(client)
expected_services = ('kubernetes',)
for service in expected_services:
api.read_namespaced_service(service, 'default')
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_list_endpoints(self):
@ -48,6 +70,35 @@ class TestClient(unittest.TestCase):
endpoints = api.list_endpoints_for_all_namespaces()
self.assertTrue(len(endpoints.items) > 0)
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_create_deployment(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = extensions_v1beta1_api.ExtensionsV1beta1Api(client)
name = 'nginx-deployment-' + str(uuid.uuid4())
deployment = '''apiVersion: extensions/v1beta1
kind: Deployment
metadata:
name: %s
spec:
replicas: 3
template:
metadata:
labels:
app: nginx
spec:
containers:
- name: nginx
image: nginx:1.7.9
ports:
- containerPort: 80
'''
resp = api.create_namespaced_deployment(
body=yaml.load(deployment % name),
namespace="default")
resp = api.read_namespaced_deployment(name, 'default')
self.assertIsNotNone(resp)
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_pod_apis(self):
@ -153,7 +204,6 @@ class TestClient(unittest.TestCase):
resp = api.delete_namespaced_replication_controller(
name=name, body={}, namespace='default')
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_configmap_apis(self):
@ -201,4 +251,38 @@ class TestClient(unittest.TestCase):
for item in api.list_node().items:
node = api.read_node(name=item.metadata.name)
self.assertTrue(len(node.metadata.labels) > 0)
self.assertTrue(isinstance(node.metadata.labels, dict))
self.assertTrue(isinstance(node.metadata.labels, dict))
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_create_daemonset(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = extensions_v1beta1_api.ExtensionsV1beta1Api(client)
name = 'nginx-app-' + str(uuid.uuid4())
daemonset = {
'apiVersion': 'extensions/v1beta1',
'kind': 'DaemonSet',
'metadata': {
'labels': {'app': 'nginx'},
'name': '%s' % name,
},
'spec': {
'template': {
'metadata': {
'labels': {'app': 'nginx'},
'name': name},
'spec': {
'containers': [
{'name': 'nginx-app',
'image': 'nginx:1.10'},
],
},
},
'updateStrategy': {
'type': 'RollingUpdate',
},
}
}
resp = api.create_namespaced_daemon_set('default', body=daemonset)
resp = api.read_namespaced_daemon_set(name, 'default')
self.assertIsNotNone(resp)