How to use is_multi_tenant method in tempest

Best Python code snippet using tempest_python

test_wdc_redfish_command.py

Source:test_wdc_redfish_command.py Github

copy

Full Screen

1# -*- coding: utf-8 -*-2# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)3from __future__ import (absolute_import, division, print_function)4__metaclass__ = type5import shutil6import uuid7import tarfile8import tempfile9import os10from ansible_collections.community.general.tests.unit.compat.mock import patch11from ansible_collections.community.general.tests.unit.compat import unittest12from ansible.module_utils import basic13import ansible_collections.community.general.plugins.modules.remote_management.redfish.wdc_redfish_command as module14from ansible_collections.community.general.tests.unit.plugins.modules.utils import AnsibleExitJson, AnsibleFailJson15from ansible_collections.community.general.tests.unit.plugins.modules.utils import set_module_args, exit_json, fail_json16MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE = {17 "ret": True,18 "data": {19 }20}21MOCK_GET_ENCLOSURE_RESPONSE_SINGLE_TENANT = {22 "ret": True,23 "data": {24 "SerialNumber": "12345"25 }26}27MOCK_GET_ENCLOSURE_RESPONSE_MULTI_TENANT = {28 "ret": True,29 "data": {30 "SerialNumber": "12345-A"31 }32}33MOCK_URL_ERROR = {34 "ret": False,35 "msg": "This is a mock URL error",36 "status": 50037}38MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE = {39 "ret": True,40 "data": {41 "UpdateService": {42 "@odata.id": "/UpdateService"43 }44 }45}46MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_AND_FW_ACTIVATE = {47 "ret": True,48 "data": {49 "Actions": {50 "#UpdateService.SimpleUpdate": {51 "target": "mocked value"52 },53 "Oem": {54 "WDC": {55 "#UpdateService.FWActivate": {56 "title": "Activate the downloaded firmware.",57 "target": "/redfish/v1/UpdateService/Actions/UpdateService.FWActivate"58 }59 }60 }61 }62 }63}64MOCK_SUCCESSFUL_RESPONSE_WITH_ACTIONS = {65 "ret": True,66 "data": {67 "Actions": {}68 }69}70MOCK_GET_IOM_A_MULTI_TENANT = {71 "ret": True,72 "data": {73 "Id": "IOModuleAFRU"74 }75}76MOCK_GET_IOM_B_MULTI_TENANAT = {77 "ret": True,78 "data": {79 "error": {80 "message": "IOM Module B cannot be read"81 }82 }83}84MOCK_READY_FOR_FW_UPDATE = {85 "ret": True,86 "entries": {87 "Description": "Ready for FW update",88 "StatusCode": 089 }90}91MOCK_FW_UPDATE_IN_PROGRESS = {92 "ret": True,93 "entries": {94 "Description": "FW update in progress",95 "StatusCode": 196 }97}98MOCK_WAITING_FOR_ACTIVATION = {99 "ret": True,100 "entries": {101 "Description": "FW update completed. Waiting for activation.",102 "StatusCode": 2103 }104}105MOCK_SIMPLE_UPDATE_STATUS_LIST = [106 MOCK_READY_FOR_FW_UPDATE,107 MOCK_FW_UPDATE_IN_PROGRESS,108 MOCK_WAITING_FOR_ACTIVATION109]110def get_bin_path(self, arg, required=False):111 """Mock AnsibleModule.get_bin_path"""112 return arg113def get_exception_message(ansible_exit_json):114 """From an AnsibleExitJson exception, get the message string."""115 return ansible_exit_json.exception.args[0]["msg"]116def is_changed(ansible_exit_json):117 """From an AnsibleExitJson exception, return the value of the changed flag"""118 return ansible_exit_json.exception.args[0]["changed"]119def mock_simple_update(*args, **kwargs):120 return {121 "ret": True122 }123def mocked_url_response(*args, **kwargs):124 """Mock to just return a generic string."""125 return "/mockedUrl"126def mock_update_url(*args, **kwargs):127 """Mock of the update url"""128 return "/UpdateService"129def mock_fw_activate_url(*args, **kwargs):130 """Mock of the FW Activate URL"""131 return "/UpdateService.FWActivate"132def empty_return(*args, **kwargs):133 """Mock to just return an empty successful return."""134 return {"ret": True}135def mock_get_simple_update_status_ready_for_fw_update(*args, **kwargs):136 """Mock to return simple update status Ready for FW update"""137 return MOCK_READY_FOR_FW_UPDATE138def mock_get_request_enclosure_single_tenant(*args, **kwargs):139 """Mock for get_request for single-tenant enclosure."""140 if args[1].endswith("/redfish/v1") or args[1].endswith("/redfish/v1/"):141 return MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE142 elif args[1].endswith("/mockedUrl"):143 return MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE144 elif args[1].endswith("Chassis/Enclosure"):145 return MOCK_GET_ENCLOSURE_RESPONSE_SINGLE_TENANT146 elif args[1].endswith("/UpdateService"):147 return MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_AND_FW_ACTIVATE148 else:149 raise RuntimeError("Illegal call to get_request in test: " + args[1])150def mock_get_request_enclosure_multi_tenant(*args, **kwargs):151 """Mock for get_request with multi-tenant enclosure."""152 if args[1].endswith("/redfish/v1") or args[1].endswith("/redfish/v1/"):153 return MOCK_SUCCESSFUL_RESPONSE_WITH_UPDATE_SERVICE_RESOURCE154 elif args[1].endswith("/mockedUrl"):155 return MOCK_SUCCESSFUL_HTTP_EMPTY_RESPONSE156 elif args[1].endswith("Chassis/Enclosure"):157 return MOCK_GET_ENCLOSURE_RESPONSE_MULTI_TENANT158 elif args[1].endswith("/UpdateService"):159 return MOCK_SUCCESSFUL_RESPONSE_WITH_SIMPLE_UPDATE_AND_FW_ACTIVATE160 elif args[1].endswith("/IOModuleAFRU"):161 return MOCK_GET_IOM_A_MULTI_TENANT162 elif args[1].endswith("/IOModuleBFRU"):163 return MOCK_GET_IOM_B_MULTI_TENANAT164 else:165 raise RuntimeError("Illegal call to get_request in test: " + args[1])166def mock_post_request(*args, **kwargs):167 """Mock post_request with successful response."""168 if args[1].endswith("/UpdateService.FWActivate"):169 return {170 "ret": True,171 "data": ACTION_WAS_SUCCESSFUL_MESSAGE172 }173 else:174 raise RuntimeError("Illegal POST call to: " + args[1])175def mock_get_firmware_inventory_version_1_2_3(*args, **kwargs):176 return {177 "ret": True,178 "entries": [179 {180 "Id": "IOModuleA_OOBM",181 "Version": "1.2.3"182 },183 {184 "Id": "IOModuleB_OOBM",185 "Version": "1.2.3"186 }187 ]188 }189ERROR_MESSAGE_UNABLE_TO_EXTRACT_BUNDLE_VERSION = "Unable to extract bundle version or multi-tenant status from update image tarfile"190ACTION_WAS_SUCCESSFUL_MESSAGE = "Action was successful"191class TestWdcRedfishCommand(unittest.TestCase):192 def setUp(self):193 self.mock_module_helper = patch.multiple(basic.AnsibleModule,194 exit_json=exit_json,195 fail_json=fail_json,196 get_bin_path=get_bin_path)197 self.mock_module_helper.start()198 self.addCleanup(self.mock_module_helper.stop)199 self.tempdir = tempfile.mkdtemp()200 def tearDown(self):201 shutil.rmtree(self.tempdir)202 def test_module_fail_when_required_args_missing(self):203 with self.assertRaises(AnsibleFailJson):204 set_module_args({})205 module.main()206 def test_module_fail_when_unknown_category(self):207 with self.assertRaises(AnsibleFailJson):208 set_module_args({209 'category': 'unknown',210 'command': 'FWActivate',211 'username': 'USERID',212 'password': 'PASSW0RD=21',213 'ioms': [],214 })215 module.main()216 def test_module_fail_when_unknown_command(self):217 with self.assertRaises(AnsibleFailJson):218 set_module_args({219 'category': 'Update',220 'command': 'unknown',221 'username': 'USERID',222 'password': 'PASSW0RD=21',223 'ioms': [],224 })225 module.main()226 def test_module_fw_activate_first_iom_unavailable(self):227 """Test that if the first IOM is not available, the 2nd one is used."""228 ioms = [229 "bad.example.com",230 "good.example.com"231 ]232 module_args = {233 'category': 'Update',234 'command': 'FWActivate',235 'username': 'USERID',236 'password': 'PASSW0RD=21',237 'ioms': ioms238 }239 set_module_args(module_args)240 def mock_get_request(*args, **kwargs):241 """Mock for get_request that will fail on the 'bad' IOM."""242 if "bad.example.com" in args[1]:243 return MOCK_URL_ERROR244 else:245 return mock_get_request_enclosure_single_tenant(*args, **kwargs)246 with patch.multiple(module.WdcRedfishUtils,247 _firmware_activate_uri=mock_fw_activate_url,248 _update_uri=mock_update_url,249 _find_updateservice_resource=empty_return,250 _find_updateservice_additional_uris=empty_return,251 get_request=mock_get_request,252 post_request=mock_post_request):253 with self.assertRaises(AnsibleExitJson) as cm:254 module.main()255 self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE,256 get_exception_message(cm))257 def test_module_fw_activate_pass(self):258 """Test the FW Activate command in a passing scenario."""259 # Run the same test twice -- once specifying ioms, and once specifying baseuri.260 # Both should work the same way.261 uri_specifiers = [262 {263 "ioms": ["example1.example.com"]264 },265 {266 "baseuri": "example1.example.com"267 }268 ]269 for uri_specifier in uri_specifiers:270 module_args = {271 'category': 'Update',272 'command': 'FWActivate',273 'username': 'USERID',274 'password': 'PASSW0RD=21',275 }276 module_args.update(uri_specifier)277 set_module_args(module_args)278 with patch.multiple("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils",279 _firmware_activate_uri=mock_fw_activate_url,280 _update_uri=mock_update_url,281 _find_updateservice_resource=empty_return,282 _find_updateservice_additional_uris=empty_return,283 get_request=mock_get_request_enclosure_single_tenant,284 post_request=mock_post_request):285 with self.assertRaises(AnsibleExitJson) as ansible_exit_json:286 module.main()287 self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE,288 get_exception_message(ansible_exit_json))289 self.assertTrue(is_changed(ansible_exit_json))290 def test_module_fw_activate_service_does_not_support_fw_activate(self):291 """Test FW Activate when it is not supported."""292 expected_error_message = "Service does not support FWActivate"293 set_module_args({294 'category': 'Update',295 'command': 'FWActivate',296 'username': 'USERID',297 'password': 'PASSW0RD=21',298 'ioms': ["example1.example.com"]299 })300 def mock_update_uri_response(*args, **kwargs):301 return {302 "ret": True,303 "data": {} # No Actions304 }305 with patch.multiple(module.WdcRedfishUtils,306 _firmware_activate_uri=mocked_url_response,307 _update_uri=mock_update_url,308 _find_updateservice_resource=empty_return,309 _find_updateservice_additional_uris=empty_return,310 get_request=mock_update_uri_response):311 with self.assertRaises(AnsibleFailJson) as cm:312 module.main()313 self.assertEqual(expected_error_message,314 get_exception_message(cm))315 def test_module_update_and_activate_image_uri_not_http(self):316 """Test Update and Activate when URI is not http(s)"""317 expected_error_message = "Bundle URI must be HTTP or HTTPS"318 set_module_args({319 'category': 'Update',320 'command': 'UpdateAndActivate',321 'username': 'USERID',322 'password': 'PASSW0RD=21',323 'ioms': ["example1.example.com"],324 'update_image_uri': "ftp://example.com/image"325 })326 with patch.multiple(module.WdcRedfishUtils,327 _firmware_activate_uri=mocked_url_response,328 _update_uri=mock_update_url,329 _find_updateservice_resource=empty_return,330 _find_updateservice_additional_uris=empty_return):331 with self.assertRaises(AnsibleFailJson) as cm:332 module.main()333 self.assertEqual(expected_error_message,334 get_exception_message(cm))335 def test_module_update_and_activate_target_not_ready_for_fw_update(self):336 """Test Update and Activate when target is not in the correct state."""337 mock_status_code = 999338 mock_status_description = "mock status description"339 expected_error_message = "Target is not ready for FW update. Current status: {0} ({1})".format(340 mock_status_code,341 mock_status_description342 )343 set_module_args({344 'category': 'Update',345 'command': 'UpdateAndActivate',346 'username': 'USERID',347 'password': 'PASSW0RD=21',348 'ioms': ["example1.example.com"],349 'update_image_uri': "http://example.com/image"350 })351 with patch.object(module.WdcRedfishUtils, "get_simple_update_status") as mock_get_simple_update_status:352 mock_get_simple_update_status.return_value = {353 "ret": True,354 "entries": {355 "StatusCode": mock_status_code,356 "Description": mock_status_description357 }358 }359 with patch.multiple(module.WdcRedfishUtils,360 _firmware_activate_uri=mocked_url_response,361 _update_uri=mock_update_url,362 _find_updateservice_resource=empty_return,363 _find_updateservice_additional_uris=empty_return):364 with self.assertRaises(AnsibleFailJson) as cm:365 module.main()366 self.assertEqual(expected_error_message,367 get_exception_message(cm))368 def test_module_update_and_activate_bundle_not_a_tarfile(self):369 """Test Update and Activate when bundle is not a tarfile"""370 mock_filename = os.path.abspath(__file__)371 expected_error_message = ERROR_MESSAGE_UNABLE_TO_EXTRACT_BUNDLE_VERSION372 set_module_args({373 'category': 'Update',374 'command': 'UpdateAndActivate',375 'username': 'USERID',376 'password': 'PASSW0RD=21',377 'ioms': ["example1.example.com"],378 'update_image_uri': "http://example.com/image",379 "update_creds": {380 "username": "image_user",381 "password": "image_password"382 }383 })384 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:385 mock_fetch_file.return_value = mock_filename386 with patch.multiple(module.WdcRedfishUtils,387 get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,388 _firmware_activate_uri=mocked_url_response,389 _update_uri=mock_update_url,390 _find_updateservice_resource=empty_return,391 _find_updateservice_additional_uris=empty_return):392 with self.assertRaises(AnsibleFailJson) as cm:393 module.main()394 self.assertEqual(expected_error_message,395 get_exception_message(cm))396 def test_module_update_and_activate_bundle_contains_no_firmware_version(self):397 """Test Update and Activate when bundle contains no firmware version"""398 expected_error_message = ERROR_MESSAGE_UNABLE_TO_EXTRACT_BUNDLE_VERSION399 set_module_args({400 'category': 'Update',401 'command': 'UpdateAndActivate',402 'username': 'USERID',403 'password': 'PASSW0RD=21',404 'ioms': ["example1.example.com"],405 'update_image_uri': "http://example.com/image",406 "update_creds": {407 "username": "image_user",408 "password": "image_password"409 }410 })411 tar_name = "empty_tarfile{0}.tar".format(uuid.uuid4())412 empty_tarfile = tarfile.open(os.path.join(self.tempdir, tar_name), "w")413 empty_tarfile.close()414 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:415 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)416 with patch.multiple(module.WdcRedfishUtils,417 get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,418 _firmware_activate_uri=mocked_url_response,419 _update_uri=mock_update_url,420 _find_updateservice_resource=empty_return,421 _find_updateservice_additional_uris=empty_return):422 with self.assertRaises(AnsibleFailJson) as cm:423 module.main()424 self.assertEqual(expected_error_message,425 get_exception_message(cm))426 def test_module_update_and_activate_version_already_installed(self):427 """Test Update and Activate when the bundle version is already installed"""428 mock_firmware_version = "1.2.3"429 expected_error_message = ACTION_WAS_SUCCESSFUL_MESSAGE430 set_module_args({431 'category': 'Update',432 'command': 'UpdateAndActivate',433 'username': 'USERID',434 'password': 'PASSW0RD=21',435 'ioms': ["example1.example.com"],436 'update_image_uri': "http://example.com/image",437 "update_creds": {438 "username": "image_user",439 "password": "image_password"440 }441 })442 tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,443 is_multi_tenant=False)444 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:445 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)446 with patch.multiple(module.WdcRedfishUtils,447 get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,448 get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,449 _firmware_activate_uri=mocked_url_response,450 _update_uri=mock_update_url,451 _find_updateservice_resource=empty_return,452 _find_updateservice_additional_uris=empty_return,453 get_request=mock_get_request_enclosure_single_tenant):454 with self.assertRaises(AnsibleExitJson) as result:455 module.main()456 self.assertEqual(expected_error_message,457 get_exception_message(result))458 self.assertFalse(is_changed(result))459 def test_module_update_and_activate_version_already_installed_multi_tenant(self):460 """Test Update and Activate on multi-tenant when version is already installed"""461 mock_firmware_version = "1.2.3"462 expected_error_message = ACTION_WAS_SUCCESSFUL_MESSAGE463 set_module_args({464 'category': 'Update',465 'command': 'UpdateAndActivate',466 'username': 'USERID',467 'password': 'PASSW0RD=21',468 'ioms': ["example1.example.com"],469 'update_image_uri': "http://example.com/image",470 "update_creds": {471 "username": "image_user",472 "password": "image_password"473 }474 })475 tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,476 is_multi_tenant=True)477 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:478 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)479 with patch.multiple(module.WdcRedfishUtils,480 get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,481 get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,482 _firmware_activate_uri=mocked_url_response,483 _update_uri=mock_update_url,484 _find_updateservice_resource=empty_return,485 _find_updateservice_additional_uris=empty_return,486 get_request=mock_get_request_enclosure_multi_tenant):487 with self.assertRaises(AnsibleExitJson) as result:488 module.main()489 self.assertEqual(expected_error_message,490 get_exception_message(result))491 self.assertFalse(is_changed(result))492 def test_module_update_and_activate_pass(self):493 """Test Update and Activate (happy path)"""494 mock_firmware_version = "1.2.2"495 set_module_args({496 'category': 'Update',497 'command': 'UpdateAndActivate',498 'username': 'USERID',499 'password': 'PASSW0RD=21',500 'ioms': ["example1.example.com"],501 'update_image_uri': "http://example.com/image",502 "update_creds": {503 "username": "image_user",504 "password": "image_password"505 }506 })507 tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,508 is_multi_tenant=False)509 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:510 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)511 with patch.multiple("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils",512 get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,513 simple_update=mock_simple_update,514 _simple_update_status_uri=mocked_url_response,515 # _find_updateservice_resource=empty_return,516 # _find_updateservice_additional_uris=empty_return,517 get_request=mock_get_request_enclosure_single_tenant,518 post_request=mock_post_request):519 with patch("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils.get_simple_update_status"520 ) as mock_get_simple_update_status:521 mock_get_simple_update_status.side_effect = MOCK_SIMPLE_UPDATE_STATUS_LIST522 with self.assertRaises(AnsibleExitJson) as ansible_exit_json:523 module.main()524 self.assertTrue(is_changed(ansible_exit_json))525 self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE, get_exception_message(ansible_exit_json))526 def test_module_update_and_activate_pass_multi_tenant(self):527 """Test Update and Activate with multi-tenant (happy path)"""528 mock_firmware_version = "1.2.2"529 set_module_args({530 'category': 'Update',531 'command': 'UpdateAndActivate',532 'username': 'USERID',533 'password': 'PASSW0RD=21',534 'ioms': ["example1.example.com"],535 'update_image_uri': "http://example.com/image",536 "update_creds": {537 "username": "image_user",538 "password": "image_password"539 }540 })541 tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,542 is_multi_tenant=True)543 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:544 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)545 with patch.multiple(module.WdcRedfishUtils,546 get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3,547 simple_update=mock_simple_update,548 _simple_update_status_uri=mocked_url_response,549 # _find_updateservice_resource=empty_return,550 # _find_updateservice_additional_uris=empty_return,551 get_request=mock_get_request_enclosure_multi_tenant,552 post_request=mock_post_request):553 with patch("ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.WdcRedfishUtils.get_simple_update_status"554 ) as mock_get_simple_update_status:555 mock_get_simple_update_status.side_effect = MOCK_SIMPLE_UPDATE_STATUS_LIST556 with self.assertRaises(AnsibleExitJson) as ansible_exit_json:557 module.main()558 self.assertTrue(is_changed(ansible_exit_json))559 self.assertEqual(ACTION_WAS_SUCCESSFUL_MESSAGE, get_exception_message(ansible_exit_json))560 def test_module_fw_update_multi_tenant_firmware_single_tenant_enclosure(self):561 """Test Update and Activate using multi-tenant bundle on single-tenant enclosure"""562 mock_firmware_version = "1.1.1"563 expected_error_message = "Enclosure multi-tenant is False but bundle multi-tenant is True"564 set_module_args({565 'category': 'Update',566 'command': 'UpdateAndActivate',567 'username': 'USERID',568 'password': 'PASSW0RD=21',569 'ioms': ["example1.example.com"],570 'update_image_uri': "http://example.com/image",571 "update_creds": {572 "username": "image_user",573 "password": "image_password"574 }575 })576 tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,577 is_multi_tenant=True)578 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:579 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)580 with patch.multiple(module.WdcRedfishUtils,581 get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3(),582 get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,583 _firmware_activate_uri=mocked_url_response,584 _update_uri=mock_update_url,585 _find_updateservice_resource=empty_return,586 _find_updateservice_additional_uris=empty_return,587 get_request=mock_get_request_enclosure_single_tenant):588 with self.assertRaises(AnsibleFailJson) as result:589 module.main()590 self.assertEqual(expected_error_message,591 get_exception_message(result))592 def test_module_fw_update_single_tentant_firmware_multi_tenant_enclosure(self):593 """Test Update and Activate using singe-tenant bundle on multi-tenant enclosure"""594 mock_firmware_version = "1.1.1"595 expected_error_message = "Enclosure multi-tenant is True but bundle multi-tenant is False"596 set_module_args({597 'category': 'Update',598 'command': 'UpdateAndActivate',599 'username': 'USERID',600 'password': 'PASSW0RD=21',601 'ioms': ["example1.example.com"],602 'update_image_uri': "http://example.com/image",603 "update_creds": {604 "username": "image_user",605 "password": "image_password"606 }607 })608 tar_name = self.generate_temp_bundlefile(mock_firmware_version=mock_firmware_version,609 is_multi_tenant=False)610 with patch('ansible_collections.community.general.plugins.module_utils.wdc_redfish_utils.fetch_file') as mock_fetch_file:611 mock_fetch_file.return_value = os.path.join(self.tempdir, tar_name)612 with patch.multiple(module.WdcRedfishUtils,613 get_firmware_inventory=mock_get_firmware_inventory_version_1_2_3(),614 get_simple_update_status=mock_get_simple_update_status_ready_for_fw_update,615 _firmware_activate_uri=mocked_url_response,616 _update_uri=mock_update_url,617 _find_updateservice_resource=empty_return,618 _find_updateservice_additional_uris=empty_return,619 get_request=mock_get_request_enclosure_multi_tenant):620 with self.assertRaises(AnsibleFailJson) as result:621 module.main()622 self.assertEqual(expected_error_message,623 get_exception_message(result))624 def generate_temp_bundlefile(self,625 mock_firmware_version,626 is_multi_tenant):627 """Generate a temporary fake bundle file.628 :param str mock_firmware_version: The simulated firmware version for the bundle.629 :param bool is_multi_tenant: Is the simulated bundle multi-tenant?630 This can be used for a mock FW update.631 """632 tar_name = "tarfile{0}.tar".format(uuid.uuid4())633 bundle_tarfile = tarfile.open(os.path.join(self.tempdir, tar_name), "w")634 package_filename = "oobm-{0}.pkg".format(mock_firmware_version)635 package_filename_path = os.path.join(self.tempdir, package_filename)636 package_file = open(package_filename_path, "w")637 package_file.close()638 bundle_tarfile.add(os.path.join(self.tempdir, package_filename), arcname=package_filename)639 bin_filename = "firmware.bin"640 bin_filename_path = os.path.join(self.tempdir, bin_filename)641 bin_file = open(bin_filename_path, "wb")642 byte_to_write = b'\x80' if is_multi_tenant else b'\xFF'643 bin_file.write(byte_to_write * 12)644 bin_file.close()645 for filename in [package_filename, bin_filename]:646 bundle_tarfile.add(os.path.join(self.tempdir, filename), arcname=filename)647 bundle_tarfile.close()...

