1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
|
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import uuid
import yaml
from kubernetes.client import api_client
from kubernetes.client.api import apps_v1_api
from kubernetes.client.models import v1_delete_options
from kubernetes.e2e_test import base
class TestClientApps(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.config = base.get_e2e_configuration()
def test_create_deployment(self):
client = api_client.ApiClient(configuration=self.config)
api = apps_v1_api.AppsV1Api(client)
name = 'nginx-deployment-' + str(uuid.uuid4())
deployment = '''apiVersion: apps/v1
kind: Deployment
metadata:
name: %s
spec:
replicas: 3
selector:
matchLabels:
app: nginx
template:
metadata:
labels:
app: nginx
spec:
containers:
- name: nginx
image: nginx:1.15.4
ports:
- containerPort: 80
'''
api.create_namespaced_deployment(
body=yaml.safe_load(deployment % name),
namespace="default")
resp = api.read_namespaced_deployment(name, 'default')
self.assertIsNotNone(resp)
self.assertEqual(resp.metadata.name, name)
self.assertEqual(resp.spec.replicas, 3)
self.assertEqual(resp.spec.selector.match_labels['app'], 'nginx')
self.assertEqual(resp.spec.template.metadata.labels['app'], 'nginx')
self.assertEqual(resp.spec.template.spec.containers[0].name, 'nginx')
self.assertEqual(resp.spec.template.spec.containers[0].image, 'nginx:1.15.4')
self.assertEqual(resp.spec.template.spec.containers[0].ports[0].container_port, 80)
options = v1_delete_options.V1DeleteOptions()
api.delete_namespaced_deployment(name, 'default', body=options)
def test_create_daemonset(self):
client = api_client.ApiClient(configuration=self.config)
api = apps_v1_api.AppsV1Api(client)
name = 'nginx-app-' + str(uuid.uuid4())
daemonset = {
'apiVersion': 'apps/v1',
'kind': 'DaemonSet',
'metadata': {
'labels': {'app': 'nginx'},
'name': '%s' % name,
},
'spec': {
'selector': {
'matchLabels': {'app': 'nginx'},
},
'template': {
'metadata': {
'labels': {'app': 'nginx'},
'name': name},
'spec': {
'containers': [
{'name': 'nginx-app',
'image': 'nginx:1.15.4'},
],
},
},
'updateStrategy': {
'type': 'RollingUpdate',
},
}
}
api.create_namespaced_daemon_set('default', body=daemonset)
resp = api.read_namespaced_daemon_set(name, 'default')
self.assertIsNotNone(resp)
self.assertEqual(resp.metadata.name, name)
self.assertEqual(resp.spec.selector.match_labels['app'], 'nginx')
self.assertEqual(resp.spec.template.metadata.labels['app'], 'nginx')
self.assertEqual(resp.spec.template.spec.containers[0].name, 'nginx-app')
self.assertEqual(resp.spec.template.spec.containers[0].image, 'nginx:1.15.4')
options = v1_delete_options.V1DeleteOptions()
api.delete_namespaced_daemon_set(name, 'default', body=options)
|
