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
38 from tornado
.platform
.asyncio
import AsyncIOMainLoop
39 from tornado
.ioloop
import IOLoop
40 from concurrent
.futures
.thread
import ThreadPoolExecutor
41 from concurrent
.futures
.process
import ProcessPoolExecutor
42 gi
.require_version('RwDts', '1.0')
43 gi
.require_version('RwPkgMgmtYang', '1.0')
44 from gi
.repository
import (
51 import rift
.tasklets
.rwlaunchpad
.uploader
as uploader
52 import rift
.tasklets
.rwlaunchpad
.message
as message
53 import rift
.tasklets
.rwlaunchpad
.export
as export
57 TEST_STRING
= "foobar"
59 class TestCase(rift
.test
.dts
.AbstractDTSTest
):
61 def configure_schema(cls
):
62 return RwPkgMgmtYang
.get_schema()
65 def configure_timeout(cls
):
68 def configure_test(self
, loop
, test_id
):
69 self
.log
.debug("STARTING - %s", test_id
)
70 self
.tinfo
= self
.new_tinfo(str(test_id
))
71 self
.dts
= rift
.tasklets
.DTS(self
.tinfo
, self
.schema
, self
.loop
)
74 mock_vnfd_catalog
= mock
.MagicMock()
75 self
.uid
, path
= self
.create_mock_package()
77 mock_vnfd
= RwVnfdYang
.YangData_Vnfd_VnfdCatalog_Vnfd
.from_dict({
80 mock_vnfd_catalog
= {self
.uid
: mock_vnfd
}
82 self
.app
= uploader
.UploaderApplication(
86 vnfd_catalog
=mock_vnfd_catalog
)
88 AsyncIOMainLoop().install()
89 self
.server
= tornado
.httpserver
.HTTPServer(
91 io_loop
=IOLoop
.current(),
97 def create_mock_package(self
):
98 uid
= str(uuid
.uuid4())
100 os
.getenv('RIFT_ARTIFACTS'),
101 "launchpad/packages/vnfd",
104 package_path
= os
.path
.join(path
, "pong_vnfd")
106 os
.makedirs(package_path
)
107 open(os
.path
.join(path
, "pong_vnfd.xml"), "wb").close()
108 open(os
.path
.join(path
, "logo.png"), "wb").close()
112 @rift.test
.dts
.async_test
113 def test_package_create_rpc(self
):
115 1. Verify the package-create RPC handler
116 2. Check if the log messages are updated which will be used by UI
118 3. Verify the package-update RPC handler
119 4. Check if the log messages are updated which will be used by UI
122 yield from self
.app
.register()
123 ip
= RwPkgMgmtYang
.YangInput_RwPkgMgmt_PackageCreate
.from_dict({
124 "package_type": "VNFD",
125 "external_url": "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
128 rpc_out
= yield from self
.dts
.query_rpc(
129 "I,/rw-pkg-mgmt:package-create",
130 rwdts
.XactFlag
.TRACE
,
135 result
= yield from itr
136 trans_id
= result
.result
.transaction_id
138 assert trans_id
is not None
140 yield from asyncio
.sleep(5, loop
=self
.loop
)
141 # Verify the message logs
142 data
= self
.app
.messages
[trans_id
]
143 assert data
is not None
145 assert type(data
) is message
.DownloadSuccess
148 ip
= RwPkgMgmtYang
.YangInput_RwPkgMgmt_PackageUpdate
.from_dict({
149 "package_type": "VNFD",
150 "external_url": "http://repo.riftio.com/releases/open.riftio.com/4.2.1/VNFS/ping_vnfd.tar.gz"
152 rpc_out
= yield from self
.dts
.query_rpc(
153 "I,/rw-pkg-mgmt:package-update",
154 rwdts
.XactFlag
.TRACE
,
159 result
= yield from itr
160 trans_id
= result
.result
.transaction_id
162 assert trans_id
is not None
163 yield from asyncio
.sleep(5, loop
=self
.loop
)
164 # Verify the message logs
165 data
= self
.app
.messages
[trans_id
]
166 assert data
is not None
168 assert type(data
) is message
.DownloadSuccess
171 @rift.test
.dts
.async_test
172 def test_package_export(self
):
174 1. Verify if the package export RPC handler work
175 2. A file is actually generated in the exports dir.
177 yield from self
.app
.register()
178 ip
= RwPkgMgmtYang
.YangInput_RwPkgMgmt_PackageExport
.from_dict({
179 "package_type": "VNFD",
180 "package_id": self
.uid
183 rpc_out
= yield from self
.dts
.query_rpc(
184 "I,/rw-pkg-mgmt:package-export",
185 rwdts
.XactFlag
.TRACE
,
191 result
= yield from itr
192 trans_id
= result
.result
.transaction_id
193 filename
= result
.result
.filename
195 assert trans_id
is not None
197 # Verify the message logs
198 data
= self
.app
.messages
[trans_id
]
199 assert data
is not None
201 assert type(data
) is export
.ExportSuccess
203 os
.getenv("RIFT_ARTIFACTS"),
209 assert os
.path
.isfile(path
)
213 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
215 parser
= argparse
.ArgumentParser()
216 parser
.add_argument('-v', '--verbose', action
='store_true')
217 parser
.add_argument('-n', '--no-runner', action
='store_true')
218 args
, unittest_args
= parser
.parse_known_args()
221 logging
.basicConfig(format
='TEST %(message)s')
222 logging
.getLogger().setLevel(logging
.DEBUG
)
225 unittest
.main(testRunner
=runner
, argv
=[sys
.argv
[0]] + unittest_args
)
227 if __name__
== '__main__':