我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用amulet.raise_status()。
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 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_106_swift_object_store_endpoint(self): """Verify the swift object-store endpoint data.""" u.log.debug('Checking keystone endpoint for swift object store...') endpoints = self.keystone.endpoints.list() admin_port = internal_port = public_port = '8080' 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 = 'object-store endpoint: {}'.format(ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_200_swift_proxy_identity_service_relation(self): """Verify the swift-proxy to keystone identity relation data.""" u.log.debug('Checking swift-proxy:keystone identity relation...') unit = self.swift_proxy_sentry relation = ['identity-service', 'keystone:identity-service'] expected = { 'swift_service': 'swift', 'swift_region': 'RegionOne', 'swift_public_url': u.valid_url, 'swift_internal_url': u.valid_url, 'swift_admin_url': u.valid_url, 's3_service': 's3', 's3_region': 'RegionOne', 's3_public_url': u.valid_url, 's3_internal_url': u.valid_url, 's3_admin_url': u.valid_url, 'private-address': u.valid_ip, } ret = u.validate_relation_data(unit, relation, expected) if ret: message = u.relation_error('swift-proxy identity-service', ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_202_keystone_identity_service_relation(self): """Verify the keystone to swift-proxy identity relation data.""" u.log.debug('Checking keystone:swift-proxy identity relation...') unit = self.keystone_sentry relation = ['identity-service', 'swift-proxy: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': 's3_swift', 'service_tenant_id': u.not_null, 'service_host': u.valid_ip } 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_204_swift_storage_swift_storage_relation(self): """Verify the swift-storage to swift-proxy swift-storage relation data.""" u.log.debug('Checking swift:swift-proxy swift-storage relation...') unit = self.swift_storage_sentry relation = ['swift-storage', 'swift-proxy:swift-storage'] expected = { 'account_port': '6002', 'zone': '1', 'object_port': '6000', 'container_port': '6001', 'private-address': u.valid_ip, 'device': 'vdb' } ret = u.validate_relation_data(unit, relation, expected) if ret: message = u.relation_error('swift-storage swift-storage', ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_206_swift_proxy_swift_storage_relation(self): """Verify the swift-proxy to swift-storage swift-storage relation data.""" u.log.debug('Checking swift-proxy:swift swift-storage relation...') unit = self.swift_proxy_sentry relation = ['swift-storage', 'swift-storage:swift-storage'] expected = { 'private-address': u.valid_ip, 'trigger': u.not_null, 'rings_url': u.valid_url, 'swift_hash': u.not_null } ret = u.validate_relation_data(unit, relation, expected) if ret: message = u.relation_error('swift-proxy swift-storage', ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_300_swift_config(self): """Verify the data in the swift-hash section of the swift config file.""" u.log.debug('Checking swift config...') unit = self.swift_storage_sentry conf = '/etc/swift/swift.conf' swift_proxy_relation = self.swift_proxy_sentry.relation( 'swift-storage', 'swift-storage:swift-storage') expected = { 'swift_hash_path_suffix': swift_proxy_relation['swift_hash'] } ret = u.validate_config_data(unit, conf, 'swift-hash', expected) if ret: message = "swift config error: {}".format(ret) amulet.raise_status(amulet.FAIL, msg=message)
def _test_pause(self): u.log.info("Testing pause action") self._assert_services(should_run=True) pause_action_id = u.run_action(self.swift_proxy_sentry, "pause") assert u.wait_on_action(pause_action_id), "Pause action failed." self._assert_services(should_run=False) status, message = u.status_get(self.swift_proxy_sentry) if status != "maintenance": msg = ("Pause action failed to move unit to maintenance " "status (got {} instead)".format(status)) amulet.raise_status(amulet.FAIL, msg=msg) if message != "Paused. Use 'resume' action to resume normal service.": msg = ("Pause action failed to set message" " (got {} instead)".format(message)) amulet.raise_status(amulet.FAIL, msg=msg)
def _test_resume(self): u.log.info("Testing resume action") # service is left paused by _test_pause self._assert_services(should_run=False) resume_action_id = u.run_action(self.swift_proxy_sentry, "resume") assert u.wait_on_action(resume_action_id), "Resume action failed." self._assert_services(should_run=True) status, message = u.status_get(self.swift_proxy_sentry) if status != "active": msg = ("Resume action failed to move unit to active " "status (got {} instead)".format(status)) amulet.raise_status(amulet.FAIL, msg=msg) if message != "Unit is ready": msg = ("Resume action failed to clear message" " (got {} instead)".format(message)) amulet.raise_status(amulet.FAIL, msg=msg)
def _image_create(self): """Create an image to be used by the heat template, verify it exists""" u.log.debug('Creating glance image ({})...'.format(IMAGE_NAME)) # Create a new image image_new = u.create_cirros_image(self.glance, IMAGE_NAME) # Confirm image is created and has status of 'active' if not image_new: message = 'glance image create failed' amulet.raise_status(amulet.FAIL, msg=message) # Verify new image name images_list = list(self.glance.images.list()) if images_list[0].name != IMAGE_NAME: message = ('glance image create failed or unexpected ' 'image name {}'.format(images_list[0].name)) amulet.raise_status(amulet.FAIL, msg=message)
def test_110_service_catalog(self): """Expect certain endpoints and endpoint data to be present in the Keystone service catalog""" u.log.debug('Checking service catalog endpoint data...') ep_validate = {'adminURL': u.valid_url, 'region': 'RegionOne', 'publicURL': u.valid_url, 'internalURL': u.valid_url, 'id': u.not_null} expected = {'compute': [ep_validate], 'orchestration': [ep_validate], 'image': [ep_validate], 'identity': [ep_validate]} 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)
def test_120_heat_endpoint(self): """Verify the heat api endpoint data.""" u.log.debug('Checking api endpoint data...') endpoints = self.keystone.endpoints.list() if self._get_openstack_release() < self.trusty_kilo: # Before Kilo admin_port = internal_port = public_port = '3333' else: # Kilo and later admin_port = internal_port = public_port = '8004' 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 = 'heat endpoint: {}'.format(ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_200_heat_mysql_shared_db_relation(self): """Verify the heat:mysql shared-db relation data""" u.log.debug('Checking heat:mysql shared-db relation data...') unit = self.heat_sentry relation = ['shared-db', 'percona-cluster:shared-db'] expected = { 'private-address': u.valid_ip, 'heat_database': 'heat', 'heat_username': 'heat', 'heat_hostname': u.valid_ip } ret = u.validate_relation_data(unit, relation, expected) if ret: message = u.relation_error('heat:mysql shared-db', ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_201_mysql_heat_shared_db_relation(self): """Verify the mysql:heat shared-db relation data""" u.log.debug('Checking mysql:heat shared-db relation data...') unit = self.pxc_sentry relation = ['shared-db', 'heat:shared-db'] expected = { 'private-address': u.valid_ip, 'db_host': u.valid_ip, 'heat_allowed_units': u.not_null, 'heat_password': u.not_null } ret = u.validate_relation_data(unit, relation, expected) if ret: message = u.relation_error('percona-cluster:heat shared-db', ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_203_keystone_heat_identity_relation(self): """Verify the keystone:heat identity-service relation data""" u.log.debug('Checking keystone:heat identity-service relation data...') unit = self.keystone_sentry relation = ['identity-service', 'heat: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': 'heat-cfn_heat', 'service_tenant_id': u.not_null, 'service_host': u.valid_ip } ret = u.validate_relation_data(unit, relation, expected) if ret: message = u.relation_error('keystone:heat identity-service', ret) amulet.raise_status(amulet.FAIL, msg=message)
def test_500_auth_encryption_key_same_on_units(self): """Test that the auth_encryption_key in heat.conf is the same on all of the units. """ u.log.debug("Checking the 'auth_encryption_key' is the same on " "all units.") output, ret = self._run_arbitrary( "--application heat " "--format json " "grep auth_encryption_key /etc/heat/heat.conf") if ret: msg = "juju run returned error: ({}) -> {}".format(ret, output) amulet.raise_status(amulet.FAIL, msg=msg) output = json.loads(output) keys = {} for r in output: k = r['Stdout'].split('=')[1].strip() keys[r['UnitId']] = k # see if keys are different. ks = keys.values() if any(((k != ks[0]) for k in ks[1:])): msg = ("'auth_encryption_key' is not identical on every unit: {}" .format("{}={}".format(k, v) for k, v in keys.items())) amulet.raise_status(amulet.FAIL, msg=msg)
def validate_sectionless_conf(self, file_contents, expected): """A crude conf parser. Useful to inspect configuration files which do not have section headers (as would be necessary in order to use the configparser). Such as openstack-dashboard or rabbitmq confs.""" for line in file_contents.split('\n'): if '=' in line: args = line.split('=') if len(args) <= 1: continue key = args[0].strip() value = args[1].strip() if key in expected.keys(): if expected[key] != value: msg = ('Config mismatch. Expected, actual: {}, ' '{}'.format(expected[key], value)) amulet.raise_status(amulet.FAIL, msg=msg)
def validate_keystone_tenants(self, client): """Verify all existing tenants.""" u.log.debug('Checking keystone tenants...') expected = [ {'name': 'services', 'enabled': True, 'description': 'Created by Juju', 'id': u.not_null}, {'name': 'demoTenant', 'enabled': True, 'description': 'demo tenant', 'id': u.not_null}, {'name': 'admin', 'enabled': True, 'description': 'Created by Juju', 'id': u.not_null} ] if self.keystone_api_version == 2: actual = client.tenants.list() else: actual = client.projects.list() ret = u.validate_tenant_data(expected, actual) if ret: amulet.raise_status(amulet.FAIL, msg=ret)
def test_140_keystone_endpoint(self): """Verify the keystone endpoint data.""" u.log.debug('Checking keystone api endpoint data...') endpoints = self.keystone_v2.endpoints.list() admin_port = '35357' internal_port = public_port = '5000' 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='keystone endpoint: {}'.format(ret))
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))
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 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_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)
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_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 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')
