1 # Copyright 2016 RIFT.IO Inc
2 # Copyright 2016 Telefónica Investigación y Desarrollo S.A.U.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 #!/usr/bin/env python3
23 import requests_toolbelt
29 logger
= logging
.getLogger(__name__
)
31 def chomp_json(string
):
32 return ''.join([line
.strip() for line
in string
.splitlines()])
34 @pytest.fixture(scope
='session')
35 def session(request
, so_user
, so_pass
):
36 client_session
= requests
.session()
37 client_session
.auth
= (so_user
, so_pass
)
38 client_session
.headers
= {
39 "Accept" : "application/vnd.yang.data+json",
40 "Content-Type" : "application/vnd.yang.data+json",
42 client_session
.verify
= False
43 _
, cert
, key
= certs
.get_bootstrap_cert_and_key()
44 client_session
.cert
= (cert
, key
)
47 @pytest.fixture(scope
='session')
48 def rest_endpoint(so_host
, so_port
):
49 return 'https://{host}:{port}'.format(host
=so_host
, port
=so_port
)
51 def test_so_started(session
, rest_endpoint
):
52 ''' Get contents of /vcs/info/components and verify all components have started successfully
54 uri
= "{endpoint}/api/operational/vcs/info/components".format(endpoint
=rest_endpoint
)
55 response
= session
.request("GET", uri
)
56 vcs_info
= json
.loads(response
.text
)
57 for component
in vcs_info
['rw-base:components']['component_info']:
58 assert component
['state'] == 'RUNNING'
60 def test_configure_vim_account(session
, rest_endpoint
, vim_type
, vim_host
, vim_user
, vim_tenant
, vim_pass
):
61 ''' Configure an openstack vim account
63 uri
= "{endpoint}/api/config/cloud/account".format(
64 endpoint
=rest_endpoint
,
70 "account-type":"openstack",
75 "auth_url": "http://%s:5000/v3/",
77 "mgmt-network": "private"
81 ''' % (vim_user
, vim_pass
, vim_host
, vim_tenant
)
82 response
= session
.request("POST", uri
, data
=chomp_json(account
))
84 uri
= "{endpoint}/api/operational/cloud/account".format(endpoint
=rest_endpoint
)
85 response
= session
.request("GET", uri
)
86 assert 'error' not in response
.text
87 vim_accounts
= json
.loads(response
.text
)
91 for account
in vim_accounts
['rw-cloud:account']:
92 if account
['name'] == 'vim-0':
95 assert found_account
== True
98 @pytest.fixture(scope
='session')
99 def ping_pkg(package_location
):
100 '''ping vnfd package location'''
101 return "%s/ping_vnfd.tar.gz" % (package_location
)
104 @pytest.fixture(scope
='session')
105 def pong_pkg(package_location
):
106 '''pong vnfd package location'''
107 return "%s/pong_vnfd.tar.gz" % (package_location
)
110 @pytest.fixture(scope
='session')
111 def ping_pong_pkg(package_location
):
112 '''ping pong nsd package location'''
113 return "%s/ping_pong_nsd.tar.gz" % (package_location
)
116 def test_onboard_pingpong_descriptors(session
, so_host
, ping_pkg
, pong_pkg
, ping_pong_pkg
, rest_endpoint
):
117 ''' Onboard ping_pong descriptors
119 onboard_command
= 'onboard_pkg -s {host} -u {ping_pkg}'.format(
123 subprocess
.check_call(onboard_command
, shell
=True)
125 onboard_command
= 'onboard_pkg -s {host} -u {pong_pkg}'.format(
129 subprocess
.check_call(onboard_command
, shell
=True)
131 onboard_command
= 'onboard_pkg -s {host} -u {ping_pong_pkg}'.format(
133 ping_pong_pkg
=ping_pong_pkg
,
135 subprocess
.check_call(onboard_command
, shell
=True)
138 def test_instantiate_ping_pong(session
, so_host
, rest_endpoint
):
139 ''' Instantiate an instance of ping pong from descriptors
141 uri
= "{endpoint}/api/operational/nsd-catalog".format(endpoint
=rest_endpoint
)
142 response
= session
.request("GET", uri
)
143 catalog
= json
.loads(response
.text
)
145 for nsd
in catalog
['nsd:nsd-catalog']['nsd']:
146 if nsd
['name'] == 'ping_pong_nsd':
149 assert ping_pong_nsd
is not None
151 instantiate_command
= 'onboard_pkg -s {host} -i instance-0 -d {nsd_id} -c vim-0'.format(
153 nsd_id
=ping_pong_nsd
['id']
155 subprocess
.check_call(instantiate_command
, shell
=True)
157 def wait_for_ping_pong(instance_name
, timeout
=600, retry_interval
=5):
158 start_time
= time
.time()
160 uri
= "{endpoint}/api/operational/ns-instance-opdata".format(endpoint
=rest_endpoint
)
161 response
= session
.request("GET", uri
)
162 print(response
, response
.text
)
163 opdata
= json
.loads(response
.text
)
166 for instance
in opdata
['nsr:ns-instance-opdata']['nsr']:
167 if instance
['name-ref'] == instance_name
:
170 assert nsr
is not None, response
.text
171 assert nsr
['operational-status'] not in ['failed']
172 assert nsr
['config-status'] not in ['failed']
174 if nsr
['operational-status'] in ['running'] and nsr
['config-status'] in ['configured']:
177 time_elapsed
= time
.time() - start_time
178 time_remaining
= timeout
- time_elapsed
179 assert time_remaining
> 0
180 time
.sleep(min(time_remaining
, retry_interval
))
182 wait_for_ping_pong('instance-0')