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
|
# -*- 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
from kubernetes import watch
from kubernetes.client import api_client
from kubernetes.client.api import core_v1_api
from kubernetes.e2e_test import base
def short_uuid():
id = str(uuid.uuid4())
return id[-12:]
def config_map_with_value(name, value):
return {
"apiVersion": "v1",
"kind": "ConfigMap",
"metadata": {
"name": name,
"labels": {"e2e-tests": "true"},
},
"data": {
"key": value,
"config": "dummy",
}
}
class TestClient(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.config = base.get_e2e_configuration()
def test_watch_configmaps(self):
client = api_client.ApiClient(configuration=self.config)
api = core_v1_api.CoreV1Api(client)
# create a configmap
name_a = 'configmap-a-' + short_uuid()
configmap_a = config_map_with_value(name_a, "a")
api.create_namespaced_config_map(
body=configmap_a, namespace='default')
# list all configmaps and extract the resource version
resp = api.list_namespaced_config_map('default', label_selector="e2e-tests=true")
rv = resp.metadata.resource_version
# create another configmap
name_b = 'configmap-b-' + short_uuid()
configmap_b = config_map_with_value(name_b, "b")
api.create_namespaced_config_map(
body=configmap_b, namespace='default')
# patch configmap b
configmap_b['data']['config'] = "{}"
api.patch_namespaced_config_map(
name=name_b, namespace='default', body=configmap_b)
# delete all configmaps
api.delete_collection_namespaced_config_map(
namespace='default', label_selector="e2e-tests=true")
w = watch.Watch()
# expect to observe all events happened after the initial LIST
expect = ['ADDED', 'MODIFIED', 'DELETED', 'DELETED']
i = 0
# start watching with the resource version we got from the LIST
for event in w.stream(api.list_namespaced_config_map,
namespace='default',
resource_version=rv,
timeout_seconds=5,
label_selector="e2e-tests=true"):
self.assertEqual(event['type'], expect[i])
# Kubernetes doesn't guarantee the order of the two objects
# being deleted
if i < 2:
self.assertEqual(event['object'].metadata.name, name_b)
i = i + 1
self.assertEqual(i, 4)
|
