python类raise_status()的实例源码

basic_deployment.py 文件源码 项目:charm-keystone 作者: openstack 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
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))
basic_deployment.py 文件源码 项目:charm-keystone 作者: openstack 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
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)
basic_deployment.py 文件源码 项目:charm-keystone 作者: openstack 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
utils.py 文件源码 项目:charm-keystone 作者: openstack 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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)
utils.py 文件源码 项目:charm-nova-cloud-controller 作者: openstack 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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
utils.py 文件源码 项目:charm-nova-compute 作者: openstack 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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
utils.py 文件源码 项目:charm-nova-compute 作者: openstack 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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()
basic_deployment.py 文件源码 项目:charm-nova-compute 作者: openstack 项目源码 文件源码 阅读 17 收藏 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-compute 作者: openstack 项目源码 文件源码 阅读 17 收藏 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-compute 作者: openstack 项目源码 文件源码 阅读 17 收藏 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-compute 作者: openstack 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
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)
basic_deployment.py 文件源码 项目:charm-nova-compute 作者: openstack 项目源码 文件源码 阅读 15 收藏 0 点赞 0 评论 0
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)
basic_deployment.py 文件源码 项目:charm-murano 作者: openstack 项目源码 文件源码 阅读 15 收藏 0 点赞 0 评论 0
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')
basic_deployment.py 文件源码 项目:charm-murano 作者: openstack 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
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')
basic_deployment.py 文件源码 项目:charm-murano 作者: openstack 项目源码 文件源码 阅读 15 收藏 0 点赞 0 评论 0
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')


问题


面经


文章

微信
公众号

扫码关注公众号