Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Input properties for a user with the same name
sn_user_props = {
'name': user_name,
'description': 'User with same name',
'type': 'standard',
'authentication-type': 'local',
}
user_mgr = self.console.users
user = user_mgr.find(name=user_name)
# Execute the deletion code to be tested
user.delete()
# Check that the user no longer exists
with pytest.raises(NotFound):
user_mgr.find(name=user_name)
# Execute the creation code to be tested.
user_mgr.create(sn_user_props)
# Check that the user exists again under that name
sn_user = user_mgr.find(name=user_name)
description = sn_user.get_property('description')
assert description == sn_user_props['description']
def test_adapter_delete(self):
"""Test Adapter.delete() for Hipersocket adapter."""
# Add two faked adapters
self.add_standard_osa()
faked_hs = self.add_standard_hipersocket()
adapter_mgr = self.cpc.adapters
hs_adapter = adapter_mgr.find(name=faked_hs.name)
# Execute the code to be tested
hs_adapter.delete()
with pytest.raises(NotFound):
hs_adapter = adapter_mgr.find(type='hipersockets')
with pytest.raises(NotFound):
hs_adapter = adapter_mgr.find(name=faked_hs.name)
adapters = adapter_mgr.list()
assert len(adapters) == 1
with pytest.raises(HTTPError) as exc_info:
hs_adapter.pull_full_properties()
exc = exc_info.value
assert exc.http_status == 404
assert exc.reason == 1
def test_crud(self, capsys):
"""Create, read, update and delete a partition."""
cpc_name, session, client, cpc, faked_cpc = \
setup_cpc(capsys, self.hmc_creds, self.fake_data)
part_name = self.NAME_PREFIX + 'test_crud.part1'
# Ensure a clean starting point for this test
try:
part = cpc.partitions.find(name=part_name)
except zhmcclient.NotFound:
pass
else:
info(capsys, "Cleaning up partition from previous run: {!r}".
format(part))
status = part.get_property('status')
if status != 'stopped':
part.stop()
part.delete()
# Test creating the partition
part_input_props = {
'name': part_name,
'description': 'Dummy partition description.',
'ifl-processors': 2,
'initial-memory': 1024,
partition = partition_mgr.find(name=partition_name)
new_partition_name = "new-" + partition_name
# Execute the code to be tested
partition.update_properties(properties={'name': new_partition_name})
# Verify that the resource is no longer found by its old name, using
# list() (this does not use the name-to-URI cache).
partitions_list = partition_mgr.list(
filter_args=dict(name=partition_name))
assert len(partitions_list) == 0
# Verify that the resource is no longer found by its old name, using
# find() (this uses the name-to-URI cache).
with pytest.raises(NotFound):
partition_mgr.find(name=partition_name)
# Verify that the resource object already reflects the update, even
# though it has not been refreshed yet.
assert partition.properties['name'] == new_partition_name
# Refresh the resource object and verify that it still reflects the
# update.
partition.pull_full_properties()
assert partition.properties['name'] == new_partition_name
# Verify that the resource can be found by its new name, using find()
new_partition_find = partition_mgr.find(name=new_partition_name)
assert new_partition_find.properties['name'] == new_partition_name
# Verify that the resource can be found by its new name, using list()
def test_notfound_repr(self, filter_args):
"""All tests for NotFound.__repr__()."""
cpc = self.client.cpcs.find(name='cpc_1')
manager = cpc.adapters
exc = NotFound(filter_args, manager)
classname = exc.__class__.__name__
# Execute the code to be tested
repr_str = repr(exc)
# We check the one-lined string just roughly
repr_str = repr_str.replace('\n', '\\n')
assert re.match(r'^{}\s*\(.*\)$'.format(classname), repr_str)
exc = exc_info.value
if isinstance(exp_exc, HTTPError):
assert exc.http_status == exp_exc.http_status
assert exc.reason == exp_exc.reason
# Check that the LDAP Server Definition still exists
ldap_srv_def_mgr.find(name=faked_ldap_srv_def.name)
else:
# Execute the code to be tested.
ldap_srv_def.delete()
# Check that the LDAP Server Definition no longer exists
with pytest.raises(NotFound) as exc_info:
ldap_srv_def_mgr.find(name=faked_ldap_srv_def.name)
exc = exc_info.value
if isinstance(exp_exc, HTTPError):
assert exc.http_status == exp_exc.http_status
assert exc.reason == exp_exc.reason
# Check that the HBA still exists
hba_mgr.find(name=faked_hba.name)
else:
# Execute the code to be tested.
hba.delete()
# Check that the HBA no longer exists
with pytest.raises(NotFound) as exc_info:
hba_mgr.find(name=faked_hba.name)
session = zhmcclient.Session(hmc, userid, password)
cl = zhmcclient.Client(session)
timestats = activation_profiles.get("timestats", False)
if timestats:
session.time_stats_keeper.enable()
print("Listing CPCs ...")
cpcs = cl.cpcs.list()
for cpc in cpcs:
print(cpc.name, cpc.get_property('status'), cpc.uri)
print("Finding CPC by name=%s ..." % cpcname)
try:
cpc = cl.cpcs.find(name=cpcname)
except zhmcclient.NotFound:
print("Could not find CPC %s on HMC %s" % (cpcname, hmc))
sys.exit(1)
print("Checking if DPM is enabled on CPC %s..." % cpcname)
if cpc.dpm_enabled:
print("CPC %s is in DPM mode." % cpcname)
sys.exit(1)
managers = {'reset': 'reset_activation_profiles',
'image' : 'image_activation_profiles',
'load' : 'load_activation_profiles'}
for profile_type, manager in managers.items():
profiles = getattr(cpc, manager).list()
print("Listing %d %s Activation Profiles ..."
client = zhmcclient.Client(cmd_ctx.session)
partition = find_partition(cmd_ctx, client, cpc_name, partition_name)
name_map = {
# The following options are handled in this function:
'adapter': None,
'port': None,
}
options = original_options(options)
properties = options_to_properties(options, name_map)
adapter_name = options['adapter']
try:
adapter = partition.manager.cpc.adapters.find(name=adapter_name)
except zhmcclient.NotFound:
raise_click_exception("Could not find adapter %s in CPC %s." %
(adapter_name, cpc_name), cmd_ctx.error_format)
port_name = options['port']
try:
port = adapter.ports.find(name=port_name)
except zhmcclient.NotFound:
raise_click_exception("Could not find port %s on adapter %s in "
"CPC %s." % (port_name, adapter_name, cpc_name),
cmd_ctx.error_format)
properties['adapter-port-uri'] = port.uri
try:
new_hba = partition.hbas.create(properties)
except zhmcclient.Error as exc:
def cmd_vfunction_create(cmd_ctx, cpc_name, partition_name, options):
client = zhmcclient.Client(cmd_ctx.session)
partition = find_partition(cmd_ctx, client, cpc_name, partition_name)
name_map = {
# The following options are handled in this function:
'adapter': None,
}
options = original_options(options)
properties = options_to_properties(options, name_map)
adapter_name = options['adapter']
try:
adapter = partition.manager.cpc.adapters.find(name=adapter_name)
except zhmcclient.NotFound:
raise_click_exception("Could not find adapter %s in CPC %s." %
(adapter_name, cpc_name), cmd_ctx.error_format)
properties['adapter-uri'] = adapter.uri
try:
new_vfunction = partition.virtual_functions.create(properties)
except zhmcclient.Error as exc:
raise_click_exception(exc, cmd_ctx.error_format)
cmd_ctx.spinner.stop()
click.echo("New virtual function %s has been created." %
new_vfunction.properties['name'])