rwlaunchpad/test/utest_rwnsm.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
  20 import argparse
  21 import logging
  22 import os
  23 import sys
  24 import unittest
  25 import uuid
  26 import xmlrunner
  27
  28 #Setting RIFT_VAR_ROOT if not already set for unit test execution
  29 if "RIFT_VAR_ROOT" not in os.environ:
  30     os.environ['RIFT_VAR_ROOT'] = os.path.join(os.environ['RIFT_INSTALL'], 'var/rift/unittest')
  31
  32 from gi.repository import (
  33         NsdYang,
  34         NsrYang,
  35         )
  36
  37 logger = logging.getLogger('test-rwnsmtasklet')
  38
  39 import rift.tasklets.rwnsmtasklet.rwnsmtasklet as rwnsmtasklet
  40 import rift.tasklets.rwnsmtasklet.xpath as rwxpath
  41
  42 class TestGiXpath(unittest.TestCase):
  43     def setUp(self):
  44         rwxpath.reset_cache()
  45
  46     def test_nsd_elements(self):
  47         """
  48         Test that a particular element in a list is corerctly retrieved. In
  49         this case, we are trying to retrieve an NSD from the NSD catalog.
  50
  51         """
  52         # Create the initial NSD catalog
  53         nsd_catalog = NsdYang.YangData_Nsd_NsdCatalog()
  54
  55         # Create an NSD, set its 'id', and add it to the catalog
  56         nsd_id = str(uuid.uuid4())
  57         nsd_catalog.nsd.append(
  58                 NsdYang.YangData_Nsd_NsdCatalog_Nsd(
  59                     id=nsd_id,
  60                     )
  61                 )
  62
  63         # Retrieve the NSD using and xpath expression
  64         xpath = '/nsd:nsd-catalog/nsd:nsd[nsd:id={}]'.format(nsd_id)
  65         nsd = rwxpath.getxattr(nsd_catalog, xpath)
  66
  67         self.assertEqual(nsd_id, nsd.id)
  68
  69         # Modified the name of the NSD using an xpath expression
  70         rwxpath.setxattr(nsd_catalog, xpath + "/nsd:name", "test-name")
  71
  72         name = rwxpath.getxattr(nsd_catalog, xpath + "/nsd:name")
  73         self.assertEqual("test-name", name)
  74
  75     def test_nsd_scalar_fields(self):
  76         """
  77         Test that setxattr correctly sets the value specified by an xpath.
  78
  79         """
  80         # Define a simple NSD
  81         nsd = NsdYang.YangData_Nsd_NsdCatalog_Nsd()
  82
  83         # Check that the unset fields are in fact set to None
  84         self.assertEqual(None, rwxpath.getxattr(nsd, "/nsd:nsd-catalog/nsd:nsd/nsd:name"))
  85         self.assertEqual(None, rwxpath.getxattr(nsd, "/nsd:nsd-catalog/nsd:nsd/nsd:short-name"))
  86
  87         # Set the values of the 'name' and 'short-name' fields
  88         rwxpath.setxattr(nsd, "/nsd:nsd-catalog/nsd:nsd/nsd:name", "test-name")
  89         rwxpath.setxattr(nsd, "/nsd:nsd-catalog/nsd:nsd/nsd:short-name", "test-short-name")
  90
  91         # Check that the 'name' and 'short-name' fields are correctly set
  92         self.assertEqual(nsd.name, rwxpath.getxattr(nsd, "/nsd:nsd-catalog/nsd:nsd/nsd:name"))
  93         self.assertEqual(nsd.short_name, rwxpath.getxattr(nsd, "/nsd:nsd-catalog/nsd:nsd/nsd:short-name"))
  94
  95
  96 class TestInputParameterSubstitution(unittest.TestCase):
  97     def setUp(self):
  98         self.substitute_input_parameters = rwnsmtasklet.InputParameterSubstitution(logger)
  99
 100     def test_null_arguments(self):
 101         """
 102         If None is passed to the substitutor for either the NSD or the NSR
 103         config, no exception should be raised.
 104
 105         """
 106         nsd = NsdYang.YangData_Nsd_NsdCatalog_Nsd()
 107         nsr_config = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr()
 108
 109         self.substitute_input_parameters(None, None)
 110         self.substitute_input_parameters(nsd, None)
 111         self.substitute_input_parameters(None, nsr_config)
 112
 113     def test_illegal_input_parameter(self):
 114         """
 115         In the NSD there is a list of the parameters that are allowed to be
 116         sbustituted by input parameters. This test checks that when an input
 117         parameter is provided in the NSR config that is not in the NSD, it is
 118         not applied.
 119
 120         """
 121         # Define the original NSD
 122         nsd = NsdYang.YangData_Nsd_NsdCatalog_Nsd()
 123         nsd.name = "robert"
 124         nsd.short_name = "bob"
 125
 126         # Define which parameters may be modified
 127         nsd.input_parameter_xpath.append(
 128                 NsdYang.YangData_Nsd_NsdCatalog_Nsd_InputParameterXpath(
 129                     xpath="/nsd:nsd-catalog/nsd:nsd/nsd:name",
 130                     label="NSD Name",
 131                     )
 132                 )
 133
 134         # Define the input parameters that are intended to be modified
 135         nsr_config = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr()
 136         nsr_config.input_parameter.extend([
 137             NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_InputParameter(
 138                 xpath="/nsd:nsd-catalog/nsd:nsd/nsd:name",
 139                 value="alice",
 140                 ),
 141             NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_InputParameter(
 142                 xpath="/nsd:nsd-catalog/nsd:nsd/nsd:short-name",
 143                 value="alice",
 144                 ),
 145             ])
 146
 147         self.substitute_input_parameters(nsd, nsr_config)
 148
 149         # Verify that only the parameter in the input_parameter_xpath list is
 150         # modified after the input parameters have been applied.
 151         self.assertEqual("alice", nsd.name)
 152         self.assertEqual("bob", nsd.short_name)
 153
 154     def test_substitution(self):
 155         """
 156         Test that substitution of input parameters occurs as expected.
 157
 158         """
 159         # Define the original NSD
 160         nsd = NsdYang.YangData_Nsd_NsdCatalog_Nsd()
 161         nsd.name = "robert"
 162         nsd.short_name = "bob"
 163
 164         # Define which parameters may be modified
 165         nsd.input_parameter_xpath.extend([
 166                 NsdYang.YangData_Nsd_NsdCatalog_Nsd_InputParameterXpath(
 167                     xpath="/nsd:nsd-catalog/nsd:nsd/nsd:name",
 168                     label="NSD Name",
 169                     ),
 170                 NsdYang.YangData_Nsd_NsdCatalog_Nsd_InputParameterXpath(
 171                     xpath="/nsd:nsd-catalog/nsd:nsd/nsd:short-name",
 172                     label="NSD Short Name",
 173                     ),
 174                 ])
 175
 176         # Define the input parameters that are intended to be modified
 177         nsr_config = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr()
 178         nsr_config.input_parameter.extend([
 179             NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_InputParameter(
 180                 xpath="/nsd:nsd-catalog/nsd:nsd/nsd:name",
 181                 value="robert",
 182                 ),
 183             NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_InputParameter(
 184                 xpath="/nsd:nsd-catalog/nsd:nsd/nsd:short-name",
 185                 value="bob",
 186                 ),
 187             ])
 188
 189         self.substitute_input_parameters(nsd, nsr_config)
 190
 191         # Verify that both the 'name' and 'short-name' fields are correctly
 192         # replaced.
 193         self.assertEqual("robert", nsd.name)
 194         self.assertEqual("bob", nsd.short_name)
 195
 196
 197 def main(argv=sys.argv[1:]):
 198     logging.basicConfig(format='TEST %(message)s')
 199
 200     parser = argparse.ArgumentParser()
 201     parser.add_argument('-v', '--verbose', action='store_true')
 202
 203     args = parser.parse_args(argv)
 204
 205     # Set the global logging level
 206     logging.getLogger().setLevel(logging.DEBUG if args.verbose else logging.FATAL)
 207
 208     # Make the test logger very quiet
 209     logger.addHandler(logging.NullHandler())
 210
 211     # The unittest framework requires a program name, so use the name of this
 212     # file instead (we do not want to have to pass a fake program name to main
 213     # when this is called from the interpreter).
 214     unittest.main(argv=[__file__] + argv,
 215             testRunner=xmlrunner.XMLTestRunner(
 216                 output=os.environ["RIFT_MODULE_TEST"]))
 217
 218 if __name__ == '__main__':
 219     main()