Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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

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

list.py

Source:list.py

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)...

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.