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 gi
.require_version('RwCloudYang', '1.0')
33 gi
.require_version('RwDts', '1.0')
34 gi
.require_version('RwNsmYang', '1.0')
35 gi
.require_version('RwLaunchpadYang', '1.0')
36 gi
.require_version('RwResourceMgrYang', '1.0')
37 gi
.require_version('RwcalYang', '1.0')
38 gi
.require_version('RwNsrYang', '1.0')
39 gi
.require_version('NsrYang', '1.0')
40 gi
.require_version('RwlogMgmtYang', '1.0')
42 from gi
.repository
import (
43 RwCloudYang
as rwcloudyang
,
45 RwLaunchpadYang
as launchpadyang
,
46 RwNsmYang
as rwnsmyang
,
47 RwNsrYang
as rwnsryang
,
49 RwResourceMgrYang
as rmgryang
,
50 RwcalYang
as rwcalyang
,
51 RwConfigAgentYang
as rwcfg_agent
,
55 from gi
.repository
.RwTypes
import RwStatus
56 import rift
.mano
.examples
.ping_pong_nsd
as ping_pong_nsd
64 class ManoTestCase(rift
.test
.dts
.AbstractDTSTest
):
66 DTS GI interface unittests
68 Note: Each tests uses a list of asyncio.Events for staging through the
69 test. These are required here because we are bring up each coroutine
70 ("tasklet") at the same time and are not implementing any re-try
71 mechanisms. For instance, this is used in numerous tests to make sure that
72 a publisher is up and ready before the subscriber sends queries. Such
73 event lists should not be used in production software.
77 def configure_suite(cls
, rwmain
):
78 nsm_dir
= os
.environ
.get('NSM_DIR')
80 rwmain
.add_tasklet(nsm_dir
, 'rwnsmtasklet')
83 def configure_schema(cls
):
84 return rwnsmyang
.get_schema()
87 def configure_timeout(cls
):
91 def get_cal_account(account_type
, account_name
):
93 Creates an object for class RwcalYang.Clo
95 account
= rwcloudyang
.CloudAccount()
96 if account_type
== 'mock':
97 account
.name
= account_name
98 account
.account_type
= "mock"
99 account
.mock
.username
= "mock_user"
100 elif ((account_type
== 'openstack_static') or (account_type
== 'openstack_dynamic')):
101 account
.name
= account_name
102 account
.account_type
= 'openstack'
103 account
.openstack
.key
= openstack_info
['username']
104 account
.openstack
.secret
= openstack_info
['password']
105 account
.openstack
.auth_url
= openstack_info
['auth_url']
106 account
.openstack
.tenant
= openstack_info
['project_name']
107 account
.openstack
.mgmt_network
= openstack_info
['mgmt_network']
111 def configure_cloud_account(self
, dts
, cloud_type
, cloud_name
="cloud1"):
112 account
= self
.get_cal_account(cloud_type
, cloud_name
)
113 account_xpath
= "C,/rw-cloud:cloud/rw-cloud:account[rw-cloud:name='{}']".format(cloud_name
)
114 self
.log
.info("Configuring cloud-account: %s", account
)
115 yield from dts
.query_create(account_xpath
,
116 rwdts
.XactFlag
.ADVISE
,
120 def wait_tasklets(self
):
121 yield from asyncio
.sleep(5, loop
=self
.loop
)
123 def configure_test(self
, loop
, test_id
):
124 self
.log
.debug("STARTING - %s", self
.id())
125 self
.tinfo
= self
.new_tinfo(self
.id())
126 self
.dts
= rift
.tasklets
.DTS(self
.tinfo
, self
.schema
, self
.loop
)
128 def test_create_nsr_record(self
):
132 yield from self
.wait_tasklets()
135 yield from self
.configure_cloud_account(self
.dts
, cloud_type
, "mock_account")
139 rpc_ip
= nsryang
.YangInput_Nsr_ExecScaleIn
.from_dict({
142 'scaling_group_name_ref': "foo"})
144 yield from self
.dts
.query_rpc("/nsr:exec-scale-in", 0, rpc_ip
)
146 future
= asyncio
.ensure_future(run_test(), loop
=self
.loop
)
147 self
.run_until(future
.done
)
148 if future
.exception() is not None:
149 self
.log
.error("Caught exception during test")
150 raise future
.exception()
154 top_dir
= __file__
[:__file__
.find('/modules/core/')]
155 build_dir
= os
.path
.join(top_dir
, '.build/modules/core/rwvx/src/core_rwvx-build')
156 launchpad_build_dir
= os
.path
.join(top_dir
, '.build/modules/core/mc/core_mc-build/rwlaunchpad')
158 if 'NSM_DIR' not in os
.environ
:
159 os
.environ
['NSM_DIR'] = os
.path
.join(launchpad_build_dir
, 'plugins/rwnsm')
161 runner
= xmlrunner
.XMLTestRunner(output
=os
.environ
["RIFT_MODULE_TEST"])
163 parser
= argparse
.ArgumentParser()
164 parser
.add_argument('-v', '--verbose', action
='store_true')
165 parser
.add_argument('-n', '--no-runner', action
='store_true')
166 args
, unittest_args
= parser
.parse_known_args()
170 ManoTestCase
.log_level
= logging
.DEBUG
if args
.verbose
else logging
.WARN
172 unittest
.main(testRunner
=runner
, argv
=[sys
.argv
[0]] + unittest_args
)
174 if __name__
== '__main__':