x/python-k8sclient
Code Issues Proposed changes

Add More tests from Magnum

Browse Source 
Change-Id: Id2ab6367e8fef23f7c02097601d6699b4e97267e
This commit is contained in:
Davanum Srinivas 2016-04-24 06:52:09 -04:00
parent 6d1d7ea3d9
commit a3a967ae38
2 changed files with 96 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -22,6 +22,7 @@ lib64
pip-log.txt
# Unit test / coverage reports
.cache
.coverage
.tox
cover

99
k8sclient/tests/test_k8sclient.py
View File

@ -24,6 +24,7 @@ and then run this test.
from testtools.testcase import unittest
import urllib3
import uuid
from k8sclient.client import api_client
from k8sclient.client.apis import apiv_api
@ -41,12 +42,102 @@ def _is_k8s_running():
class TestK8sclient(base.TestCase):
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_list_nodes_and_endpoints(self):
def test_list_endpoints(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = apiv_api.ApivApi(client)
pod = api.list_pod()
self.assertEquals(3, len(pod.items))
endpoints = api.list_endpoints()
self.assertEquals(1, len(endpoints.items))
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_pod_apis(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = apiv_api.ApivApi(client)
name = 'test-' + str(uuid.uuid4())
pod_manifest = {'apiVersion': 'v1',
'kind': 'Pod',
'metadata': {'color': 'blue', 'name': name},
'spec': {'containers': [{'image': 'dockerfile/redis',
'name': 'redis'}]}}
resp = api.create_namespaced_pod(body=pod_manifest,
namespace='default')
self.assertEqual(name, resp.metadata.name)
self.assertTrue(resp.status.phase)
resp = api.read_namespaced_pod(name=name,
namespace='default')
self.assertEqual(name, resp.metadata.name)
self.assertTrue(resp.status.phase)
number_of_pods = len(api.list_pod().items)
self.assertTrue(number_of_pods > 0)
resp = api.delete_namespaced_pod(name=name, body={},
namespace='default')
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_service_apis(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = apiv_api.ApivApi(client)
service_manifest = {'apiVersion': 'v1',
'kind': 'Service',
'metadata': {'labels': {'name': 'frontend'},
'name': 'frontend',
'resourceversion': 'v1'},
'spec': {'ports': [{'port': 80,
'protocol': 'TCP',
'targetPort': 80}],
'selector': {'name': 'frontend'}}}
resp = api.create_namespaced_service(body=service_manifest,
namespace='default')
self.assertEqual('frontend', resp.metadata.name)
self.assertTrue(resp.status)
resp = api.read_namespaced_service(name='frontend',
namespace='default')
self.assertEqual('frontend', resp.metadata.name)
self.assertTrue(resp.status)
resp = api.delete_namespaced_service(name='frontend',
namespace='default')
@unittest.skipUnless(
_is_k8s_running(), "Kubernetes is not available")
def test_replication_controller_apis(self):
client = api_client.ApiClient('http://127.0.0.1:8080/')
api = apiv_api.ApivApi(client)
rc_manifest = {
'apiVersion': 'v1',
'kind': 'ReplicationController',
'metadata': {'labels': {'name': 'frontend'},
'name': 'frontend'},
'spec': {'replicas': 2,
'selector': {'name': 'frontend'},
'template': {'metadata': {
'labels': {'name': 'frontend'}},
'spec': {'containers': [{
'image': 'nginx',
'name': 'nginx',
'ports': [{'containerPort': 80,
'protocol': 'TCP'}]}]}}}}
resp = api.create_namespaced_replication_controller(
body=rc_manifest, namespace='default')
self.assertEqual('frontend', resp.metadata.name)
self.assertEqual(2, resp.spec.replicas)
resp = api.read_namespaced_replication_controller(
name='frontend', namespace='default')
self.assertEqual('frontend', resp.metadata.name)
self.assertEqual(2, resp.spec.replicas)
resp = api.delete_namespaced_replication_controller(
name='frontend', body={}, namespace='default')