Full Screen

Full Screen

list.py

Source:list.py Github

copy

Full Screen

1# -*- coding: utf-8 -*- #2# Copyright 2017 Google LLC. All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Command for listing available services."""16from __future__ import absolute_import17from __future__ import division18from __future__ import unicode_literals19from googlecloudsdk.api_lib.run import global_methods20from googlecloudsdk.calliope import base21from googlecloudsdk.command_lib.run import commands22from googlecloudsdk.command_lib.run import connection_context23from googlecloudsdk.command_lib.run import flags24from googlecloudsdk.command_lib.run import platforms25from googlecloudsdk.command_lib.run import pretty_print26from googlecloudsdk.command_lib.run import resource_args27from googlecloudsdk.command_lib.run import serverless_operations28from googlecloudsdk.command_lib.util.concepts import concept_parsers29from googlecloudsdk.command_lib.util.concepts import presentation_specs30from googlecloudsdk.core import log31from googlecloudsdk.core import properties32@base.ReleaseTracks(base.ReleaseTrack.BETA, base.ReleaseTrack.GA)33class List(commands.List):34 """List available services."""35 detailed_help = {36 'DESCRIPTION': """\37 {description}38 """,39 'EXAMPLES': """\40 To list available services:41 $ {command}42 """,43 }44 @classmethod45 def CommonArgs(cls, parser):46 # Flags specific to connecting to a cluster47 cluster_group = flags.GetClusterArgGroup(parser)48 namespace_presentation = presentation_specs.ResourcePresentationSpec(49 '--namespace',50 resource_args.GetNamespaceResourceSpec(),51 'Namespace to list services in.',52 required=True,53 prefixes=False)54 concept_parsers.ConceptParser(55 [namespace_presentation]).AddToParser(cluster_group)56 parser.display_info.AddUriFunc(cls._GetResourceUri)57 @classmethod58 def Args(cls, parser):59 cls.CommonArgs(parser)60 def _SetFormat(self,61 args,62 show_region=False,63 show_namespace=False,64 show_description=False):65 """Set display format for output.66 Args:67 args: Namespace, the args namespace68 show_region: bool, True to show region of listed services69 show_namespace: bool, True to show namespace of listed services70 show_description: bool, True to show description of listed services71 """72 columns = [73 pretty_print.READY_COLUMN,74 'firstof(id,metadata.name):label=SERVICE',75 ]76 if show_region:77 columns.append('region:label=REGION')78 if show_namespace:79 columns.append('namespace:label=NAMESPACE')80 if show_description:81 columns.append('description:label=DESCRIPTION')82 columns.extend([83 'domain:label=URL',84 'last_modifier:label="LAST DEPLOYED BY"',85 'last_transition_time:label="LAST DEPLOYED AT"',86 ])87 args.GetDisplayInfo().AddFormat(88 'table({})'.format(','.join(columns)))89 def Run(self, args):90 """List available services."""91 is_managed = platforms.GetPlatform() == platforms.PLATFORM_MANAGED92 if is_managed and not args.IsSpecified('region'):93 self._SetFormat(args, show_region=True)94 client = global_methods.GetServerlessClientInstance()95 self.SetPartialApiEndpoint(client.url)96 args.CONCEPTS.namespace.Parse() # Error if no proj.97 # Don't consider region property here, we'll default to all regions98 return commands.SortByName(global_methods.ListServices(client))99 else:100 conn_context = connection_context.GetConnectionContext(101 args, flags.Product.RUN, self.ReleaseTrack())102 self._SetFormat(103 args, show_region=is_managed, show_namespace=(not is_managed))104 namespace_ref = args.CONCEPTS.namespace.Parse()105 with serverless_operations.Connect(conn_context) as client:106 self.SetCompleteApiEndpoint(conn_context.endpoint)107 if not is_managed:108 location_msg = ' in [{}]'.format(conn_context.cluster_location)109 project_msg = ' in project [{}]'.format(conn_context.cluster_project)110 is_multi_tenant = conn_context.cluster_project != properties.VALUES.core.project.Get(111 required=False)112 log.status.Print('For cluster [{cluster}]{zone}{project}:'.format(113 cluster=conn_context.cluster_name,114 zone=location_msg if conn_context.cluster_location else '',115 project=project_msg if is_multi_tenant else ''))116 return commands.SortByName(client.ListServices(namespace_ref))117@base.ReleaseTracks(base.ReleaseTrack.ALPHA)118class AlphaList(List):119 """List available services."""120 @classmethod121 def Args(cls, parser):122 cls.CommonArgs(parser)...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful