Use cnf certifaction criteria
[functest-kubernetes.git] / functest_kubernetes / cnf_conformance / conformance.py
1 #!/usr/bin/env python
2
3 # Copyright (c) 2020 Orange and others.
4 #
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
9
10 """
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
15 """
16
17 from __future__ import division
18
19 import glob
20 import logging
21 import os
22 import re
23 import shutil
24 import subprocess
25 import time
26 import yaml
27
28 import prettytable
29
30 from xtesting.core import testcase
31
32
33 class CNFConformance(testcase.TestCase):
34     """ Implement CNF Conformance driver.
35
36     https://hackmd.io/@vulk/SkY54QnsU
37     """
38
39     src_dir = '/src/cnf-testsuite'
40     bin_dir = '/usr/local/bin'
41     default_tag = 'cert'
42
43     __logger = logging.getLogger(__name__)
44
45     def __init__(self, **kwargs):
46         super().__init__(**kwargs)
47         self.output_log_name = 'functest-kubernetes.log'
48         self.output_debug_log_name = 'functest-kubernetes.debug.log'
49
50     def check_requirements(self):
51         """Check if cnf-testsuite is in $PATH"""
52         if not os.path.exists(os.path.join(self.bin_dir, 'cnf-testsuite')):
53             self.__logger.warning(
54                 "cnf-testsuite is not compiled for arm and arm64 for the "
55                 "time being")
56             self.is_skipped = True
57
58     def setup(self):
59         """Implement initialization and pre-reqs steps"""
60         if os.path.exists(os.path.join(self.src_dir, "results")):
61             shutil.rmtree(os.path.join(self.src_dir, "results"))
62         os.chdir(self.src_dir)
63         cmd = ['cnf-testsuite', 'setup']
64         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
65         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
66         cmd = ['cnf-testsuite', 'cnf_setup',
67                'cnf-config=cnf-testsuite.yml']
68         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
69         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
70
71     def run_conformance(self, **kwargs):
72         """Run CNF Conformance"""
73         cmd = ['cnf-testsuite', kwargs.get("tag", self.default_tag)]
74         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
75         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
76         lfiles = glob.glob(os.path.join(
77             self.src_dir, 'results', 'cnf-testsuite-results-*.yml'))
78         results = max(lfiles, key=os.path.getmtime)
79         with open(os.path.join(
80                 self.src_dir, 'results', results), encoding='utf-8') as yfile:
81             self.details = yaml.safe_load(yfile)
82             msg = prettytable.PrettyTable(
83                 header_style='upper', padding_width=5,
84                 field_names=['name', 'status'])
85             item_criteria = 0
86             for item in self.details['items']:
87                 msg.add_row([item['name'], item['status']])
88                 if item['status'] == "passed":
89                     item_criteria += 1
90                 elif item['status'] == "failed":
91                     self.__logger.warning(
92                         "%s %s", item['name'], item['status'])
93             self.__logger.info("\n\n%s\n", msg.get_string())
94         grp = re.search(
95             r'(\d+) of (\d+) essential tests passed', output.decode("utf-8"))
96         if grp:
97             # https://github.com/cncf/cnf-certification/blob/main/reviewing.md
98             self.result = int(grp.group(1))
99         else:
100             self.result = 0
101         if not os.path.exists(self.res_dir):
102             os.makedirs(self.res_dir)
103         shutil.copy2(
104             os.path.join(self.src_dir, 'results', results),
105             os.path.join(self.res_dir, 'cnf-testsuite-results.yml'))
106
107     def run(self, **kwargs):
108         """"Running the test with example CNF"""
109         self.start_time = time.time()
110         try:
111             self.setup()
112             self.run_conformance(**kwargs)
113         except subprocess.CalledProcessError as exc:
114             self.__logger.warning(
115                 "Can not run CNT Conformance: \n%s\n%s\n",
116                 " ".join(exc.cmd), exc.output.decode("utf-8"))
117         except Exception:  # pylint: disable=broad-except
118             self.__logger.warning("CNF Conformance exited with errors")
119         self.stop_time = time.time()
120
121     def clean(self):
122         for clean_cmd in ['uninstall_falco',
123                           'cnf_cleanup']:
124             cmd = ['cnf-testsuite', clean_cmd,
125                    'cnf-config=cnf-testsuite.yml']
126             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
127             self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))