Rv1.10.0

ChangeLog.md

@@ -1,5 +1,8 @@
 # PyPowerFlex Change Log
 
+## Version 1.10.0 - released on 29/03/24
+- Added support for retrieving all the firmware repository, validating, deploying, editing, adding nodes and deleting a resource group from PowerFlex Manager.
+
 ## Version 1.9.0 - released on 29/02/24
 - Added support for retrieving managed devices, service templates and deployments from PowerFlex Manager.
 

PyPowerFlex/__init__.py

@@ -47,7 +47,8 @@ class PowerFlexClient:
         'replication_pair',
         'service_template',
         'managed_device',
-        'deployment'
+        'deployment',
+        'firmware_repository'
     )
 
     def __init__(self,
@@ -98,6 +99,7 @@ def initialize(self):
         self.__add_storage_entity('service_template', objects.ServiceTemplate)
         self.__add_storage_entity('managed_device', objects.ManagedDevice)
         self.__add_storage_entity('deployment', objects.Deployment)
+        self.__add_storage_entity('firmware_repository', objects.FirmwareRepository)
         utils.init_logger(self.configuration.log_level)
         if version.parse(self.system.api_version()) < version.Version('3.0'):
             raise exceptions.PowerFlexClientException(

PyPowerFlex/base_client.py

@@ -27,6 +27,9 @@
 
 class Request:
     GET = "get"
+    POST = "post"
+    PUT = "put"
+    DELETE = "delete"
 
     def __init__(self, token, configuration):
         self.token = token
@@ -67,39 +70,39 @@ def get_auth_headers(self, request_type=None):
         return {'Authorization': 'Bearer {0}'.format(self.token.get()),
                 'content-type': 'application/json'}
 
-    def send_get_request(self, url, **url_params):
-        request_url = self.base_url + url.format(**url_params)
+    def send_request(self, method, url, params=None, **url_params):
+        params = params or {}
+        request_url = f"{self.base_url}{url.format(**url_params)}"
         version = self.login()
-        request_params = {'url': request_url,
-                          'headers': self.get_auth_headers(request_type=self.GET),
-                          'verify': self.verify_certificate,
-                          'timeout': self.configuration.timeout}
+        request_params = {
+            'headers': self.get_auth_headers(method),
+            'verify': self.verify_certificate,
+            'timeout': self.configuration.timeout
+        }
         if utils.is_version_3(version):
-            request_params['auth'] = (self.configuration.username,
-                                      self.token.get())
+            request_params['auth'] = (self.configuration.username, self.token.get())
             request_params['headers'] = None
-        r = requests.get(**request_params)
+
+        if method in [self.PUT, self.POST]:
+            request_params['data'] = utils.prepare_params(params)
+        response = requests.request(method, request_url, **request_params)
         self.logout(version)
-        response = r.json()
-        return r, response
+        return response
+
+    def send_get_request(self, url, params=None, **url_params):
+        response = self.send_request(self.GET, url, params, **url_params)
+        return response, response.json()
 
     def send_post_request(self, url, params=None, **url_params):
-        if params is None:
-            params = dict()
-        version = self.login()
-        request_url = self.base_url + url.format(**url_params)
-        r = requests.post(request_url,
-                          auth=(
-                              self.configuration.username,
-                              self.token.get()
-                          ),
-                          headers=self.headers,
-                          data=utils.prepare_params(params),
-                          verify=self.verify_certificate,
-                          timeout=self.configuration.timeout)
-        response = r.json()
-        self.logout(version)
-        return r, response
+        response = self.send_request(self.POST, url, params, **url_params)
+        return response, response.json()
+
+    def send_put_request(self, url, params=None, **url_params):
+        response = self.send_request(self.PUT, url, params, **url_params)
+        return response, response.json()
+
+    def send_delete_request(self, url, params=None, **url_params):
+        return self.send_request(self.DELETE, url, params, **url_params)
 
     def send_mdm_cluster_post_request(self, url, params=None, **url_params):
         if params is None:
@@ -234,6 +237,7 @@ class EntityRequest(Request):
     service_template_url = '/V1/ServiceTemplate'
     managed_device_url = '/V1/ManagedDevice'
     deployment_url = '/V1/Deployment'
+    firmware_repository_url = '/V1/FirmwareRepository'
     entity_name = None
 
     @property

PyPowerFlex/objects/__init__.py

@@ -29,6 +29,7 @@
 from PyPowerFlex.objects.service_template import ServiceTemplate
 from PyPowerFlex.objects.managed_device import ManagedDevice
 from PyPowerFlex.objects.deployment import Deployment
+from PyPowerFlex.objects.firmware_repository import FirmwareRepository
 
 
 __all__ = [
@@ -48,4 +49,5 @@
     'ServiceTemplate',
     'ManagedDevice',
     'Deployment',
+    'FirmwareRepository',
 ]

PyPowerFlex/objects/deployment.py

@@ -50,3 +50,93 @@ def get(self, filters=None, full=None, include_devices=None, include_template=No
             LOG.error(msg)
             raise exceptions.PowerFlexClientException(msg)
         return response
+
+    def get_by_id(self, deployment_id):
+        """
+        Retrieve Deployment for specified ID.
+        :param deployment_id: Deployment ID.
+        :return: A dictionary containing the retrieved Deployment.
+        """
+        r, response = self.send_get_request(f'{self.deployment_url}/{deployment_id}')
+        if r.status_code != requests.codes.ok:
+            msg = (f'Failed to retrieve deployment by id {deployment_id}. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+        return response
+
+    def validate(self, rg_data):
+        """
+        Validates a new deployment.
+        Args:
+            rg_data (dict): The resource group data to be deployed.
+        Returns:
+            dict: The response from the deployment API.
+        Raises:
+            PowerFlexClientException: If the deployment fails.
+        """
+        r, response = self.send_post_request(f'{self.deployment_url}/validate', rg_data)
+        if r.status_code != requests.codes.ok:
+            msg = (f'Failed to validate the deployment. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+
+        return response
+
+    def create(self, rg_data):
+        """
+        Creates a new deployment.
+        Args:
+            rg_data (dict): The resource group data to be deployed.
+        Returns:
+            dict: The response from the deployment API.
+        Raises:
+            PowerFlexClientException: If the deployment fails.
+        """
+        r, response = self.send_post_request(self.deployment_url, rg_data)
+        if r.status_code != requests.codes.ok:
+            msg = (f'Failed to create a new deployment. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+
+        return response
+
+    def edit(self, deployment_id, rg_data):
+        """
+        Edit a deployment with the given ID using the provided data.
+        Args:
+            deployment_id (str): The ID of the deployment to edit.
+            rg_data (dict): The data to use for editing the deployment.
+        Returns:
+            dict: The response from the API.
+        Raises:
+            PowerFlexClientException: If the request fails.
+        """
+        request_url = f'{self.deployment_url}/{deployment_id}'
+        r, response = self.send_put_request(request_url, rg_data)
+
+        if r.status_code != requests.codes.ok:
+            msg = (f'Failed to edit the deployment. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+
+        return response
+
+    def delete(self, deployment_id):
+        """
+        Deletes a deployment with the given ID.
+        Args:
+            deployment_id (str): The ID of the deployment to delete.
+        Returns:
+            str: The response from the delete request.
+        Raises:
+            exceptions.PowerFlexClientException: If the delete request fails.
+        """
+        request_url = f'{self.deployment_url}/{deployment_id}'
+        response = self.send_delete_request(request_url)
+
+        if response.status_code != requests.codes.no_content:
+            msg = (f'Failed to delete deployment. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+
+        return response

PyPowerFlex/objects/firmware_repository.py

@@ -0,0 +1,51 @@
+# Copyright (c) 2024 Dell Inc. or its subsidiaries.
+# All Rights Reserved.
+#
+# 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 logging
+import requests
+from PyPowerFlex import base_client
+from PyPowerFlex import exceptions
+from PyPowerFlex import utils
+LOG = logging.getLogger(__name__)
+
+
+class FirmwareRepository(base_client.EntityRequest):
+    def get(self, filters=None, limit=None, offset=None, sort=None, related=False, bundles=False, components=False):
+        """
+        Retrieve all firmware repository with filter, sort, pagination
+        :param filters: (Optional) The filters to apply to the results.
+        :param limit: (Optional) Page limit.
+        :param offset: (Optional) Pagination offset.
+        :param sort: (Optional) The field to sort the results by.
+        :param related: Whether to include related entities in the response.
+        :param bundles: Whether to include bundles in the response.
+        :param components: Whether to include components in the response.
+        :return: A list of dictionary containing the retrieved firmware repository.
+        """
+        params = dict(
+            filter=filters,
+            sort=sort,
+            offset=offset,
+            limit=limit,
+            related=related,
+            bundles=bundles,
+            components=components
+        )
+        r, response = self.send_get_request(utils.build_uri_with_params(self.firmware_repository_url, **params))
+        if r.status_code != requests.codes.ok:
+            msg = (f'Failed to retrieve firmware repository. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+        return response

PyPowerFlex/objects/service_template.py

@@ -46,3 +46,20 @@ def get(self, filters=None, full=None, limit=None, offset=None, sort=None, inclu
             LOG.error(msg)
             raise exceptions.PowerFlexClientException(msg)
         return response
+
+    def get_by_id(self, service_template_id, for_deployment=False):
+        """
+        Retrieve a Service Template by its ID.
+        :param service_template_id: The ID of the Service Template to retrieve.
+        :param for_deployment: (Optional) Whether to retrieve the Service Template for deployment.
+        :return: A dictionary containing the retrieved Service Template.
+        """
+        url = f'{self.service_template_url}/{service_template_id}'
+        if for_deployment:
+            url += '?forDeployment=true'
+        r, response = self.send_get_request(url)
+        if r.status_code != requests.codes.ok:
+            msg = (f'Failed to retrieve service template by id {service_template_id}. Error: {response}')
+            LOG.error(msg)
+            raise exceptions.PowerFlexClientException(msg)
+        return response

PyPowerFlex/utils.py

@@ -120,6 +120,9 @@ def prepare_params(params, dump=True):
     :return: prepared parameters
     """
 
+    if not isinstance(params, dict):
+        return params
+
     prepared = dict()
     for name, value in params.items():
         if value is not None:

README.md

@@ -44,6 +44,7 @@ python setup.py install
 * ManagedDevice
 * Deployment
 * ServiceTemplate
+* FirmwareRepository
 
 #### Initialize PowerFlex client
 

setup.py

@@ -17,7 +17,7 @@
 
 setup(
     name='PyPowerFlex',
-    version='1.9.0',
+    version='1.10.0',
     description='Python library for Dell PowerFlex',
     author='Ansible Team at Dell',
     author_email='ansible.team@dell.com',

tests/__init__.py

@@ -34,7 +34,6 @@ class MockResponse(requests.Response):
 
     def __init__(self, content, status_code=200):
         super(MockResponse, self).__init__()
-
         self._content = content
         self.request = mock.MagicMock()
         self.status_code = status_code
@@ -103,6 +102,7 @@ def setUp(self):
                                                   self.username,
                                                   self.password,
                                                   log_level=logging.DEBUG)
+        requests.request = self.get_mock_response
         self.get_mock = self.mock_object(requests,
                                          'get',
                                          side_effect=self.get_mock_response)
@@ -117,7 +117,6 @@ def mock_object(self, obj, attr_name, *args, **kwargs):
         Mocks the specified objects attribute with the given value.
         Automatically performs 'addCleanup' for the mock.
         """
-
         patcher = mock.patch.object(obj, attr_name, *args, **kwargs)
         result = patcher.start()
         self.addCleanup(patcher.stop)
@@ -131,10 +130,11 @@ def http_response_mode(self, mode):
         yield
         self.__http_response_mode = previous_response_mode
 
-    def get_mock_response(self, url, mode=None, *args, **kwargs):
+    def get_mock_response(self, url, request_url=None, mode=None, *args, **kwargs):
         if mode is None:
             mode = self.__http_response_mode
-        api_path = url.split('/api')[1]
+
+        api_path = url.split('/api')[1] if ('/api' in url) else request_url.split('/api')[1]
         try:
             if api_path == "/login":
                 response = self.RESPONSE_MODE.Valid[0]
@@ -155,7 +155,14 @@ def get_mock_response(self, url, mode=None, *args, **kwargs):
                         )
                     )
         if not isinstance(response, MockResponse):
-            response = MockResponse(response, 200)
+            response = self._get_mock_response(response)
+
         response.request.url = url
         response.request.body = kwargs.get('data')
         return response
+
+    def _get_mock_response(self, response):
+        if "204" in str(response):
+            return MockResponse(response, 204)
+        else:
+            return MockResponse(response, 200)