3 # Copyright 2016 RIFT.io Inc
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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.
29 import rift
.mano
.examples
.ping_pong_nsd
as ping_pong_nsd
31 import rift
.mano
.tosca_translator
.shell
as tshell
33 import rift
.mano
.utils
.compare_desc
as cmpdesc
35 from rift
.mano
.tosca_translator
.common
.utils
import ChecksumUtils
37 from rift
.mano
.yang_translator
.common
.utils
import _
38 import rift
.mano
.yang_translator
.shell
as shell
41 _TRUE_VALUES
= ('True', 'true', '1', 'yes')
44 class PingPongDescriptors(object):
46 def __init__(self
, output_dir
, log
):
47 ping_vnfd
, pong_vnfd
, nsd
= \
48 ping_pong_nsd
.generate_ping_pong_descriptors(
56 ping_md5sum
='1234567890abcdefg',
57 pong_md5sum
='1234567890abcdefg',
61 use_placement_group
= False,
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
)
69 for root
, dirs
, files
in os
.walk(cur_dir
):
70 rel_dir
= root
.replace(cur_dir
+'/', '')
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
))
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
))
82 # Create the tar archive
83 tar_cmd
= "tar zcvf {0}.tar.gz {0}"
84 subprocess
.check_call(tar_cmd
.format(desc
),
86 stdout
=subprocess
.DEVNULL
)
90 for d
in os
.listdir(output_dir
):
94 class TestYangTranslator(unittest
.TestCase
):
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"
102 failure_msg
= _('The program raised an exception unexpectedly.')
105 log_level
= logging
.WARN
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
)
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
))
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
)
129 cls
.log
.warn("Descriptor directory: {}".format(cls
.desc_dir
))
131 def test_missing_arg(self
):
132 self
.assertRaises(SystemExit, shell
.main
, '')
134 def test_invalid_file_arg(self
):
135 self
.assertRaises(SystemExit, shell
.main
, 'translate me')
137 def test_invalid_file_value(self
):
138 self
.assertRaises(SystemExit,
140 ('--template-file=template.txt'))
142 def test_invalid_type_value(self
):
143 self
.assertRaises(SystemExit,
146 '--template-type=xyz'))
148 def compare_tosca(self
, gen_desc
, exp_desc
):
149 gen
= "--generated="+gen_desc
150 exp
= "--expected="+exp_desc
151 cmpdesc
.main([gen
, exp
])
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()
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
))
168 shell
.main(args
, log
=self
.log
)
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
))
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
)
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
)
189 # Check the dirs are present
190 dirs
= os
.listdir(yang_out_dir
)
191 self
.assertTrue(len(dirs
) >= 3)
193 except Exception as e
:
194 self
.log
.exception(e
)
195 self
.fail(_("Exception {}").format(e
))
199 shutil
.rmtree(temp_dir
)
203 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
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()
212 TestYangTranslator
.log_level
= logging
.DEBUG
if args
.verbose
else logging
.WARN
214 unittest
.main(testRunner
=runner
, argv
=[sys
.argv
[0]] + unittest_args
)
216 if __name__
== '__main__':