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.
34 import tornado
.httputil
37 from tornado
.platform
.asyncio
import AsyncIOMainLoop
38 from tornado
.ioloop
import IOLoop
39 from concurrent
.futures
.thread
import ThreadPoolExecutor
40 from concurrent
.futures
.process
import ProcessPoolExecutor
43 gi
.require_version('RwDts', '1.0')
44 gi
.require_version('RwPkgMgmtYang', '1.0')
45 gi
.require_version('RwProjectVnfdYang', '1.0')
46 from gi
.repository
import (
49 RwProjectVnfdYang
as RwVnfdYang
,
52 import rift
.tasklets
.rwlaunchpad
.uploader
as uploader
53 import rift
.tasklets
.rwlaunchpad
.message
as message
54 import rift
.tasklets
.rwlaunchpad
.export
as export
58 TEST_STRING
= "foobar"
60 class TestCase(rift
.test
.dts
.AbstractDTSTest
):
62 def configure_schema(cls
):
63 return RwPkgMgmtYang
.get_schema()
66 def configure_timeout(cls
):
69 def configure_test(self
, loop
, test_id
):
70 self
.log
.debug("STARTING - %s", test_id
)
71 self
.tinfo
= self
.new_tinfo(str(test_id
))
72 self
.dts
= rift
.tasklets
.DTS(self
.tinfo
, self
.schema
, self
.loop
)
75 mock_vnfd_catalog
= mock
.MagicMock()
76 self
.uid
, path
= self
.create_mock_package()
78 mock_vnfd
= RwVnfdYang
.YangData_RwProject_Project_VnfdCatalog_Vnfd
.from_dict({
81 mock_vnfd_catalog
= {self
.uid
: mock_vnfd
}
83 self
.app
= uploader
.UploaderApplication(
87 vnfd_catalog
=mock_vnfd_catalog
)
89 AsyncIOMainLoop().install()
90 self
.server
= tornado
.httpserver
.HTTPServer(
92 io_loop
=IOLoop
.current(),
98 def create_mock_package(self
):
99 uid
= str(uuid
.uuid4())
101 os
.getenv('RIFT_ARTIFACTS'),
102 "launchpad/packages/vnfd",
105 package_path
= os
.path
.join(path
, "pong_vnfd")
107 os
.makedirs(package_path
)
108 open(os
.path
.join(path
, "pong_vnfd.xml"), "wb").close()
109 open(os
.path
.join(path
, "logo.png"), "wb").close()
113 @rift.test
.dts
.async_test
114 def test_package_create_rpc(self
):
116 1. Verify the package-create RPC handler
117 2. Check if the log messages are updated which will be used by UI
119 3. Verify the package-update RPC handler
120 4. Check if the log messages are updated which will be used by UI
123 yield from self
.app
.register()
124 ip
= RwPkgMgmtYang
.YangInput_RwPkgMgmt_PackageCreate
.from_dict({
125 "package_type": "VNFD",
126 "external_url": "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
129 rpc_out
= yield from self
.dts
.query_rpc(
130 "I,/rw-pkg-mgmt:package-create",
131 rwdts
.XactFlag
.TRACE
,
136 result
= yield from itr
137 trans_id
= result
.result
.transaction_id
139 assert trans_id
is not None
141 yield from asyncio
.sleep(5, loop
=self
.loop
)
142 # Verify the message logs
143 data
= self
.app
.messages
[trans_id
]
144 assert data
is not None
146 assert type(data
) is message
.DownloadSuccess
149 ip
= RwPkgMgmtYang
.YangInput_RwPkgMgmt_PackageUpdate
.from_dict({
150 "package_type": "VNFD",
151 "external_url": "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
153 rpc_out
= yield from self
.dts
.query_rpc(
154 "I,/rw-pkg-mgmt:package-update",
155 rwdts
.XactFlag
.TRACE
,
160 result
= yield from itr
161 trans_id
= result
.result
.transaction_id
163 assert trans_id
is not None
164 yield from asyncio
.sleep(5, loop
=self
.loop
)
165 # Verify the message logs
166 data
= self
.app
.messages
[trans_id
]
167 assert data
is not None
169 assert type(data
) is message
.DownloadSuccess
172 @rift.test
.dts
.async_test
173 def test_package_export(self
):
175 1. Verify if the package export RPC handler work
176 2. A file is actually generated in the exports dir.
178 yield from self
.app
.register()
179 ip
= RwPkgMgmtYang
.YangInput_RwPkgMgmt_PackageExport
.from_dict({
180 "package_type": "VNFD",
181 "package_id": self
.uid
184 rpc_out
= yield from self
.dts
.query_rpc(
185 "I,/rw-pkg-mgmt:package-export",
186 rwdts
.XactFlag
.TRACE
,
192 result
= yield from itr
193 trans_id
= result
.result
.transaction_id
194 filename
= result
.result
.filename
196 assert trans_id
is not None
198 # Verify the message logs
199 data
= self
.app
.messages
[trans_id
]
200 assert data
is not None
202 assert type(data
) is export
.ExportSuccess
204 os
.getenv("RIFT_ARTIFACTS"),
210 assert os
.path
.isfile(path
)
214 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
216 parser
= argparse
.ArgumentParser()
217 parser
.add_argument('-v', '--verbose', action
='store_true')
218 parser
.add_argument('-n', '--no-runner', action
='store_true')
219 args
, unittest_args
= parser
.parse_known_args()
222 logging
.basicConfig(format
='TEST %(message)s')
223 logging
.getLogger().setLevel(logging
.DEBUG
)
226 unittest
.main(testRunner
=runner
, argv
=[sys
.argv
[0]] + unittest_args
)
228 if __name__
== '__main__':