1 # ============LICENSE_START=======================================================
3 # ================================================================================
4 # Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
5 # ================================================================================
6 # Licensed under the Apache License, Version 2.0 (the "License");
7 # you may not use this file except in compliance with the License.
8 # You may obtain a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS,
14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
17 # ============LICENSE_END=========================================================
21 def test_parse_interval():
22 from k8sclient.k8sclient import _parse_interval
24 good_intervals = [{"in": input, "ex": expected}
25 for (input, expected) in [
31 ("24h", 24 * 60 * 60),
35 (1234567890123456789012345678901234567890L,1234567890123456789012345678901234567890L),
36 ("1234567890123456789012345678901234567890",1234567890123456789012345678901234567890L),
37 ("1234567890123456789012345678901234567890s",1234567890123456789012345678901234567890L),
39 ("00000000000000000000000000000000005m", 5 * 60)
64 "i want an interval of 30s",
81 for test_case in good_intervals:
82 assert _parse_interval(test_case["in"]) == test_case["ex"]
84 for interval in bad_intervals:
85 with pytest.raises(ValueError):
86 _parse_interval(interval)
88 def test_parse_ports():
89 from k8sclient.k8sclient import _parse_ports
91 good_ports = [{"in": input, "ex": expected}
92 for (input, expected) in [
93 ("9101:0", (9101, 0, "TCP")),
94 ("9101/TCP:0", (9101, 0, "TCP")),
95 ("9101/tcp:0", (9101, 0, "TCP")),
96 ("9101/UDP:0", (9101, 0, "UDP")),
97 ("9101/udp:0", (9101, 0, "UDP")),
98 ("9101:31043", (9101, 31043, "TCP")),
99 ("9101/TCP:31043", (9101, 31043, "TCP")),
100 ("9101/tcp:31043", (9101, 31043, "TCP")),
101 ("9101/UDP:31043", (9101, 31043, "UDP")),
102 ("9101/udp:31043", (9101, 31043, "UDP"))
126 expected_port_map = {
130 (9661,"TCP") : 19661,
131 (9661,"UDP") : 19661,
135 for test_case in good_ports:
136 container_ports, port_map = _parse_ports([test_case["in"]])
137 (cport, hport, proto) = test_case["ex"]
138 assert container_ports == [(cport, proto)]
139 assert port_map == {(cport, proto) : hport}
141 for port in bad_ports:
142 with pytest.raises(ValueError):
145 container_ports, port_map = _parse_ports(port_list)
146 assert port_map == expected_port_map
148 def test_create_container():
149 from k8sclient.k8sclient import _create_container_object
150 from kubernetes import client
152 container = _create_container_object("c1","nginx",False, container_ports=[(80, "TCP"), (53, "UDP")])
154 assert container.ports[0].container_port == 80 and container.ports[0].protocol == "TCP"
155 assert container.ports[1].container_port == 53 and container.ports[1].protocol == "UDP"
157 def test_create_probe():
158 from k8sclient.k8sclient import _create_probe
159 from kubernetes import client
162 {"type" : "http", "endpoint" : "/example/health"}
166 {"type" : "docker", "script": "/opt/app/health_check.sh"}
169 for hc in http_checks:
170 probe = _create_probe(hc, 13131)
171 assert probe.http_get.path == hc["endpoint"]
172 assert probe.http_get.scheme == hc["type"].upper()
174 for hc in script_checks:
175 probe = _create_probe(hc, 13131)
176 assert probe._exec.command[0] == hc["script"]
178 def test_deploy(monkeypatch):
179 import k8sclient.k8sclient
180 from kubernetes import client
182 # We need to patch the kubernetes 'client' module
183 # Awkward because of the way it requires a function call
184 # to get an API object
185 core = client.CoreV1Api()
186 ext = client.ExtensionsV1beta1Api()
188 def pseudo_deploy(namespace, dep):
191 def pseudo_service(namespace, svc):
194 # patched_core returns a CoreV1Api object with the
195 # create_namespaced_service method stubbed out so that there
196 # is no attempt to call the k8s API server
198 monkeypatch.setattr(core, "create_namespaced_service", pseudo_service)
201 # patched_ext returns an ExtensionsV1beta1Api object with the
202 # create_namespaced_deployment method stubbed out so that there
203 # is no attempt to call the k8s API server
205 monkeypatch.setattr(ext,"create_namespaced_deployment", pseudo_deploy)
208 def pseudo_configure():
211 monkeypatch.setattr(k8sclient.k8sclient,"_configure_api", pseudo_configure)
212 monkeypatch.setattr(client, "CoreV1Api", patched_core)
213 monkeypatch.setattr(client,"ExtensionsV1beta1Api", patched_ext)
216 "image_pull_secrets" : ["secret0", "secret1"],
218 "log_path": "/var/log/onap",
219 "data_path": "/usr/share/filebeat/data",
220 "config_path": "/usr/share/filebeat/filebeat.yml",
221 "config_subpath": "filebeat.yml",
222 "image" : "filebeat-repo/filebeat:latest",
223 "config_map" : "dcae-filebeat-configmap"
226 "cert_path": "/opt/certs",
227 "image": "tlsrepo/tls-init-container:1.2.3"
244 {"host":{"path": "/path/on/host"}, "container":{"bind":"/path/on/container","mode":"rw"}}
246 "ports": ["80:0", "443:0"],
247 "env": {"name0": "value0", "name1": "value1"},
248 "log_info": {"log_directory": "/path/to/container/log/directory"},
249 "tls_info": {"use_tls": True, "cert_directory": "/path/to/container/cert/directory" },
250 "readiness": {"type": "http", "endpoint" : "/ready"}
252 dep, deployment_description = k8sclient.k8sclient.deploy("k8stest","testcomponent","example.com/testcomponent:1.4.3",1,False, k8s_test_config, resources, **kwargs)
254 assert deployment_description["deployment"] == "dep-testcomponent"
255 assert deployment_description["namespace"] == "k8stest"
256 assert deployment_description["services"][0] == "testcomponent"
258 # For unit test purposes, we want to make sure that the deployment object
259 # we're passing to the k8s API is correct
260 app_container = dep.spec.template.spec.containers[0]
261 assert app_container.image == "example.com/testcomponent:1.4.3"
262 assert app_container.image_pull_policy == "IfNotPresent"
263 assert len(app_container.ports) == 2
264 assert app_container.ports[0].container_port == 80
265 assert app_container.ports[1].container_port == 443
266 assert app_container.readiness_probe.http_get.path == "/ready"
267 assert app_container.readiness_probe.http_get.scheme == "HTTP"
268 assert len(app_container.volume_mounts) == 3
269 assert app_container.volume_mounts[0].mount_path == "/path/on/container"
270 assert app_container.volume_mounts[1].mount_path == "/path/to/container/log/directory"
271 assert app_container.volume_mounts[2].mount_path == "/path/to/container/cert/directory"
273 # Needs to be correctly labeled so that the Service can find it
274 assert dep.spec.template.metadata.labels["app"] == "testcomponent"