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
.checksums
38 import rift
.package
.convert
39 import rift
.package
.icon
40 import rift
.package
.package
41 import rift
.package
.script
42 import rift
.package
.store
44 from rift
.tasklets
.rwlaunchpad
import export
47 gi
.require_version('ProjectVnfdYang', '1.0')
48 gi
.require_version('RwProjectVnfdYang', '1.0')
49 from gi
.repository
import (
50 RwProjectVnfdYang
as RwVnfdYang
,
51 ProjectVnfdYang
as VnfdYang
,
57 class TestExport(utest_package
.PackageTestCase
):
60 self
._exporter
= export
.DescriptorPackageArchiveExporter(self
._log
)
61 self
._rw
_vnfd
_serializer
= rift
.package
.convert
.RwVnfdSerializer()
62 self
._vnfd
_serializer
= rift
.package
.convert
.VnfdSerializer()
64 def test_create_archive(self
):
65 rw_vnfd_msg
= RwVnfdYang
.YangData_RwProject_Project_VnfdCatalog_Vnfd(
66 id="new_id", name
="new_name", description
="new_description"
68 json_desc_str
= self
._rw
_vnfd
_serializer
.to_json_string(rw_vnfd_msg
)
70 vnfd_package
= self
.create_vnfd_package()
71 with io
.BytesIO() as archive_hdl
:
72 archive
= self
._exporter
.create_archive(
73 archive_hdl
, vnfd_package
, json_desc_str
, self
._rw
_vnfd
_serializer
78 # Create a new read-only archive from the archive handle and a package from that archive
79 archive
= rift
.package
.archive
.TarPackageArchive(self
._log
, archive_hdl
)
80 package
= archive
.create_package()
82 # Ensure that the descriptor in the package has been overwritten
83 self
.assertEqual(package
.descriptor_msg
, rw_vnfd_msg
)
85 def test_export_package(self
):
86 rw_vnfd_msg
= RwVnfdYang
.YangData_RwProject_Project_VnfdCatalog_Vnfd(
87 id="new_id", name
="new_name", description
="new_description",
88 meta
="THIS FIELD IS NOT IN REGULAR VNFD"
90 vnfd_msg
= VnfdYang
.YangData_RwProject_Project_VnfdCatalog_Vnfd()
91 vnfd_msg
.from_dict(rw_vnfd_msg
.as_dict(), ignore_missing_keys
=True)
93 self
.assertNotEqual(rw_vnfd_msg
, vnfd_msg
)
95 json_desc_str
= self
._rw
_vnfd
_serializer
.to_json_string(rw_vnfd_msg
)
97 with tempfile
.TemporaryDirectory() as tmp_dir
:
98 vnfd_package
= self
.create_vnfd_package()
99 pkg_id
= str(uuid
.uuid4())
100 exported_path
= self
._exporter
.export_package(
101 vnfd_package
, tmp_dir
, pkg_id
, json_desc_str
, self
._vnfd
_serializer
104 self
.assertTrue(os
.path
.isfile(exported_path
))
105 self
.assertTrue(tarfile
.is_tarfile(exported_path
))
107 with
open(exported_path
, "rb") as archive_hdl
:
108 archive
= rift
.package
.archive
.TarPackageArchive(self
._log
, archive_hdl
)
109 package
= archive
.create_package()
111 self
.assertEqual(package
.descriptor_msg
, vnfd_msg
)
113 def test_export_cleanup(self
):
114 loop
= asyncio
.get_event_loop()
115 with tempfile
.TemporaryDirectory() as tmp_dir
:
116 archive_files
= [tempfile
.mkstemp(dir=tmp_dir
, suffix
=".tar.gz")[1] for _
in range(2)]
118 # Set the mtime on only one of the files to test the min_age_secs argument
119 times
= (time
.time(), time
.time() - 10)
120 os
.utime(archive_files
[0], times
)
122 task
= loop
.create_task(
123 export
.periodic_export_cleanup(
124 self
._log
, loop
, tmp_dir
, period_secs
=.01, min_age_secs
=5
127 loop
.run_until_complete(asyncio
.sleep(.05, loop
=loop
))
129 if task
.done() and task
.exception() is not None:
130 raise task
.exception()
132 self
.assertFalse(task
.done())
134 self
.assertFalse(os
.path
.exists(archive_files
[0]))
135 self
.assertTrue(os
.path
.exists(archive_files
[1]))
137 def main(argv
=sys
.argv
[1:]):
138 logging
.basicConfig(format
='TEST %(message)s')
140 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
141 parser
= argparse
.ArgumentParser()
142 parser
.add_argument('-v', '--verbose', action
='store_true')
143 parser
.add_argument('-n', '--no-runner', action
='store_true')
145 args
, unknown
= parser
.parse_known_args(argv
)
149 # Set the global logging level
150 logging
.getLogger().setLevel(logging
.DEBUG
if args
.verbose
else logging
.ERROR
)
152 # The unittest framework requires a program name, so use the name of this
153 # file instead (we do not want to have to pass a fake program name to main
154 # when this is called from the interpreter).
155 unittest
.main(argv
=[__file__
] + unknown
+ ["-v"], testRunner
=runner
)
157 if __name__
== '__main__':