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
113
114
115
116
117
118
119
120
121
122
|
#!/usr/bin/env python
# Copyright (c) 2020 Orange and others.
#
# 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
"""
The CNF Conformance program enables interoperability of Cloud native Network
Functions (CNFs) from multiple vendors running on top of Kubernetes supplied by
different vendors [1].
[1] https://github.com/cncf/cnf-testsuite
"""
from __future__ import division
import glob
import logging
import os
import re
import shutil
import subprocess
import time
import yaml
import prettytable
from xtesting.core import testcase
class CNFConformance(testcase.TestCase):
""" Implement CNF Conformance driver.
https://hackmd.io/@vulk/SkY54QnsU
"""
src_dir = '/src/cnf-testsuite'
bin_dir = '/usr/local/bin'
default_tag = 'cert'
__logger = logging.getLogger(__name__)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.output_log_name = 'functest-kubernetes.log'
self.output_debug_log_name = 'functest-kubernetes.debug.log'
def check_requirements(self):
"""Check if cnf-testsuite is in $PATH"""
if not os.path.exists(os.path.join(self.bin_dir, 'cnf-testsuite')):
self.__logger.warning(
"cnf-testsuite is not compiled for arm and arm64 for the "
"time being")
self.is_skipped = True
def setup(self):
"""Implement initialization and pre-reqs steps"""
if os.path.exists(os.path.join(self.src_dir, "results")):
shutil.rmtree(os.path.join(self.src_dir, "results"))
os.chdir(self.src_dir)
cmd = ['cnf-testsuite', 'setup']
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
cmd = ['cnf-testsuite', 'cnf_setup',
'cnf-config=cnf-testsuite.yml']
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
def run_conformance(self, **kwargs):
"""Run CNF Conformance"""
cmd = ['cnf-testsuite', kwargs.get("tag", self.default_tag)]
output = subprocess.run(
cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
check=False).stdout
self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
lfiles = glob.glob(os.path.join(
self.src_dir, 'results', 'cnf-testsuite-results-*.yml'))
results = max(lfiles, key=os.path.getmtime)
with open(os.path.join(
self.src_dir, 'results', results), encoding='utf-8') as yfile:
self.details = yaml.safe_load(yfile)
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
field_names=['name', 'status'])
item_criteria = 0
for item in self.details['items']:
msg.add_row([item['name'], item['status']])
if item['status'] == "passed":
item_criteria += 1
elif item['status'] == "failed":
self.__logger.warning(
"%s %s", item['name'], item['status'])
self.__logger.info("\n\n%s\n", msg.get_string())
grp = re.search(
r'(\d+) of (\d+) essential tests passed', output.decode("utf-8"))
if grp:
# https://github.com/cncf/cnf-certification/blob/main/reviewing.md
self.result = int(grp.group(1))
else:
self.result = 0
if not os.path.exists(self.res_dir):
os.makedirs(self.res_dir)
shutil.copy2(
os.path.join(self.src_dir, 'results', results),
os.path.join(self.res_dir, 'cnf-testsuite-results.yml'))
def run(self, **kwargs):
""""Running the test with example CNF"""
self.start_time = time.time()
self.setup()
self.run_conformance(**kwargs)
self.stop_time = time.time()
def clean(self):
for clean_cmd in ['uninstall_falco',
'cnf_cleanup']:
cmd = ['cnf-testsuite', clean_cmd,
'cnf-config=cnf-testsuite.yml']
output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
|
