3 ############################################################################
4 # Copyright 2016 RIFT.io Inc #
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 #
10 # http://www.apache.org/licenses/LICENSE-2.0 #
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 ############################################################################
29 import rift
.mano
.examples
.ping_pong_nsd
as ping_pong_nsd
31 from rift
.mano
.utils
.compare_desc
import CompareDescShell
33 from rift
.tasklets
.rwlaunchpad
.tosca
import ExportTosca
34 from rift
.tasklets
.rwlaunchpad
.tosca
import ImportTosca
36 from rift
.package
.package
import TarPackageArchive
38 class PingPongDescriptors(object):
40 ping_vnfd
, pong_vnfd
, nsd
= \
41 ping_pong_nsd
.generate_ping_pong_descriptors(
46 ping_md5sum
='1234567890abcdefg',
47 pong_md5sum
='1234567890abcdefg',
51 use_placement_group
=False,
52 use_ns_init_conf
=False,
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]
59 class ToscaTestCase(unittest
.TestCase
):
60 """ Unittest for YANG to TOSCA and back translations
62 This generates the Ping Pong descrptors using the script
63 in examles and then converts it to TOSCA and back to YANG.
66 top_dir
= __file__
[:__file__
.find('/modules/core/')]
67 log_level
= logging
.WARN
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
)
81 """Run before each test method to initialize test environment."""
83 super(ToscaTestCase
, self
).setUp()
84 self
.output_dir
= tempfile
.mkdtemp()
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
)
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
)
98 return pkg
.create_archive('ping_pong_nsd', self
.output_dir
)
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
))
110 self
.log
.info("Tosca package was translated successfully")
113 raise ValueError("Not a valid TOSCA archive: {}".
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())
137 raise Exception("Unknown descriptor type {} found: {}".
138 format(desc_type
, pkg
.files
))
140 raise Exception("Did not find a valid tar file for yang model: {}".
143 def test_output(self
):
145 # Generate the Ping Pong descriptors
146 descs
= PingPongDescriptors()
148 # Translate the descriptors to TOSCA
149 tosca_file
= self
.yang_to_tosca(descs
)
151 # Now translate back to YANG
152 yang_files
= self
.tosca_to_yang(tosca_file
)
154 # Compare the generated YANG to original
155 self
.compare_descs(descs
, yang_files
)
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
162 except Exception as e
:
163 self
.log
.exception(e
)
164 self
.fail("Exception {}".format(e
))
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()
176 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
178 ToscaTestCase
.log_level
= logging
.DEBUG
if args
.verbose
else logging
.WARN
180 unittest
.main(testRunner
=runner
, argv
=[sys
.argv
[0]] + unittest_args
)
182 if __name__
== '__main__':