def test_142_cinder_endpoint(self):
"""Verify the cinder endpoint data."""
u.log.debug('Checking cinder endpoint...')
endpoints = self.keystone_v2.endpoints.list()
admin_port = internal_port = public_port = '8776'
expected = {
'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null
}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
amulet.raise_status(amulet.FAIL,
msg='cinder endpoint: {}'.format(ret))
python类raise_status()的实例源码
def test_202_keystone_cinder_identity_service_relation(self):
"""Verify the keystone identity-service relation data"""
u.log.debug('Checking keystone to cinder id relation data...')
relation = ['identity-service', 'cinder:identity-service']
expected = {
'service_protocol': 'http',
'service_tenant': 'services',
'admin_token': 'ubuntutesting',
'service_password': u.not_null,
'service_port': '5000',
'auth_port': '35357',
'auth_protocol': 'http',
'private-address': u.valid_ip,
'auth_host': u.valid_ip,
'service_username': 'cinder_cinderv2',
'service_tenant_id': u.not_null,
'service_host': u.valid_ip
}
if self._get_openstack_release() >= self.xenial_pike:
expected['service_username'] = 'cinderv3_cinderv2'
for unit in self.keystone_sentries:
ret = u.validate_relation_data(unit, relation, expected)
if ret:
message = u.relation_error('keystone identity-service', ret)
amulet.raise_status(amulet.FAIL, msg=message)
def test_302_keystone_logging_config(self):
"""Verify the data in the keystone logging config file"""
u.log.debug('Checking keystone config file...')
expected = {
'logger_root': {
'level': 'WARNING',
'handlers': 'file,production',
},
'handlers': {
'keys': 'production,file,devel'
},
'handler_file': {
'level': 'DEBUG',
'args': "('{}', 'a')".format(self.log_file)
}
}
for unit in self.keystone_sentries:
for section, pairs in expected.iteritems():
ret = u.validate_config_data(unit, self.logging_config,
section, pairs)
if ret:
message = "keystone logging config error: {}".format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
def get_ceph_pools(self, sentry_unit):
"""Return a dict of ceph pools from a single ceph unit, with
pool name as keys, pool id as vals."""
pools = {}
cmd = 'sudo ceph osd lspools'
output, code = sentry_unit.run(cmd)
if code != 0:
msg = ('{} `{}` returned {} '
'{}'.format(sentry_unit.info['unit_name'],
cmd, code, output))
amulet.raise_status(amulet.FAIL, msg=msg)
# Example output: 0 data,1 metadata,2 rbd,3 cinder,4 glance,
for pool in str(output).split(','):
pool_id_name = pool.split(' ')
if len(pool_id_name) == 2:
pool_id = pool_id_name[0]
pool_name = pool_id_name[1]
pools[pool_name] = int(pool_id)
self.log.debug('Pools on {}: {}'.format(sentry_unit.info['unit_name'],
pools))
return pools
basic_deployment.py 文件源码
项目:charm-nova-cloud-controller
作者: openstack
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def test_102_service_catalog(self):
"""Verify that the service catalog endpoint data is valid."""
u.log.debug('Checking keystone service catalog...')
endpoint_vol = {'adminURL': u.valid_url,
'region': 'RegionOne',
'id': u.not_null,
'publicURL': u.valid_url,
'internalURL': u.valid_url}
endpoint_id = {'adminURL': u.valid_url,
'region': 'RegionOne',
'id': u.not_null,
'publicURL': u.valid_url,
'internalURL': u.valid_url}
if self._get_openstack_release() >= self.trusty_kilo:
expected = {'compute': [endpoint_vol], 'identity': [endpoint_id]}
else:
expected = {'s3': [endpoint_vol], 'compute': [endpoint_vol],
'ec2': [endpoint_vol], 'identity': [endpoint_id]}
actual = self.keystone_demo.service_catalog.get_endpoints()
ret = u.validate_svc_catalog_endpoint_data(expected, actual)
if ret:
amulet.raise_status(amulet.FAIL, msg=ret)
basic_deployment.py 文件源码
项目:charm-nova-cloud-controller
作者: openstack
项目源码
文件源码
阅读 16
收藏 0
点赞 0
评论 0
def test_104_openstack_compute_api_endpoint(self):
"""Verify the openstack compute api (osapi) endpoint data."""
u.log.debug('Checking compute endpoint data...')
endpoints = self.keystone.endpoints.list()
admin_port = internal_port = public_port = '8774'
expected = {
'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null
}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
message = 'osapi endpoint: {}'.format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
basic_deployment.py 文件源码
项目:charm-nova-cloud-controller
作者: openstack
项目源码
文件源码
阅读 16
收藏 0
点赞 0
评论 0
def test_106_ec2_api_endpoint(self):
"""Verify the EC2 api endpoint data."""
if self._get_openstack_release() >= self.trusty_kilo:
return
u.log.debug('Checking ec2 endpoint data...')
endpoints = self.keystone.endpoints.list()
admin_port = internal_port = public_port = '8773'
expected = {
'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null
}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
message = 'EC2 endpoint: {}'.format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
basic_deployment.py 文件源码
项目:charm-nova-cloud-controller
作者: openstack
项目源码
文件源码
阅读 16
收藏 0
点赞 0
评论 0
def test_108_s3_api_endpoint(self):
"""Verify the S3 api endpoint data."""
if self._get_openstack_release() >= self.trusty_kilo:
return
u.log.debug('Checking s3 endpoint data...')
endpoints = self.keystone.endpoints.list()
admin_port = internal_port = public_port = '3333'
expected = {
'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null
}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
message = 'S3 endpoint: {}'.format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
basic_deployment.py 文件源码
项目:charm-nova-cloud-controller
作者: openstack
项目源码
文件源码
阅读 17
收藏 0
点赞 0
评论 0
def test_302_api_rate_limiting_is_enabled(self):
"""
Check that API rate limiting is enabled.
"""
u.log.debug('Checking api-paste config file data...')
unit = self.nova_cc_sentry
conf = '/etc/nova/api-paste.ini'
if self._get_openstack_release() >= self.trusty_mitaka:
section = "filter:legacy_ratelimit"
else:
section = "filter:ratelimit"
factory = ("nova.api.openstack.compute.limits:RateLimitingMiddleware"
".factory")
expected = {"paste.filter_factory": factory,
"limits": "( POST, '*', .*, 9999, MINUTE );"}
ret = u.validate_config_data(unit, conf, section, expected)
if ret:
message = "api paste config error: {}".format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
def get_ceph_pools(self, sentry_unit):
"""Return a dict of ceph pools from a single ceph unit, with
pool name as keys, pool id as vals."""
pools = {}
cmd = 'sudo ceph osd lspools'
output, code = sentry_unit.run(cmd)
if code != 0:
msg = ('{} `{}` returned {} '
'{}'.format(sentry_unit.info['unit_name'],
cmd, code, output))
amulet.raise_status(amulet.FAIL, msg=msg)
# Example output: 0 data,1 metadata,2 rbd,3 cinder,4 glance,
for pool in str(output).split(','):
pool_id_name = pool.split(' ')
if len(pool_id_name) == 2:
pool_id = pool_id_name[0]
pool_name = pool_id_name[1]
pools[pool_name] = int(pool_id)
self.log.debug('Pools on {}: {}'.format(sentry_unit.info['unit_name'],
pools))
return pools
def get_ceph_pools(self, sentry_unit):
"""Return a dict of ceph pools from a single ceph unit, with
pool name as keys, pool id as vals."""
pools = {}
cmd = 'sudo ceph osd lspools'
output, code = sentry_unit.run(cmd)
if code != 0:
msg = ('{} `{}` returned {} '
'{}'.format(sentry_unit.info['unit_name'],
cmd, code, output))
amulet.raise_status(amulet.FAIL, msg=msg)
# Example output: 0 data,1 metadata,2 rbd,3 cinder,4 glance,
for pool in str(output).split(','):
pool_id_name = pool.split(' ')
if len(pool_id_name) == 2:
pool_id = pool_id_name[0]
pool_name = pool_id_name[1]
pools[pool_name] = int(pool_id)
self.log.debug('Pools on {}: {}'.format(sentry_unit.info['unit_name'],
pools))
return pools
def get_process_id_list(self, sentry_unit, process_name,
expect_success=True):
"""Get a list of process ID(s) from a single sentry juju unit
for a single process name.
:param sentry_unit: Amulet sentry instance (juju unit)
:param process_name: Process name
:param expect_success: If False, expect the PID to be missing,
raise if it is present.
:returns: List of process IDs
"""
cmd = 'pidof -x "{}"'.format(process_name)
if not expect_success:
cmd += " || exit 0 && exit 1"
output, code = sentry_unit.run(cmd)
if code != 0:
msg = ('{} `{}` returned {} '
'{}'.format(sentry_unit.info['unit_name'],
cmd, code, output))
amulet.raise_status(amulet.FAIL, msg=msg)
return str(output).split()
def test_102_service_catalog(self):
"""Verify that the service catalog endpoint data is valid."""
u.log.debug('Checking keystone service catalog...')
endpoint_vol = {'adminURL': u.valid_url,
'region': 'RegionOne',
'id': u.not_null,
'publicURL': u.valid_url,
'internalURL': u.valid_url}
endpoint_id = {'adminURL': u.valid_url,
'region': 'RegionOne',
'id': u.not_null,
'publicURL': u.valid_url,
'internalURL': u.valid_url}
if self._get_openstack_release() >= self.trusty_kilo:
expected = {'compute': [endpoint_vol], 'identity': [endpoint_id]}
else:
expected = {'s3': [endpoint_vol], 'compute': [endpoint_vol],
'ec2': [endpoint_vol], 'identity': [endpoint_id]}
actual = self.keystone_demo.service_catalog.get_endpoints()
ret = u.validate_svc_catalog_endpoint_data(expected, actual)
if ret:
amulet.raise_status(amulet.FAIL, msg=ret)
def test_106_ec2_api_endpoint(self):
"""Verify the EC2 api endpoint data."""
if self._get_openstack_release() >= self.trusty_kilo:
return
u.log.debug('Checking ec2 endpoint data...')
endpoints = self.keystone.endpoints.list()
admin_port = internal_port = public_port = '8773'
expected = {
'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null
}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
message = 'EC2 endpoint: {}'.format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
def test_108_s3_api_endpoint(self):
"""Verify the S3 api endpoint data."""
if self._get_openstack_release() >= self.trusty_kilo:
return
u.log.debug('Checking s3 endpoint data...')
endpoints = self.keystone.endpoints.list()
admin_port = internal_port = public_port = '3333'
expected = {
'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null
}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
message = 'S3 endpoint: {}'.format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
def test_200_nova_shared_db_relation(self):
"""Verify the nova-compute to mysql shared-db relation data"""
u.log.debug('Checking n-c:mysql db relation data...')
unit = self.nova_compute_sentry
relation = ['shared-db', 'percona-cluster:shared-db']
expected = {
'private-address': u.valid_ip,
'nova_database': 'nova',
'nova_username': 'nova',
'nova_hostname': u.valid_ip
}
ret = u.validate_relation_data(unit, relation, expected)
if ret:
message = u.relation_error('nova-compute shared-db', ret)
amulet.raise_status(amulet.FAIL, msg=message)
def test_210_nova_cc_cloud_compute_relation(self):
"""Verify the nova-cc to nova-compute cloud-compute relation data"""
u.log.debug('Checking n-c-c:n-c cloud-compute relation data...')
unit = self.nova_cc_sentry
relation = ['cloud-compute', 'nova-compute:cloud-compute']
expected = {
'volume_service': 'cinder',
'network_manager': 'flatdhcpmanager',
'ec2_host': u.valid_ip,
'private-address': u.valid_ip,
'restart_trigger': u.not_null
}
if self._get_openstack_release() >= self.xenial_ocata:
expected['network_manager'] = 'neutron'
ret = u.validate_relation_data(unit, relation, expected)
if ret:
message = u.relation_error('nova-cc cloud-compute', ret)
amulet.raise_status(amulet.FAIL, msg=message)
def test_100_services(self):
"""Verify the expected services are running on the corresponding
service units."""
u.log.debug('Checking system services on units...')
murano_svcs = [
'murano-api', 'murano-engine'
]
service_names = {
self.murano_sentry: murano_svcs,
}
ret = u.validate_services_by_name(service_names)
if ret:
amulet.raise_status(amulet.FAIL, msg=ret)
u.log.debug('OK')
def test_110_service_catalog(self):
"""Verify that the service catalog endpoint data is valid."""
u.log.debug('Checking keystone service catalog data...')
endpoint_check = {
'adminURL': u.valid_url,
'id': u.not_null,
'region': 'RegionOne',
'publicURL': u.valid_url,
'internalURL': u.valid_url
}
expected = {
'application-catalog': [endpoint_check],
}
actual = self.keystone.service_catalog.get_endpoints()
ret = u.validate_svc_catalog_endpoint_data(expected, actual)
if ret:
amulet.raise_status(amulet.FAIL, msg=ret)
u.log.debug('OK')
def test_114_murano_api_endpoint(self):
"""Verify the murano api endpoint data."""
u.log.debug('Checking murano api endpoint data...')
endpoints = self.keystone.endpoints.list()
u.log.debug(endpoints)
admin_port = internal_port = public_port = '8082'
expected = {'id': u.not_null,
'region': 'RegionOne',
'adminurl': u.valid_url,
'internalurl': u.valid_url,
'publicurl': u.valid_url,
'service_id': u.not_null}
ret = u.validate_endpoint_data(endpoints, admin_port, internal_port,
public_port, expected)
if ret:
message = 'murano endpoint: {}'.format(ret)
amulet.raise_status(amulet.FAIL, msg=message)
u.log.debug('OK')