python-dracclient/dracclient/tests/test_job.py
Sean McGinnis 1f2b0ed263
Use unittest.mock instead of third party mock
Now that we no longer support py27, we can use the standard library
unittest.mock module instead of the third party mock lib.

Change-Id: I1e30d77b7ca98c08ec8c42a18df863b202525418
Signed-off-by: Sean McGinnis <sean.mcginnis@gmail.com>
2020-04-28 07:46:23 -05:00

574 lines
26 KiB
Python

#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import lxml.etree
from unittest import mock
import requests_mock
import dracclient.client
from dracclient import exceptions
import dracclient.resources.job
from dracclient.resources import uris
from dracclient.tests import base
from dracclient.tests import utils as test_utils
from dracclient import utils
class ClientJobManagementTestCase(base.BaseTest):
def setUp(self):
super(ClientJobManagementTestCase, self).setUp()
self.drac_client = dracclient.client.DRACClient(
**test_utils.FAKE_ENDPOINT)
@requests_mock.Mocker()
@mock.patch.object(dracclient.client.WSManClient,
'wait_until_idrac_is_ready', spec_set=True,
autospec=True)
def test_list_jobs(self, mock_requests, mock_wait_until_idrac_is_ready):
mock_requests.post(
'https://1.2.3.4:443/wsman',
text=test_utils.JobEnumerations[uris.DCIM_LifecycleJob]['ok'])
jobs = self.drac_client.list_jobs()
self.assertEqual(6, len(jobs))
@mock.patch.object(dracclient.client.WSManClient, 'enumerate',
spec_set=True, autospec=True)
def test_list_jobs_only_unfinished(self, mock_enumerate):
expected_filter_query = ('select * from DCIM_LifecycleJob '
'where Name != "CLEARALL" and '
'JobStatus != "Reboot Completed" and '
'JobStatus != "Reboot Failed" and '
'JobStatus != "Completed" and '
'JobStatus != "Completed with Errors" and '
'JobStatus != "Failed"')
mock_enumerate.return_value = lxml.etree.fromstring(
test_utils.JobEnumerations[uris.DCIM_LifecycleJob]['ok'])
self.drac_client.list_jobs(only_unfinished=True)
mock_enumerate.assert_called_once_with(
mock.ANY, uris.DCIM_LifecycleJob,
filter_query=expected_filter_query)
@mock.patch.object(dracclient.client.WSManClient, 'enumerate',
spec_set=True, autospec=True)
def test_get_job(self, mock_enumerate):
expected_filter_query = ('select * from DCIM_LifecycleJob'
' where InstanceID="42"')
# NOTE: This is the first job in the xml. Filtering the job is the
# responsibility of the controller, so not testing it.
expected_job = dracclient.resources.job.Job(id='JID_CLEARALL',
name='CLEARALL',
start_time='TIME_NA',
until_time='TIME_NA',
message='NA',
status='Pending',
percent_complete='0')
mock_enumerate.return_value = lxml.etree.fromstring(
test_utils.JobEnumerations[uris.DCIM_LifecycleJob]['ok'])
job = self.drac_client.get_job(42)
mock_enumerate.assert_called_once_with(
mock.ANY, uris.DCIM_LifecycleJob,
filter_query=expected_filter_query)
self.assertEqual(expected_job, job)
@mock.patch.object(dracclient.client.WSManClient, 'enumerate',
spec_set=True, autospec=True)
def test_get_job_not_found(self, mock_enumerate):
expected_filter_query = ('select * from DCIM_LifecycleJob'
' where InstanceID="42"')
mock_enumerate.return_value = lxml.etree.fromstring(
test_utils.JobEnumerations[uris.DCIM_LifecycleJob]['not_found'])
job = self.drac_client.get_job(42)
mock_enumerate.assert_called_once_with(
mock.ANY, uris.DCIM_LifecycleJob,
filter_query=expected_filter_query)
self.assertIsNone(job)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_delete_jobs_all(self, mock_invoke):
expected_selectors = {'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'idrac',
'CreationClassName': 'DCIM_JobService',
'Name': 'JobService'}
expected_properties = {'JobID': 'JID_CLEARALL'}
self.drac_client.delete_jobs()
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_JobService, 'DeleteJobQueue',
expected_selectors, expected_properties,
expected_return_value=utils.RET_SUCCESS)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_delete_jobs_force(self, mock_invoke):
expected_selectors = {'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'idrac',
'CreationClassName': 'DCIM_JobService',
'Name': 'JobService'}
expected_properties = {'JobID': 'JID_CLEARALL_FORCE'}
self.drac_client.delete_jobs(['JID_CLEARALL_FORCE'])
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_JobService, 'DeleteJobQueue',
expected_selectors, expected_properties,
expected_return_value=utils.RET_SUCCESS)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_delete_jobs_one(self, mock_invoke):
expected_selectors = {'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'idrac',
'CreationClassName': 'DCIM_JobService',
'Name': 'JobService'}
expected_properties = {'JobID': 'JID_442507917525'}
self.drac_client.delete_jobs(['JID_442507917525'])
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_JobService, 'DeleteJobQueue',
expected_selectors, expected_properties,
expected_return_value=utils.RET_SUCCESS)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_delete_jobs_multi(self, mock_invoke):
expected_selectors = {'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'idrac',
'CreationClassName': 'DCIM_JobService',
'Name': 'JobService'}
self.drac_client.delete_jobs(['JID_442507917525',
'JID_442507917526'])
calls_expected = [
mock.call(mock.ANY,
uris.DCIM_JobService,
'DeleteJobQueue',
expected_selectors,
{'JobID': 'JID_442507917525'},
expected_return_value=utils.RET_SUCCESS),
mock.call(mock.ANY,
uris.DCIM_JobService,
'DeleteJobQueue',
expected_selectors,
{'JobID': 'JID_442507917526'},
expected_return_value=utils.RET_SUCCESS)]
mock_invoke.assert_has_calls(calls_expected)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_delete_jobs_none(self, mock_invoke):
self.drac_client.delete_jobs(None)
self.assertFalse(mock_invoke.called)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_delete_jobs_empty_list(self, mock_invoke):
self.drac_client.delete_jobs([])
self.assertFalse(mock_invoke.called)
@requests_mock.Mocker()
@mock.patch.object(dracclient.client.WSManClient,
'wait_until_idrac_is_ready', spec_set=True,
autospec=True)
def test_delete_job_not_found(
self, mock_requests,
mock_wait_until_idrac_is_ready):
mock_requests.post(
'https://1.2.3.4:443/wsman',
text=test_utils.JobService[uris.DCIM_JobService][
'DeleteJobQueue']['error'])
self.assertRaises(
exceptions.DRACOperationFailed,
self.drac_client.delete_jobs,
['JID_1234'])
@requests_mock.Mocker()
@mock.patch.object(dracclient.client.WSManClient,
'wait_until_idrac_is_ready', spec_set=True,
autospec=True)
def test_delete_some_jobs_not_found(
self, mock_requests,
mock_wait_until_idrac_is_ready):
mock_requests.post(
'https://1.2.3.4:443/wsman',
[{'text': test_utils.JobService[uris.DCIM_JobService][
'DeleteJobQueue']['error']},
{'text': test_utils.JobService[uris.DCIM_JobService][
'DeleteJobQueue']['ok']}])
self.assertRaises(
exceptions.DRACOperationFailed,
self.drac_client.delete_jobs,
['JID_1234', 'JID_442507917525'])
self.assertEqual(mock_requests.call_count, 2)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_create_config_job_for_lifecycle(self, mock_invoke):
cim_creation_class_name = 'DCIM_LCService'
cim_name = 'DCIM:LCService'
target = ''
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target,
'ScheduledStartTime': 'TIME_NOW'}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_LCService][
'CreateConfigJob']['ok'])
job_id = self.drac_client.create_config_job(
uris.DCIM_LCService, cim_creation_class_name, cim_name, target,
start_time='TIME_NOW',
wait_for_idrac=False, method_name='CreateConfigJob')
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_LCService, 'CreateConfigJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED,
wait_for_idrac=False)
self.assertEqual('JID_442507917525', job_id)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_create_config_job(self, mock_invoke):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
wait_for_idrac = True
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target,
'ScheduledStartTime': 'TIME_NOW'}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_BIOSService][
'CreateTargetedConfigJob']['ok'])
job_id = self.drac_client.create_config_job(
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_BIOSService, 'CreateTargetedConfigJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED,
wait_for_idrac=wait_for_idrac)
self.assertEqual('JID_442507917525', job_id)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_create_config_job_with_start_time(self, mock_invoke):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
start_time = "20140924120105"
wait_for_idrac = True
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target,
'ScheduledStartTime': start_time}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_BIOSService][
'CreateTargetedConfigJob']['ok'])
job_id = self.drac_client.create_config_job(
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target,
start_time=start_time)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_BIOSService, 'CreateTargetedConfigJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED,
wait_for_idrac=wait_for_idrac)
self.assertEqual('JID_442507917525', job_id)
@mock.patch.object(dracclient.client.WSManClient, 'invoke',
spec_set=True, autospec=True)
def test_create_config_job_with_no_start_time(self, mock_invoke):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
start_time = None
wait_for_idrac = True
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_BIOSService][
'CreateTargetedConfigJob']['ok'])
job_id = self.drac_client.create_config_job(
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target,
start_time=start_time)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_BIOSService, 'CreateTargetedConfigJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED,
wait_for_idrac=wait_for_idrac)
self.assertEqual('JID_442507917525', job_id)
@requests_mock.Mocker()
@mock.patch.object(dracclient.client.WSManClient,
'wait_until_idrac_is_ready', spec_set=True,
autospec=True)
def test_create_config_job_failed(self, mock_requests,
mock_wait_until_idrac_is_ready):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
mock_requests.post(
'https://1.2.3.4:443/wsman',
text=test_utils.JobInvocations[uris.DCIM_BIOSService][
'CreateTargetedConfigJob']['error'])
self.assertRaises(
exceptions.DRACOperationFailed, self.drac_client.create_config_job,
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target)
@requests_mock.Mocker()
@mock.patch.object(dracclient.client.WSManClient,
'wait_until_idrac_is_ready', spec_set=True,
autospec=True)
def test_create_config_job_for_lifecycle_failed(
self, mock_requests,
mock_wait_until_idrac_is_ready):
cim_creation_class_name = 'DCIM_LCService'
cim_name = 'DCIM:LCService'
target = ''
mock_requests.post(
'https://1.2.3.4:443/wsman',
text=test_utils.JobInvocations[uris.DCIM_LCService][
'CreateConfigJob']['error'])
self.assertRaises(
exceptions.DRACOperationFailed, self.drac_client.create_config_job,
uris.DCIM_LCService, cim_creation_class_name, cim_name, target)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_create_config_job_with_reboot(self, mock_invoke):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
wait_for_idrac = True
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target,
'RebootJobType': '3',
'ScheduledStartTime': 'TIME_NOW'}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_BIOSService][
'CreateTargetedConfigJob']['ok'])
job_id = self.drac_client.create_config_job(
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target,
reboot=True, realtime=False)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_BIOSService, 'CreateTargetedConfigJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED,
wait_for_idrac=wait_for_idrac)
self.assertEqual('JID_442507917525', job_id)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_create_config_job_with_realtime(self, mock_invoke):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
wait_for_idrac = True
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target,
'ScheduledStartTime': 'TIME_NOW',
'RealTime': '1'}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_BIOSService][
'CreateTargetedConfigJob']['ok'])
job_id = self.drac_client.create_config_job(
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target,
reboot=False, realtime=True)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_BIOSService, 'CreateTargetedConfigJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED,
wait_for_idrac=wait_for_idrac)
self.assertEqual('JID_442507917525', job_id)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_create_reboot_job(self, mock_invoke):
expected_selectors = {
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'idrac',
'CreationClassName': 'DCIM_JobService',
'Name': 'JobService'}
expected_properties = {'RebootJobType': '3'}
self.drac_client.create_reboot_job()
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_JobService, 'CreateRebootJob',
expected_selectors, expected_properties,
expected_return_value=utils.RET_CREATED)
def test_create_reboot_job_bad_type(self):
self.assertRaises(
exceptions.InvalidParameterValue,
self.drac_client.create_reboot_job, 'BAD REBOOT TYPE')
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_delete_pending_config(self, mock_invoke):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
expected_selectors = {'CreationClassName': cim_creation_class_name,
'Name': cim_name,
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'DCIM:ComputerSystem'}
expected_properties = {'Target': target}
mock_invoke.return_value = lxml.etree.fromstring(
test_utils.JobInvocations[uris.DCIM_BIOSService][
'DeletePendingConfiguration']['ok'])
self.drac_client.delete_pending_config(
uris.DCIM_BIOSService, cim_creation_class_name, cim_name, target)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_BIOSService, 'DeletePendingConfiguration',
expected_selectors, expected_properties,
expected_return_value=utils.RET_SUCCESS)
@requests_mock.Mocker()
@mock.patch.object(dracclient.client.WSManClient,
'wait_until_idrac_is_ready', spec_set=True,
autospec=True)
def test_delete_pending_config_failed(self, mock_requests,
mock_wait_until_idrac_is_ready):
cim_creation_class_name = 'DCIM_BIOSService'
cim_name = 'DCIM:BIOSService'
target = 'BIOS.Setup.1-1'
mock_requests.post(
'https://1.2.3.4:443/wsman',
text=test_utils.JobInvocations[uris.DCIM_BIOSService][
'DeletePendingConfiguration']['error'])
self.assertRaises(
exceptions.DRACOperationFailed,
self.drac_client.delete_pending_config, uris.DCIM_BIOSService,
cim_creation_class_name, cim_name, target)
class ClientJobScheduleTestCase(base.BaseTest):
def setUp(self):
super(ClientJobScheduleTestCase, self).setUp()
self.drac_client = dracclient.client.DRACClient(
**test_utils.FAKE_ENDPOINT)
def _test_schedule_job_execution(self,
mock_invoke,
job_ids,
start_time,
expected_properties):
expected_selectors = {
'SystemCreationClassName': 'DCIM_ComputerSystem',
'SystemName': 'idrac',
'CreationClassName': 'DCIM_JobService',
'Name': 'JobService'}
if start_time is None:
self.drac_client.schedule_job_execution(job_ids)
else:
self.drac_client.schedule_job_execution(job_ids, start_time)
mock_invoke.assert_called_once_with(
mock.ANY, uris.DCIM_JobService, 'SetupJobQueue',
expected_selectors, expected_properties,
expected_return_value=utils.RET_SUCCESS)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_schedule_job_execution_one_job(self, mock_invoke):
job_ids = ['JID_442507917525']
expected_properties = {'StartTimeInterval': 'TIME_NOW',
'JobArray': job_ids}
self._test_schedule_job_execution(mock_invoke, job_ids, None,
expected_properties)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_schedule_job_execution_multi_job(self, mock_invoke):
job_ids = ['JID_442507917525', 'JID_442507917526']
expected_properties = {'StartTimeInterval': 'TIME_NOW',
'JobArray': job_ids}
self._test_schedule_job_execution(mock_invoke, job_ids, None,
expected_properties)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_schedule_job_execution_one_job_with_time(self, mock_invoke):
job_ids = ['JID_442507917525']
timestamp = datetime.datetime.today().strftime('%Y%m%d%H%M%S')
expected_properties = {'StartTimeInterval': timestamp,
'JobArray': job_ids}
self._test_schedule_job_execution(mock_invoke, job_ids, timestamp,
expected_properties)
@mock.patch.object(dracclient.client.WSManClient, 'invoke', spec_set=True,
autospec=True)
def test_schedule_job_execution_multi_job_with_time(self, mock_invoke):
job_ids = ['JID_442507917525', 'JID_442507917526']
timestamp = datetime.datetime.today().strftime('%Y%m%d%H%M%S')
expected_properties = {'StartTimeInterval': timestamp,
'JobArray': job_ids}
self._test_schedule_job_execution(mock_invoke, job_ids, timestamp,
expected_properties)
@mock.patch.object(dracclient.client.WSManClient, 'invoke')
def test_schedule_job_execution_no_jobs(self, mock_invoke):
self.drac_client.schedule_job_execution(job_ids=[])
mock_invoke.assert_not_called()