# # 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 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()