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