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         for namespace in ["cnf-testsuite", "default", "litmus"]:
  67             api_response = self.corev1.create_namespace(
  68                 client.V1Namespace(metadata=client.V1ObjectMeta(
  69                     generate_name=namespace, labels={
  70                         "pod-security.kubernetes.io/enforce": "baseline"})))
  71             self.__logger.debug(
  72                 "create_namespace: %s", api_response.metadata.name)
  73         os.chdir(self.src_dir)
  74         cmd = ['cnf-testsuite', 'setup', '-l', 'debug']
  75         try:
  76             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
  77         except subprocess.CalledProcessError as exc:
  78             self.__logger.exception(
  79                 "Cannot run %s:\n%s", ' '.join(exc.cmd),
  80                 exc.output.decode("utf-8"))
  81             self.result = 0
  82             return False
  83         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  84         cmd = ['cnf-testsuite', 'cnf_setup',
  85                'cnf-config=cnf-testsuite.yml', '-l', 'debug']
  86         try:
  87             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
  88         except subprocess.CalledProcessError as exc:
  89             self.__logger.exception(
  90                 "Cannot run %s:\n%s", ' '.join(exc.cmd),
  91                 exc.output.decode("utf-8"))
  92             self.result = 0
  93             return False
  94         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  95         return True
  96
  97     def run_conformance(self, **kwargs):
  98         """Run CNF Conformance"""
  99         cmd = ['cnf-testsuite', kwargs.get("tag", self.default_tag),
 100                '-l', 'debug']
 101         output = subprocess.run(
 102             cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
 103             check=False).stdout
 104         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
 105         lfiles = glob.glob(os.path.join(
 106             self.src_dir, 'results', 'cnf-testsuite-results-*.yml'))
 107         results = max(lfiles, key=os.path.getmtime)
 108         with open(os.path.join(
 109                 self.src_dir, 'results', results), encoding='utf-8') as yfile:
 110             self.details = yaml.safe_load(yfile)
 111             msg = prettytable.PrettyTable(
 112                 header_style='upper', padding_width=5,
 113                 field_names=['name', 'status'])
 114             item_criteria = 0
 115             for item in self.details['items']:
 116                 msg.add_row([item['name'], item['status']])
 117                 if item['status'] == "passed":
 118                     item_criteria += 1
 119                 elif item['status'] == "failed":
 120                     self.__logger.warning(
 121                         "%s %s", item['name'], item['status'])
 122             self.__logger.info("\n\n%s\n", msg.get_string())
 123         grp = re.search(
 124             r'(\d+) of (\d+) essential tests passed', output.decode("utf-8"))
 125         if grp:
 126             # https://github.com/cncf/cnf-certification/blob/main/reviewing.md
 127             self.result = int(grp.group(1))
 128         else:
 129             self.result = 0
 130         if not os.path.exists(self.res_dir):
 131             os.makedirs(self.res_dir)
 132         shutil.copy2(
 133             os.path.join(self.src_dir, 'results', results),
 134             os.path.join(self.res_dir, 'cnf-testsuite-results.yml'))
 135
 136     def run(self, **kwargs):
 137         """"Running the test with example CNF"""
 138         self.start_time = time.time()
 139         if self.setup():
 140             self.run_conformance(**kwargs)
 141         self.stop_time = time.time()
 142
 143     def clean(self):
 144         for clean_cmd in ['uninstall_falco',
 145                           'cnf_cleanup']:
 146             cmd = ['cnf-testsuite', clean_cmd,
 147                    'cnf-config=cnf-testsuite.yml']
 148             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 149             self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
 150         try:
 151             for namespace in ["cnf-testsuite", "litmus"]:
 152                 self.corev1.delete_namespace(namespace)
 153                 self.__logger.debug("delete_namespace: %s", namespace)
 154         except client.rest.ApiException:
 155             pass