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.
32 #Setting RIFT_VAR_ROOT if not already set for unit test execution
33 if "RIFT_VAR_ROOT" not in os
.environ
:
34 os
.environ
['RIFT_VAR_ROOT'] = os
.path
.join(os
.environ
['RIFT_INSTALL'], 'var/rift/unittest')
36 import rift
.package
.archive
37 import rift
.package
.charm
38 import rift
.package
.checksums
39 import rift
.package
.config
40 import rift
.package
.convert
41 import rift
.package
.icon
42 import rift
.package
.package
43 import rift
.package
.script
44 import rift
.package
.store
46 from rift
.tasklets
.rwlaunchpad
import export
49 gi
.require_version('RwVnfdYang', '1.0')
50 from gi
.repository
import (
58 class TestExport(utest_package
.PackageTestCase
):
61 self
._exporter
= export
.DescriptorPackageArchiveExporter(self
._log
)
62 self
._rw
_vnfd
_serializer
= rift
.package
.convert
.RwVnfdSerializer()
63 self
._vnfd
_serializer
= rift
.package
.convert
.VnfdSerializer()
65 def test_create_archive(self
):
66 rw_vnfd_msg
= RwVnfdYang
.YangData_Vnfd_VnfdCatalog_Vnfd(
67 id="new_id", name
="new_name", description
="new_description"
69 json_desc_str
= self
._rw
_vnfd
_serializer
.to_json_string(rw_vnfd_msg
)
71 vnfd_package
= self
.create_vnfd_package()
72 with io
.BytesIO() as archive_hdl
:
73 archive
= self
._exporter
.create_archive(
74 archive_hdl
, vnfd_package
, json_desc_str
, self
._rw
_vnfd
_serializer
79 # Create a new read-only archive from the archive handle and a package from that archive
80 archive
= rift
.package
.archive
.TarPackageArchive(self
._log
, archive_hdl
)
81 package
= archive
.create_package()
83 # Ensure that the descriptor in the package has been overwritten
84 self
.assertEqual(package
.descriptor_msg
, rw_vnfd_msg
)
86 def test_export_package(self
):
87 rw_vnfd_msg
= RwVnfdYang
.YangData_Vnfd_VnfdCatalog_Vnfd(
88 id="new_id", name
="new_name", description
="new_description",
89 meta
="THIS FIELD IS NOT IN REGULAR VNFD"
91 vnfd_msg
= VnfdYang
.YangData_Vnfd_VnfdCatalog_Vnfd()
92 vnfd_msg
.from_dict(rw_vnfd_msg
.as_dict(), ignore_missing_keys
=True)
94 self
.assertNotEqual(rw_vnfd_msg
, vnfd_msg
)
96 json_desc_str
= self
._rw
_vnfd
_serializer
.to_json_string(rw_vnfd_msg
)
98 with tempfile
.TemporaryDirectory() as tmp_dir
:
99 vnfd_package
= self
.create_vnfd_package()
100 pkg_id
= str(uuid
.uuid4())
101 exported_path
= self
._exporter
.export_package(
102 vnfd_package
, tmp_dir
, pkg_id
, json_desc_str
, self
._vnfd
_serializer
105 self
.assertTrue(os
.path
.isfile(exported_path
))
106 self
.assertTrue(tarfile
.is_tarfile(exported_path
))
108 with
open(exported_path
, "rb") as archive_hdl
:
109 archive
= rift
.package
.archive
.TarPackageArchive(self
._log
, archive_hdl
)
110 package
= archive
.create_package()
112 self
.assertEqual(package
.descriptor_msg
, vnfd_msg
)
114 def test_export_cleanup(self
):
115 loop
= asyncio
.get_event_loop()
116 with tempfile
.TemporaryDirectory() as tmp_dir
:
117 archive_files
= [tempfile
.mkstemp(dir=tmp_dir
, suffix
=".tar.gz")[1] for _
in range(2)]
119 # Set the mtime on only one of the files to test the min_age_secs argument
120 times
= (time
.time(), time
.time() - 10)
121 os
.utime(archive_files
[0], times
)
123 task
= loop
.create_task(
124 export
.periodic_export_cleanup(
125 self
._log
, loop
, tmp_dir
, period_secs
=.01, min_age_secs
=5
128 loop
.run_until_complete(asyncio
.sleep(.05, loop
=loop
))
130 if task
.done() and task
.exception() is not None:
131 raise task
.exception()
133 self
.assertFalse(task
.done())
135 self
.assertFalse(os
.path
.exists(archive_files
[0]))
136 self
.assertTrue(os
.path
.exists(archive_files
[1]))
138 def main(argv
=sys
.argv
[1:]):
139 logging
.basicConfig(format
='TEST %(message)s')
141 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
142 parser
= argparse
.ArgumentParser()
143 parser
.add_argument('-v', '--verbose', action
='store_true')
144 parser
.add_argument('-n', '--no-runner', action
='store_true')
146 args
, unknown
= parser
.parse_known_args(argv
)
150 # Set the global logging level
151 logging
.getLogger().setLevel(logging
.DEBUG
if args
.verbose
else logging
.ERROR
)
153 # The unittest framework requires a program name, so use the name of this
154 # file instead (we do not want to have to pass a fake program name to main
155 # when this is called from the interpreter).
156 unittest
.main(argv
=[__file__
] + unknown
+ ["-v"], testRunner
=runner
)
158 if __name__
== '__main__':