rwlaunchpad/test/tosca_ut.py

   1 #!/usr/bin/env python3
   2
   3 ############################################################################
   4 # Copyright 2016 RIFT.io Inc                                               #
   5 #                                                                          #
   6 # Licensed under the Apache License, Version 2.0 (the "License");          #
   7 # you may not use this file except in compliance with the License.         #
   8 # You may obtain a copy of the License at                                  #
   9 #                                                                          #
  10 #     http://www.apache.org/licenses/LICENSE-2.0                           #
  11 #                                                                          #
  12 # Unless required by applicable law or agreed to in writing, software      #
  13 # distributed under the License is distributed on an "AS IS" BASIS,        #
  14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
  15 # See the License for the specific language governing permissions and      #
  16 # limitations under the License.                                           #
  17 ############################################################################
  18
  19 import argparse
  20 import logging
  21 import os
  22 import shutil
  23 import sys
  24 import tarfile
  25 import tempfile
  26 import unittest
  27 import xmlrunner
  28
  29 import rift.mano.examples.ping_pong_nsd as ping_pong_nsd
  30
  31 from rift.mano.utils.compare_desc import CompareDescShell
  32
  33 from rift.tasklets.rwlaunchpad.tosca import ExportTosca
  34 from rift.tasklets.rwlaunchpad.tosca import ImportTosca
  35
  36 from rift.package.package import TarPackageArchive
  37
  38 class PingPongDescriptors(object):
  39     def __init__(self):
  40         ping_vnfd, pong_vnfd, nsd = \
  41                 ping_pong_nsd.generate_ping_pong_descriptors(
  42                     pingcount=1,
  43                     external_vlr_count=1,
  44                     internal_vlr_count=0,
  45                     num_vnf_vms=1,
  46                     ping_md5sum='1234567890abcdefg',
  47                     pong_md5sum='1234567890abcdefg',
  48                     mano_ut=False,
  49                     use_scale_group=True,
  50                     use_mon_params=True,
  51                     use_placement_group=False,
  52                     use_ns_init_conf=False,
  53                 )
  54         self.ping_pong_nsd = nsd.descriptor.nsd[0]
  55         self.ping_vnfd = ping_vnfd.descriptor.vnfd[0]
  56         self.pong_vnfd = pong_vnfd.descriptor.vnfd[0]
  57
  58
  59 class ToscaTestCase(unittest.TestCase):
  60     """ Unittest for YANG to TOSCA and back translations
  61
  62     This generates the Ping Pong descrptors using the script
  63     in examles and then converts it to TOSCA and back to YANG.
  64     """
  65     default_timeout = 0
  66     top_dir = __file__[:__file__.find('/modules/core/')]
  67     log_level = logging.WARN
  68     log = None
  69
  70     @classmethod
  71     def setUpClass(cls):
  72         fmt = logging.Formatter(
  73                 '%(asctime)-23s %(levelname)-5s  (%(name)s@%(process)d:%(filename)s:%(lineno)d) - %(message)s')
  74         stderr_handler = logging.StreamHandler(stream=sys.stderr)
  75         stderr_handler.setFormatter(fmt)
  76         logging.basicConfig(level=cls.log_level)
  77         cls.log = logging.getLogger('tosca-ut')
  78         cls.log.addHandler(stderr_handler)
  79
  80     def setUp(self):
  81         """Run before each test method to initialize test environment."""
  82
  83         super(ToscaTestCase, self).setUp()
  84         self.output_dir = tempfile.mkdtemp()
  85
  86     def compare_dict(self, gen_d, exp_d):
  87         gen = "--generated="+str(gen_d)
  88         exp = "--expected="+str(exp_d)
  89         CompareDescShell.compare_dicts(gen, exp, log=self.log)
  90
  91     def yang_to_tosca(self, descs):
  92         """Convert YANG model to TOSCA model"""
  93         pkg = ExportTosca(self.log)
  94         nsd_id = pkg.add_nsd(descs.ping_pong_nsd)
  95         pkg.add_vnfd(nsd_id, descs.ping_vnfd)
  96         pkg.add_vnfd(nsd_id, descs.pong_vnfd)
  97
  98         return pkg.create_archive('ping_pong_nsd', self.output_dir)
  99
 100     def tosca_to_yang(self, tosca_file):
 101         """Convert TOSCA model to YANG model"""
 102         if ImportTosca.is_tosca_package(tosca_file):
 103             # This could be a tosca package, try processing
 104             tosca = ImportTosca(self.log, tosca_file, out_dir=self.output_dir)
 105             files = tosca.translate()
 106             if files is None or len(files) < 3:
 107                 raise ValueError("Could not process as a "
 108                                  "TOSCA package {}: {}".format(tosca_file, files))
 109             else:
 110                  self.log.info("Tosca package was translated successfully")
 111                  return files
 112         else:
 113             raise ValueError("Not a valid TOSCA archive: {}".
 114                              format(tosca_file))
 115
 116     def compare_descs(self, descs, yang_files):
 117         """Compare the sescriptors generated with original"""
 118         for yang_file in yang_files:
 119             if tarfile.is_tarfile(yang_file):
 120                 with open(yang_file, "r+b") as tar:
 121                     archive = TarPackageArchive(self.log, tar)
 122                     pkg = archive.create_package()
 123                     desc_type = pkg.descriptor_type
 124                     if desc_type == 'nsd':
 125                         nsd_yang = pkg.descriptor_msg.as_dict()
 126                         self.compare_dict(nsd_yang,
 127                                           descs.ping_pong_nsd.as_dict())
 128                     elif desc_type == 'vnfd':
 129                         vnfd_yang = pkg.descriptor_msg.as_dict()
 130                         if 'ping_vnfd' == vnfd_yang['name']:
 131                             self.compare_dict(vnfd_yang,
 132                                               descs.ping_vnfd.as_dict())
 133                         elif 'pong_vnfd' == vnfd_yang['name']:
 134                             self.compare_dict(vnfd_yang,
 135                                               descs.pong_vnfd.as_dict())
 136                         else:
 137                             raise Exception("Unknown descriptor type {} found: {}".
 138                                             format(desc_type, pkg.files))
 139             else:
 140                 raise Exception("Did not find a valid tar file for yang model: {}".
 141                                 format(yang_file))
 142
 143     def test_output(self):
 144         try:
 145             # Generate the Ping Pong descriptors
 146             descs = PingPongDescriptors()
 147
 148             # Translate the descriptors to TOSCA
 149             tosca_file = self.yang_to_tosca(descs)
 150
 151             # Now translate back to YANG
 152             yang_files = self.tosca_to_yang(tosca_file)
 153
 154             # Compare the generated YANG to original
 155             self.compare_descs(descs, yang_files)
 156
 157             # Removing temp dir only on success to allow debug in case of failures
 158             if self.output_dir is not None:
 159                 shutil.rmtree(self.output_dir)
 160                 self.output_dir = None
 161
 162         except Exception as e:
 163             self.log.exception(e)
 164             self.fail("Exception {}".format(e))
 165
 166
 167
 168 def main():
 169     parser = argparse.ArgumentParser()
 170     parser.add_argument('-v', '--verbose', action='store_true')
 171     parser.add_argument('-n', '--no-runner', action='store_true')
 172     args, unittest_args = parser.parse_known_args()
 173     if args.no_runner:
 174         runner = None
 175     else:
 176         runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 177
 178     ToscaTestCase.log_level = logging.DEBUG if args.verbose else logging.WARN
 179
 180     unittest.main(testRunner=runner, argv=[sys.argv[0]] + unittest_args)
 181
 182 if __name__ == '__main__':
 183     main()