Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_check_complete_success_noresult(self):
"""Test check_for_completion() with successful complete job without
result."""
with requests_mock.mock() as m:
self.mock_server_1(m)
session = Session('fake-host', 'fake-user', 'fake-pw')
op_method = 'POST'
op_uri = '/api/foo'
job = Job(session, self.job_uri, op_method, op_uri)
query_job_status_result = {
'status': 'complete',
'job-status-code': 200,
# 'job-reason-code' omitted because HTTP status good
# 'job-results' is optional and is omitted
}
m.get(self.job_uri, json=query_job_status_result)
m.delete(self.job_uri, status_code=204)
job_status, op_result = job.check_for_completion()
assert job_status == 'complete'
assert op_result is None
def setup_method(self):
self.session = Session('vswitch-dpm-host', 'vswitch-user',
'vswitch-pwd')
self.client = Client(self.session)
with requests_mock.mock() as m:
# Because logon is deferred until needed, we perform it
# explicitly in order to keep mocking in the actual test simple.
m.post('/api/sessions', json={'api-session': 'vswitch-session-id'})
self.session.logon()
self.cpc_mgr = self.client.cpcs
with requests_mock.mock() as m:
result = {
'cpcs': [
{
'object-uri': '/api/cpcs/vswitch-cpc-id-1',
'name': 'CPC',
'status': 'service-required',
rt_config = zhmcclient.RetryTimeoutConfig(
connect_timeout=10,
connect_retries=1,
)
# Check HMCs and their CPCs
for cpc_name in cpc_items:
cpc_item = cpc_items[cpc_name]
hmc_host = cpc_item['hmc_host']
info(capsys, "Checking HMC %r for CPC %r", (hmc_host, cpc_name))
session = zhmcclient.Session(
hmc_host, cpc_item['hmc_userid'], cpc_item['hmc_password'],
retry_timeout_config=rt_config)
client = zhmcclient.Client(session)
try:
session.logon()
except zhmcclient.ConnectionError as exc:
info(capsys, "Skipping HMC %r for CPC %r: %s",
(hmc_host, cpc_name, exc))
continue
cpcs = client.cpcs.list()
cpc_names = [cpc.name for cpc in cpcs]
if cpc_name not in cpc_names:
raise AssertionError(
# Enable debug logging if specified
if LOG_HANDLER:
logger = logging.getLogger('zhmcclient.hmc')
if LOG_HANDLER not in logger.handlers:
logger.addHandler(LOG_HANDLER)
logger.setLevel(logging.DEBUG)
logger = logging.getLogger('zhmcclient.api')
if LOG_HANDLER not in logger.handlers:
logger.addHandler(LOG_HANDLER)
logger.setLevel(logging.DEBUG)
# Creating a session does not interact with the HMC (logon is deferred)
session = zhmcclient.Session(
hd.hmc_host, hd.hmc_userid, hd.hmc_password)
# Check access to the HMC
try:
session.logon()
except zhmcclient.Error as exc:
msg = "Cannot log on to HMC {0} at {1} due to {2}: {3}". \
format(hd.nickname, hd.hmc_host, exc.__class__.__name__, exc)
hd.skip_msg = msg
pytest.skip(msg)
hd.skip_msg = None
session.hmc_definition = hd
yield session
else:
# A test CPC is defined in the environment -> use it!
info(capsys, "Testing with CPC %r", cpc_name)
eff_rt_config = DEFAULT_RT_CONFIG
if rt_config:
eff_rt_config.override_with(rt_config)
cpc_item = hmc_creds.get_cpc_item(cpc_name)
assert cpc_item, "HMC credentials file not found: {!r}".\
format(hmc_creds.filepath)
session = zhmcclient.Session(
cpc_item['hmc_host'], cpc_item['hmc_userid'],
cpc_item['hmc_password'], retry_timeout_config=eff_rt_config)
faked_cpc = None
client = zhmcclient.Client(session)
cpc = client.cpcs.find(name=cpc_name)
return cpc_name, session, client, cpc, faked_cpc
def setup_method(self):
self.session = Session(host='fake-host', userid='fake-user',
password='fake-pw')
self.manager = MyManager(self.session)
self.resource_uri = "/api/fake-uri-1"
self.resource_name = "fake-name-1"
self.resource = MyResource(
self.manager, uri=self.resource_uri,
properties={
self.manager._name_prop: self.resource_name,
"other": "fake-other-1",
})
self.manager._list_resources = [self.resource]
def _do_parse_error_logon(self, m, json_content, exp_msg_pattern, exp_line,
exp_col):
"""
Perform a session logon, and mock the provided (invalid) JSON content
for the response so that a JSON parsing error is triggered.
Assert that this is surfaced via a `zhmcclient.ParseError` exception,
with the expected message (as a regexp pattern), line and column.
"""
m.register_uri('POST', '/api/sessions',
content=json_content,
headers={'X-Request-Id': 'fake-request-id'})
session = Session('fake-host', 'fake-user', 'fake-pw')
exp_pe_pattern = \
r"^JSON parse error in HTTP response: %s\. " \
r"HTTP request: [^ ]+ [^ ]+\. " \
r"Response status .*" % \
exp_msg_pattern
with pytest.raises(ParseError) as exc_info:
session.logon()
exc = exc_info.value
assert re.match(exp_pe_pattern, str(exc))
assert exc.line == exp_line
assert exc.column == exp_col
cpcstatus = async_operation_polling["cpcstatus"]
lparname = async_operation_polling["lparname"]
cred = hmccreds.get(hmc, None)
if cred is None:
print("Credentials for HMC %s not found in credentials file %s" % \
(hmc, hmccreds_file))
sys.exit(1)
userid = cred['userid']
password = cred['password']
print(__doc__)
print("Using HMC %s with userid %s ..." % (hmc, userid))
session = zhmcclient.Session(hmc, userid, password)
cl = zhmcclient.Client(session)
timestats = async_operation_polling.get("timestats", None)
if timestats:
session.time_stats_keeper.enable()
print("Finding CPC by name=%s and status=%s ..." % (cpcname, cpcstatus))
try:
cpc = cl.cpcs.find(name=cpcname, status=cpcstatus)
except zhmcclient.NotFound:
print("Could not find CPC %s with status %s on HMC %s" %
(cpcname, cpcstatus, hmc))
sys.exit(1)
print("Finding LPAR by name=%s ..." % lparname)
try:
loaddev = lpar_operations["loaddev"]
deactivate = lpar_operations["deactivate"]
cred = hmccreds.get(hmc, None)
if cred is None:
print("Credentials for HMC %s not found in credentials file %s" % \
(hmc, hmccreds_file))
sys.exit(1)
userid = cred['userid']
password = cred['password']
print(__doc__)
print("Using HMC %s with userid %s ..." % (hmc, userid))
session = zhmcclient.Session(hmc, userid, password)
cl = zhmcclient.Client(session)
timestats = lpar_operations.get("timestats", None)
if timestats:
session.time_stats_keeper.enable()
retries = 10
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("Found CPC %s at: %s" % (cpc.name, cpc.uri))
`Client` object using that `Session` object, and return it.
If no userid and password are specified, and if no previous call to this
method was made, userid and password are interactively inquired.
Userid and password are saved in module-global variables for future calls
to this method.
"""
global USERID, PASSWORD # pylint: disable=global-statement
USERID = userid or USERID or \
six.input('Enter userid for HMC {}: '.format(zhmc))
PASSWORD = password or PASSWORD or \
getpass.getpass('Enter password for {}: '.format(USERID))
session = zhmcclient.Session(zhmc, USERID, PASSWORD)
session.logon()
client = zhmcclient.Client(session)
print('Established logged-on session with HMC {} using userid {}'.
format(zhmc, USERID))
return client