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
|
#!/usr/bin/env python
#
# Copyright (c) 2018 All rights reserved
# This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
"""
Define the parent for Kubernetes testing.
"""
from __future__ import division
import logging
import os
import subprocess
import time
from functest.core import testcase
LOGGER = logging.getLogger(__name__)
class K8sTesting(testcase.TestCase):
"""Kubernetes test runner"""
def __init__(self, **kwargs):
super(K8sTesting, self).__init__(**kwargs)
self.cmd = []
self.result = 0
self.start_time = 0
self.stop_time = 0
def run_kubetest(self):
"""Run the test suites"""
cmd_line = self.cmd
LOGGER.info("Starting k8s test: '%s'.", cmd_line)
process = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
remark = []
lines = process.stdout.readlines()
for i in range(len(lines) - 1, -1, -1):
new_line = str(lines[i])
if 'SUCCESS!' in new_line or 'FAIL!' in new_line:
remark = new_line.replace('--', '|').split('|')
break
if remark and 'SUCCESS!' in remark[0]:
self.result = 100
def run(self):
if not os.path.isfile(os.getenv('KUBECONFIG')):
LOGGER.error("Cannot run k8s testcases. Config file not found ")
return self.EX_RUN_ERROR
self.start_time = time.time()
try:
self.run_kubetest()
res = self.EX_OK
except Exception as ex: # pylint: disable=broad-except
LOGGER.error("Error with running %s", str(ex))
res = self.EX_RUN_ERROR
self.stop_time = time.time()
return res
def check_envs(self): # pylint: disable=no-self-use
"""Check if required environment variables are set"""
try:
assert 'DEPLOY_SCENARIO' in os.environ
assert 'KUBECONFIG' in os.environ
assert 'KUBE_MASTER' in os.environ
assert 'KUBE_MASTER_IP' in os.environ
assert 'KUBERNETES_PROVIDER' in os.environ
assert 'KUBE_MASTER_URL' in os.environ
except Exception as ex:
raise Exception("Cannot run k8s testcases. "
"Please check env var: %s" % str(ex))
class K8sSmokeTest(K8sTesting):
"""Kubernetes smoke test suite"""
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs.get("case_name", 'k8s_smoke')
super(K8sSmokeTest, self).__init__(**kwargs)
self.check_envs()
self.cmd = ['/src/k8s.io/kubernetes/cluster/test-smoke.sh', '--host',
os.getenv('KUBE_MASTER_URL')]
|
