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', '-l', 'debug']
  64         try:
  65             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
  66         except subprocess.CalledProcessError as exc:
  67             self.__logger.exception(
  68                 "Cannot run %s:\n%s", ' '.join(exc.cmd),
  69                 exc.output.decode("utf-8"))
  70             self.result = 0
  71             return False
  72         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  73         cmd = ['cnf-testsuite', 'cnf_setup',
  74                'cnf-config=cnf-testsuite.yml', '-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         return True
  85
  86     def run_conformance(self, **kwargs):
  87         """Run CNF Conformance"""
  88         cmd = ['cnf-testsuite', kwargs.get("tag", self.default_tag),
  89                '-l', 'debug']
  90         output = subprocess.run(
  91             cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
  92             check=False).stdout
  93         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  94         lfiles = glob.glob(os.path.join(
  95             self.src_dir, 'results', 'cnf-testsuite-results-*.yml'))
  96         results = max(lfiles, key=os.path.getmtime)
  97         with open(os.path.join(
  98                 self.src_dir, 'results', results), encoding='utf-8') as yfile:
  99             self.details = yaml.safe_load(yfile)
 100             msg = prettytable.PrettyTable(
 101                 header_style='upper', padding_width=5,
 102                 field_names=['name', 'status'])
 103             item_criteria = 0
 104             for item in self.details['items']:
 105                 msg.add_row([item['name'], item['status']])
 106                 if item['status'] == "passed":
 107                     item_criteria += 1
 108                 elif item['status'] == "failed":
 109                     self.__logger.warning(
 110                         "%s %s", item['name'], item['status'])
 111             self.__logger.info("\n\n%s\n", msg.get_string())
 112         grp = re.search(
 113             r'(\d+) of (\d+) essential tests passed', output.decode("utf-8"))
 114         if grp:
 115             # https://github.com/cncf/cnf-certification/blob/main/reviewing.md
 116             self.result = int(grp.group(1))
 117         else:
 118             self.result = 0
 119         if not os.path.exists(self.res_dir):
 120             os.makedirs(self.res_dir)
 121         shutil.copy2(
 122             os.path.join(self.src_dir, 'results', results),
 123             os.path.join(self.res_dir, 'cnf-testsuite-results.yml'))
 124
 125     def run(self, **kwargs):
 126         """"Running the test with example CNF"""
 127         self.start_time = time.time()
 128         if self.setup():
 129             self.run_conformance(**kwargs)
 130         self.stop_time = time.time()
 131
 132     def clean(self):
 133         for clean_cmd in ['uninstall_falco',
 134                           'cnf_cleanup']:
 135             cmd = ['cnf-testsuite', clean_cmd,
 136                    'cnf-config=cnf-testsuite.yml']
 137             output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 138             self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))