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 from kubernetes import client
  29 from kubernetes import config
  30 import prettytable
  31 from xtesting.core import testcase
  32
  33
  34 class CNFConformance(testcase.TestCase):
  35     # pylint: disable=too-many-instance-attributes
  36     """ Implement CNF Conformance driver.
  37
  38     https://hackmd.io/@vulk/SkY54QnsU
  39     """
  40
  41     src_dir = '/src/cnf-testsuite'
  42     bin_dir = '/usr/local/bin'
  43     default_tag = 'cert'
  44
  45     __logger = logging.getLogger(__name__)
  46
  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'
  53
  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 "
  59                 "time being")
  60             self.is_skipped = True
  61
  62     def setup(self):
  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(
  71                         namespace,
  72                         client.V1Namespace(metadata=client.V1ObjectMeta(
  73                             labels={
  74                                 "pod-security.kubernetes.io/enforce":
  75                                     "baseline"})))
  76                     self.__logger.debug(
  77                         "patch_namespace: %s", namespace)
  78                     break
  79             else:
  80                 self.corev1.create_namespace(
  81                     client.V1Namespace(metadata=client.V1ObjectMeta(
  82                         name=namespace, labels={
  83                             "pod-security.kubernetes.io/enforce":
  84                                 "baseline"})))
  85                 self.__logger.debug(
  86                     "create_namespace: %s", namespace)
  87         os.chdir(self.src_dir)
  88         cmd = ['cnf-testsuite', 'setup', '-l', 'debug']
  89         try:
  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"))
  95             self.result = 0
  96             return False
  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']
 100         try:
 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"))
 106             self.result = 0
 107             return False
 108         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
 109         return True
 110
 111     def run_conformance(self, **kwargs):
 112         """Run CNF Conformance"""
 113         cmd = ['cnf-testsuite', kwargs.get("tag", self.default_tag),
 114                '-l', 'debug']
 115         output = subprocess.run(
 116             cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
 117             check=False).stdout
 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'])
 128             item_criteria = 0
 129             for item in self.details['items']:
 130                 msg.add_row([item['name'], item['status']])
 131                 if item['status'] == "passed":
 132                     item_criteria += 1
 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())
 137         grp = re.search(
 138             r'(\d+) of (\d+) essential tests passed', output.decode("utf-8"))
 139         if grp:
 140             # https://github.com/cncf/cnf-certification/blob/main/reviewing.md
 141             self.result = int(grp.group(1))
 142         else:
 143             self.result = 0
 144         if not os.path.exists(self.res_dir):
 145             os.makedirs(self.res_dir)
 146         shutil.copy2(
 147             os.path.join(self.src_dir, 'results', results),
 148             os.path.join(self.res_dir, 'cnf-testsuite-results.yml'))
 149
 150     def run(self, **kwargs):
 151         """"Running the test with example CNF"""
 152         self.start_time = time.time()
 153         if self.setup():
 154             self.run_conformance(**kwargs)
 155         self.stop_time = time.time()
 156
 157     def clean(self):
 158         for clean_cmd in ['uninstall_falco',
 159                           'cnf_cleanup']:
 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"))
 164         try:
 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:
 169             pass