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-conformance
  15 """
  16
  17 from __future__ import division
  18
  19 import fnmatch
  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-conformance'
  40     bin_dir = '/usr/local/bin'
  41     default_tag = 'all'
  42
  43     __logger = logging.getLogger(__name__)
  44
  45     def setup(self):
  46         """Implement initialization and pre-reqs steps"""
  47         os.makedirs(self.res_dir, exist_ok=True)
  48
  49         shutil.copy2(os.path.join(self.src_dir, 'points.yml'), self.res_dir)
  50         shutil.copy2(
  51             os.path.join(self.src_dir, 'cnf-conformance.yml'), self.res_dir)
  52         os.chdir(self.res_dir)
  53         # cnf-conformance must be in the working dir
  54         # https://github.com/cncf/cnf-conformance/issues/388
  55         if not os.path.exists(os.path.join(self.res_dir, 'cnf-conformance')):
  56             os.symlink(
  57                 os.path.join(self.bin_dir, 'cnf-conformance'),
  58                 os.path.join(self.res_dir, 'cnf-conformance'))
  59         cmd = ['cnf-conformance', 'setup']
  60         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
  61         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  62         cmd = ['cnf-conformance', 'cnf_setup',
  63                'cnf-config=cnf-conformance.yml']
  64         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
  65         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  66
  67     def run_conformance(self, **kwargs):
  68         """Run CNF Conformance"""
  69         # a previous results.yml leads to interactive mode
  70         if os.path.exists(os.path.join(self.res_dir, 'results.yml')):
  71             os.remove(os.path.join(self.res_dir, 'results.yml'))
  72         cmd = ['cnf-conformance', kwargs.get("tag", self.default_tag)]
  73         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
  74         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
  75         for lfile in os.listdir(self.res_dir):
  76             if fnmatch.fnmatch(lfile, 'cnf-conformance-results-*.yml'):
  77                 with open(os.path.join(self.res_dir, lfile)) as yfile:
  78                     self.details = yaml.safe_load(yfile)
  79                     msg = prettytable.PrettyTable(
  80                         header_style='upper', padding_width=5,
  81                         field_names=['name', 'status'])
  82                     for item in self.details['items']:
  83                         msg.add_row([item['name'], item['status']])
  84                     self.__logger.info("\n\n%s\n", msg.get_string())
  85         grp = re.search(r'Final score: (\d+) of (\d+)', output.decode("utf-8"))
  86         if grp:
  87             self.result = int(grp.group(1)) / int(grp.group(2)) * 100
  88
  89     def run(self, **kwargs):
  90         """"Running the test with example CNF"""
  91         self.start_time = time.time()
  92         try:
  93             self.setup()
  94             self.run_conformance(**kwargs)
  95         except Exception:  # pylint: disable=broad-except
  96             self.__logger.exception("Can not run CNF Conformance")
  97         self.stop_time = time.time()
  98
  99     def clean(self):
 100         cmd = ['cnf-conformance', 'cnf_cleanup',
 101                'cnf-config=cnf-conformance.yml']
 102         output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 103         self.__logger.info("%s\n%s", " ".join(cmd), output.decode("utf-8"))
 104         shutil.rmtree(self.res_dir)