3 # Copyright (c) 2020 Orange and others.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
11 The CNF Conformance program enables interoperability of Cloud native Network
12 Functions (CNFs) from multiple vendors running on top of Kubernetes supplied by
13 different vendors [1].
14 [1] https://github.com/cncf/cnf-testsuite
17 from __future__ import division
28 from kubernetes import client
29 from kubernetes import config
31 from xtesting.core import testcase
34 class CNFConformance(testcase.TestCase):
35 # pylint: disable=too-many-instance-attributes
36 """ Implement CNF Conformance driver.
38 https://hackmd.io/@vulk/SkY54QnsU
41 src_dir = '/src/cnf-testsuite'
42 bin_dir = '/usr/local/bin'
45 __logger = logging.getLogger(__name__)
47 def __init__(self, **kwargs):
48 super().__init__(**kwargs)
49 config.load_kube_config()
50 self.corev1 = client.CoreV1Api()
51 self.output_log_name = 'functest-kubernetes.log'
52 self.output_debug_log_name = 'functest-kubernetes.debug.log'
54 def check_requirements(self):
55 """Check if cnf-testsuite is in $PATH"""
56 if not os.path.exists(os.path.join(self.bin_dir, 'cnf-testsuite')):
57 self.__logger.warning(
58 "cnf-testsuite is not compiled for arm and arm64 for the "
60 self.is_skipped = True
63 """Implement initialization and pre-reqs steps"""
64 if os.path.exists(os.path.join(self.src_dir, "results")):
65 shutil.rmtree(os.path.join(self.src_dir, "results"))
66 api_response = self.corev1.list_namespace()
67 for namespace in ["cnf-testsuite", "default", "litmus"]:
68 for ns in api_response.items:
69 if ns.metadata.name == namespace:
70 self.corev1.patch_namespace(
72 client.V1Namespace(metadata=client.V1ObjectMeta(
74 "pod-security.kubernetes.io/enforce":
77 "patch_namespace: %s", namespace)
80 self.corev1.create_namespace(
81 client.V1Namespace(metadata=client.V1ObjectMeta(
82 name=namespace, labels={
83 "pod-security.kubernetes.io/enforce":
86 "create_namespace: %s", namespace)
87 os.chdir(self.src_dir)
88 cmd = ['cnf-testsuite', 'setup', '-l', 'debug']
90 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
91 except subprocess.CalledProcessError as exc:
92 self.__logger.exception(
93 "Cannot run %s:\n%s", ' '.join(exc.cmd),
94 exc.output.decode("utf-8"))
97 self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
98 cmd = ['cnf-testsuite', 'cnf_setup',
99 'cnf-config=cnf-testsuite.yml', '-l', 'debug']
101 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
102 except subprocess.CalledProcessError as exc:
103 self.__logger.exception(
104 "Cannot run %s:\n%s", ' '.join(exc.cmd),
105 exc.output.decode("utf-8"))
108 self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
111 def run_conformance(self, **kwargs):
112 """Run CNF Conformance"""
113 cmd = ['cnf-testsuite', kwargs.get("tag", self.default_tag),
115 output = subprocess.run(
116 cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
118 self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
119 lfiles = glob.glob(os.path.join(
120 self.src_dir, 'results', 'cnf-testsuite-results-*.yml'))
121 results = max(lfiles, key=os.path.getmtime)
122 with open(os.path.join(
123 self.src_dir, 'results', results), encoding='utf-8') as yfile:
124 self.details = yaml.safe_load(yfile)
125 msg = prettytable.PrettyTable(
126 header_style='upper', padding_width=5,
127 field_names=['name', 'status'])
129 for item in self.details['items']:
130 msg.add_row([item['name'], item['status']])
131 if item['status'] == "passed":
133 elif item['status'] == "failed":
134 self.__logger.warning(
135 "%s %s", item['name'], item['status'])
136 self.__logger.info("\n\n%s\n", msg.get_string())
138 r'(\d+) of (\d+) essential tests passed', output.decode("utf-8"))
140 # https://github.com/cncf/cnf-certification/blob/main/reviewing.md
141 self.result = int(grp.group(1))
144 if not os.path.exists(self.res_dir):
145 os.makedirs(self.res_dir)
147 os.path.join(self.src_dir, 'results', results),
148 os.path.join(self.res_dir, 'cnf-testsuite-results.yml'))
150 def run(self, **kwargs):
151 """"Running the test with example CNF"""
152 self.start_time = time.time()
154 self.run_conformance(**kwargs)
155 self.stop_time = time.time()
158 for clean_cmd in ['uninstall_falco',
160 cmd = ['cnf-testsuite', clean_cmd,
161 'cnf-config=cnf-testsuite.yml']
162 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
163 self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
165 for namespace in ["cnf-testsuite", "litmus"]:
166 self.corev1.delete_namespace(namespace)
167 self.__logger.debug("delete_namespace: %s", namespace)
168 except client.rest.ApiException: