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