common/python/rift/mano/yang_translator/test/yang_translator_ut.py

   1 #!/usr/bin/env python3
   2
   3 # Copyright 2016 RIFT.io Inc
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16
  17
  18 import argparse
  19 import logging
  20 import os
  21 import shutil
  22 import subprocess
  23 import sys
  24 import tempfile
  25 import xmlrunner
  26
  27 import unittest
  28
  29 import rift.mano.examples.ping_pong_nsd as ping_pong_nsd
  30
  31 import rift.mano.tosca_translator.shell as tshell
  32
  33 import rift.mano.utils.compare_desc as cmpdesc
  34
  35 from rift.mano.tosca_translator.common.utils import ChecksumUtils
  36
  37 from rift.mano.yang_translator.common.utils import _
  38 import rift.mano.yang_translator.shell as shell
  39
  40
  41 _TRUE_VALUES = ('True', 'true', '1', 'yes')
  42
  43
  44 class PingPongDescriptors(object):
  45
  46     def __init__(self, output_dir, log):
  47         ping_vnfd, pong_vnfd, nsd = \
  48                 ping_pong_nsd.generate_ping_pong_descriptors(
  49                     fmt='yaml',
  50                     write_to_file=True,
  51                     out_dir=output_dir,
  52                     pingcount=1,
  53                     external_vlr_count=1,
  54                     internal_vlr_count=0,
  55                     num_vnf_vms=1,
  56                     ping_md5sum='1234567890abcdefg',
  57                     pong_md5sum='1234567890abcdefg',
  58                     mano_ut=False,
  59                     use_scale_group=True,
  60                     use_mon_params=True,
  61                     use_placement_group = False,
  62                 )
  63
  64         # Create the tar files in output dir
  65         def create_archive(desc):
  66             # Create checksum file
  67             cur_dir = os.path.join(output_dir, desc)
  68             flist = {}
  69             for root, dirs, files in os.walk(cur_dir):
  70                 rel_dir = root.replace(cur_dir+'/', '')
  71                 for f in files:
  72                     fpath = os.path.join(root, f)
  73                     flist[os.path.join(rel_dir, f)] = \
  74                                         ChecksumUtils.get_md5(fpath)
  75             log.debug(_("Files in {}: {}").format(cur_dir, flist))
  76
  77             chksumfile = os.path.join(cur_dir, 'checksums.txt')
  78             with open(chksumfile, 'w') as c:
  79                 for key in sorted(flist.keys()):
  80                     c.write("{}  {}\n".format(flist[key], key))
  81
  82             # Create the tar archive
  83             tar_cmd = "tar zcvf {0}.tar.gz {0}"
  84             subprocess.check_call(tar_cmd.format(desc),
  85                                   shell=True,
  86                                   stdout=subprocess.DEVNULL)
  87
  88         prevdir = os.getcwd()
  89         os.chdir(output_dir)
  90         for d in os.listdir(output_dir):
  91             create_archive(d)
  92         os.chdir(prevdir)
  93
  94 class TestYangTranslator(unittest.TestCase):
  95
  96     yang_helloworld = os.path.join(
  97         os.path.dirname(os.path.abspath(__file__)),
  98         "data/yang_helloworld.json")
  99     template_file = '--template-file=' + yang_helloworld
 100     template_validation = "--validate-only"
 101     debug="--debug"
 102     failure_msg = _('The program raised an exception unexpectedly.')
 103
 104     default_timeout = 0
 105     log_level = logging.WARN
 106     log = None
 107
 108     @classmethod
 109     def setUpClass(cls):
 110         fmt = logging.Formatter(
 111             '%(asctime)-23s %(levelname)-5s  (%(name)s@%(process)d:%(filename)s:%(lineno)d) - %(message)s')
 112         stderr_handler = logging.StreamHandler(stream=sys.stderr)
 113         stderr_handler.setFormatter(fmt)
 114         logging.basicConfig(level=cls.log_level)
 115         cls.log = logging.getLogger('yang-translator-ut')
 116         cls.log.addHandler(stderr_handler)
 117
 118         cls.desc_dir = tempfile.mkdtemp()
 119         PingPongDescriptors(cls.desc_dir, cls.log)
 120         cls.log.debug("Yang comaprison descs in {}".format(cls.desc_dir))
 121
 122     @classmethod
 123     def tearDownClass(cls):
 124         '''Clean up temporary directory'''
 125         # Remove directory if not debug level
 126         if cls.log_level != logging.DEBUG:
 127             shutil.rmtree(cls.desc_dir)
 128         else:
 129             cls.log.warn("Descriptor directory: {}".format(cls.desc_dir))
 130
 131     def test_missing_arg(self):
 132         self.assertRaises(SystemExit, shell.main, '')
 133
 134     def test_invalid_file_arg(self):
 135         self.assertRaises(SystemExit, shell.main, 'translate me')
 136
 137     def test_invalid_file_value(self):
 138         self.assertRaises(SystemExit,
 139                           shell.main,
 140                           ('--template-file=template.txt'))
 141
 142     def test_invalid_type_value(self):
 143         self.assertRaises(SystemExit,
 144                           shell.main,
 145                           (self.template_file,
 146                            '--template-type=xyz'))
 147
 148     def compare_tosca(self, gen_desc, exp_desc):
 149         gen = "--generated="+gen_desc
 150         exp = "--expected="+exp_desc
 151         cmpdesc.main([gen, exp])
 152
 153     def test_output(self):
 154         test_base_dir = os.path.join(os.path.dirname(
 155             os.path.abspath(__file__)), 'data')
 156         temp_dir = tempfile.mkdtemp()
 157         args = []
 158         for f in os.listdir(self.desc_dir):
 159             fpath = os.path.join(self.desc_dir, f)
 160             if os.path.isfile(fpath):
 161                 template = '--template-file='+fpath
 162                 args.append(template)
 163         output_dir = "--output-dir=" + temp_dir
 164         args.append(output_dir)
 165         self.log.debug("Args passed: {}".format(args))
 166
 167         try:
 168             shell.main(args, log=self.log)
 169
 170             # Check the dirs are present
 171             out_dir = os.path.join(temp_dir, 'ping_pong_nsd')
 172             self.assertTrue(os.path.isdir(out_dir))
 173             dirs = os.listdir(out_dir)
 174             expected_dirs = ['TOSCA-Metadata', 'Definitions']
 175             self.assertTrue(set(expected_dirs) <= set(dirs))
 176
 177             # Compare the descriptors
 178             gen_desc = os.path.join(out_dir, 'Definitions', 'ping_pong_nsd.yaml')
 179             exp_desc = os.path.join(test_base_dir,
 180                                     'ping_pong_tosca.yaml')
 181             self.compare_tosca(gen_desc, exp_desc)
 182
 183             # Convert back to yang and compare
 184             template = '--template-file='+gen_desc
 185             yang_out_dir = os.path.join(temp_dir, 'ping_pong_yang')
 186             output_dir = "--output-dir=" + yang_out_dir
 187             tshell.main([template, output_dir], log=self.log)
 188
 189             # Check the dirs are present
 190             dirs = os.listdir(yang_out_dir)
 191             self.assertTrue(len(dirs) >= 3)
 192
 193         except Exception as e:
 194             self.log.exception(e)
 195             self.fail(_("Exception {}").format(e))
 196
 197         finally:
 198             if temp_dir:
 199                 shutil.rmtree(temp_dir)
 200
 201
 202 def main():
 203     runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"])
 204
 205     parser = argparse.ArgumentParser()
 206     parser.add_argument('-v', '--verbose', action='store_true')
 207     parser.add_argument('-n', '--no-runner', action='store_true')
 208     args, unittest_args = parser.parse_known_args()
 209     if args.no_runner:
 210         runner = None
 211
 212     TestYangTranslator.log_level = logging.DEBUG if args.verbose else logging.WARN
 213
 214     unittest.main(testRunner=runner, argv=[sys.argv[0]] + unittest_args)
 215
 216 if __name__ == '__main__':
 217     main()