Convert f-strings to str.format() (#19)

* Convert f-strings to str.format()

* Fix sanity ansible-format-automatic-specification

* skip tests on PRs

* typo

* load_file_common_args has path in params

* python 2.7 urllib open has no context manager

Author: guidograzioli

molecule/prod_eap/converge.yml

@@ -6,41 +6,45 @@
     client_id: "{{ lookup('env', 'PROD_JBOSSNETWORK_API_CLIENTID') }}"
     client_secret: "{{ lookup('env', 'PROD_JBOSSNETWORK_API_SECRET') }}"
   tasks:
-    - name: "Ensures required secret are provided."
-      ansible.builtin.assert:
-        that:
-          - client_id is defined and client_id | length > 0
-          - client_secret is defined and client_secret | length > 0
-        fail_msg: "Missing required env vars PROD_JBOSSNETWORK_API_CLIENTID and/or PROD_JBOSSNETWORK_API_SECRET"
+    - name: Execute scenario if staging credentials are provided
+      when:
+        - client_id is defined and client_password is defined
+      block:
+        - name: "Ensures required secret are provided."
+          ansible.builtin.assert:
+            that:
+              - client_id is defined and client_id | length > 0
+              - client_secret is defined and client_secret | length > 0
+            fail_msg: "Missing required env vars PROD_JBOSSNETWORK_API_CLIENTID and/or PROD_JBOSSNETWORK_API_SECRET"
 
-    - name: Search EAP Product
-      middleware_automation.common.product_search:
-        client_id: "{{ client_id }}"
-        client_secret: "{{ client_secret }}"
-        product_type: DISTRIBUTION
-        product_version: '7.4'
-        product_category: appplatform
-      register: rhn_products
-      no_log: false
+        - name: Search EAP Product
+          middleware_automation.common.product_search:
+            client_id: "{{ client_id }}"
+            client_secret: "{{ client_secret }}"
+            product_type: DISTRIBUTION
+            product_version: '7.4'
+            product_category: appplatform
+          register: rhn_products
+          no_log: false
 
-    - name: Search install zipfile
-      ansible.builtin.set_fact:
-        rhn_filtered_products: "{{ rhn_products.results | selectattr('file_path', 'match', '[^/]*/jboss-eap-7.4.0.zip$') }}"
+        - name: Search install zipfile
+          ansible.builtin.set_fact:
+            rhn_filtered_products: "{{ rhn_products.results | selectattr('file_path', 'match', '[^/]*/jboss-eap-7.4.0.zip$') }}"
 
-    - name: Retrieve product download using JBossNetwork API
-      middleware_automation.common.product_search:
-        client_id: "{{ client_id }}"
-        client_secret: "{{ client_secret }}"
-        product_type: BUGFIX
-        product_version: '7.4'
-        product_category: appplatform
-      register: rhn_products
-      no_log: true
+        - name: Retrieve product download using JBossNetwork API
+          middleware_automation.common.product_search:
+            client_id: "{{ client_id }}"
+            client_secret: "{{ client_secret }}"
+            product_type: BUGFIX
+            product_version: '7.4'
+            product_category: appplatform
+          register: rhn_products
+          no_log: true
 
-    - name: Determine patch versions list
-      ansible.builtin.set_fact:
-        filtered_versions: "{{ rhn_products.results | map(attribute='file_path') | select('match', '^[^/]*/jboss-eap-.*[0-9]*[.][0-9]*[.][0-9]*.*$') | map('regex_replace','[^/]*/jboss-eap-([0-9]*[.][0-9]*[.][0-9]*)-.*','\\1' ) | list | unique }}"
+        - name: Determine patch versions list
+          ansible.builtin.set_fact:
+            filtered_versions: "{{ rhn_products.results | map(attribute='file_path') | select('match', '^[^/]*/jboss-eap-.*[0-9]*[.][0-9]*[.][0-9]*.*$') | map('regex_replace','[^/]*/jboss-eap-([0-9]*[.][0-9]*[.][0-9]*)-.*','\\1' ) | list | unique }}"
 
-    - name: Determine latest version
-      ansible.builtin.debug:
-         msg: "JBoss EAP 7.4 most recent CP is version {{ filtered_versions | middleware_automation.common.version_sort | last }}"
+        - name: Determine latest version
+          ansible.builtin.debug:
+            msg: "JBoss EAP 7.4 most recent CP is version {{ filtered_versions | middleware_automation.common.version_sort | last }}"

molecule/stage_eap/converge.yml

@@ -1,16 +1,15 @@
 ---
 - name: Converge
-  hosts: all
-  gather_facts: yes
+  hosts: localhost
+  gather_facts: no
   vars:
-        client_id: "{{ lookup('env', 'STAGE_JBOSSNETWORK_API_CLIENTID') }}"
-        client_secret: "{{ lookup('env', 'STAGE_JBOSSNETWORK_API_SECRET') }}"
+    client_id: "{{ lookup('env', 'STAGE_JBOSSNETWORK_API_CLIENTID') }}"
+    client_secret: "{{ lookup('env', 'STAGE_JBOSSNETWORK_API_SECRET') }}"
   tasks:
     - name: Execute scenario if staging credentials are provided
       when:
         - client_id is defined and client_password is defined
       block:
-
         - name: Search EAP Product
           middleware_automation.common.product_search:
             api_url: "https://jbossnetwork.stage.api.redhat.com"

plugins/module_utils/jbossnetwork.py

@@ -49,7 +49,7 @@ def get_authenticated_session(module, sso_url, validate_certs, client_id, client
     session = Request(validate_certs=validate_certs, headers={})
 
     try:
-        token_request = session.post(f"{sso_url}/auth/realms/redhat-external/protocol/openid-connect/token", data=urlencode(token_request_data))
+        token_request = session.post("{0}/auth/realms/redhat-external/protocol/openid-connect/token".format(sso_url), data=urlencode(token_request_data))
 
     except Exception as err:
         module.fail_json(msg="Error Retrieving SSO Access Token: %s" % (to_native(err)))
@@ -58,7 +58,7 @@ def get_authenticated_session(module, sso_url, validate_certs, client_id, client
 
     # Setup Session
     session.headers = {
-        "Authorization": f"Bearer {access_token}",
+        "Authorization": "Bearer {0}".format(access_token),
         "Content-Type": "application/json",
     }
 

plugins/modules/product_download.py

@@ -186,41 +186,41 @@ def main():
 
     if not client_id:
         module.fail_json(msg=str("Client ID not specified and unable to determine Client ID "
-                                 f"from '{REDHAT_PRODUCT_DOWNLOAD_CLIENT_ID_ENV_VAR}' environment variable."))
+                                 "from '{0}' environment variable.".format(REDHAT_PRODUCT_DOWNLOAD_CLIENT_ID_ENV_VAR)))
 
     if not client_secret:
         client_secret = os.environ.get(REDHAT_PRODUCT_DOWNLOAD_CLIENT_SECRET_ENV_VAR)
 
     if not client_secret:
         module.fail_json(msg=str("Client Secret not specified and unable to determine Client Secret "
-                                 f"from '{REDHAT_PRODUCT_DOWNLOAD_CLIENT_SECRET_ENV_VAR}' environment variable."))
+                                 "from '{0}' environment variable.".format(REDHAT_PRODUCT_DOWNLOAD_CLIENT_SECRET_ENV_VAR)))
 
     if not dest:
         module.fail_json(msg=str("Destination path not provided"))
 
     session = get_authenticated_session(module, sso_url, validate_certs, client_id, client_secret)
 
-    api_base_url = f"{api_url}{API_SERVICE_PATH}"
+    api_base_url = "{0}{1}".format(api_url, API_SERVICE_PATH)
 
     if product_category is not None:
         # List Product Categories
         product_categories = []
 
         try:
-            product_categories = perform_search(session, f"{api_base_url}{LIST_PRODUCT_CATEGORIES_ENDPOINT}", validate_certs)
+            product_categories = perform_search(session, "{0}{1}".format(api_base_url, LIST_PRODUCT_CATEGORIES_ENDPOINT), validate_certs)
         except Exception as err:
             module.fail_json(msg="Error Listing Available Product Categories: %s" % (to_native(err)))
 
         if product_category not in product_categories:
-            module.fail_json(msg=f"'{product_category}' is not a valid Product Category")
+            module.fail_json(msg="'{0}' is not a valid Product Category".format(product_category))
 
     # Search for Products
     search_results = []
 
     search_params = generate_search_params(product_category, product_id, product_type, product_version)
 
     try:
-        search_results = perform_search(session, f"{api_base_url}{SEARCH_ENDPOINT}", validate_certs, search_params)
+        search_results = perform_search(session, "{0}{1}".format(api_base_url, SEARCH_ENDPOINT), validate_certs, search_params)
     except Exception as err:
         module.fail_json(msg="Error Searching for Products: %s" % (to_native(err)))
 
@@ -229,11 +229,11 @@ def main():
     # Print error with results if more than 1 exists
     if products_found != 1:
         msg = [
-            (f"Error: Unable to locate a single product to download. '{products_found}' products found.")
+            ("Error: Unable to locate a single product to download. '{0}' products found.".format(products_found))
         ]
 
         for productIdx, product in enumerate(search_results):
-            msg.append(f"{productIdx+1} - ({product['id']}) {product['title']}.")
+            msg.append("{0} - ({1}) {2}.".format(productIdx + 1, product['id'], product['title']))
 
         module.fail_json(msg=" ".join(msg))
 
@@ -250,7 +250,7 @@ def main():
     )
 
     if os.path.exists(dest) and not force:
-        file_args = module.load_file_common_arguments(module.params, path=dest)
+        file_args = module.load_file_common_arguments(module.params)
         result['changed'] = module.set_fs_attributes_if_different(file_args, False)
 
         if result['changed']:
@@ -272,20 +272,20 @@ def main():
             module.fail_json(msg="Destination %s is not writable" % (os.path.dirname(dest)), **result)
 
     try:
-        with session.get(search_results[0]["download_path"], follow_redirects=True, headers={"User-Agent": "product_download"}) as r:
-            chunk_size = 8192
-            with open(dest, 'wb') as f:
-                while True:
-                    chunk = r.read(chunk_size)
-                    if not chunk:
-                        break
-                    f.write(chunk)
+        r = session.get(search_results[0]["download_path"], follow_redirects=True, headers={"User-Agent": "product_download"})
+        chunk_size = 8192
+        with open(dest, 'wb') as f:
+            while True:
+                chunk = r.read(chunk_size)
+                if not chunk:
+                    break
+                f.write(chunk)
 
         result['changed'] = True
     except Exception as err:
         module.fail_json(msg="Error Downloading %s: %s" % (search_results[0]['title'], to_native(err)))
 
-    file_args = module.load_file_common_arguments(module.params, path=dest)
+    file_args = module.load_file_common_arguments(module.params)
     result['changed'] = module.set_fs_attributes_if_different(file_args, result['changed'])
 
     try:

plugins/modules/product_search.py

@@ -167,38 +167,38 @@ def main():
 
     if not client_id:
         module.fail_json(msg=str("Client ID not specified and unable to determine Client ID "
-                                 f"from '{REDHAT_PRODUCT_DOWNLOAD_CLIENT_ID_ENV_VAR}' environment variable."))
+                                 "from '{0}' environment variable.".format(REDHAT_PRODUCT_DOWNLOAD_CLIENT_ID_ENV_VAR)))
 
     if not client_secret:
         client_secret = os.environ.get(REDHAT_PRODUCT_DOWNLOAD_CLIENT_SECRET_ENV_VAR)
 
     if not client_secret:
         module.fail_json(msg=str("Client Secret not specified and unable to determine Client Secret "
-                                 f"from '{REDHAT_PRODUCT_DOWNLOAD_CLIENT_SECRET_ENV_VAR}' environment variable."))
+                                 "from '{0}' environment variable.".format(REDHAT_PRODUCT_DOWNLOAD_CLIENT_SECRET_ENV_VAR)))
 
     session = get_authenticated_session(module, sso_url, validate_certs, client_id, client_secret)
 
-    api_base_url = f"{api_url}{API_SERVICE_PATH}"
+    api_base_url = "{0}{1}".format(api_url, API_SERVICE_PATH)
 
     if product_category is not None:
         # List Product Categories
         product_categories = []
 
         try:
-            product_categories = perform_search(session, f"{api_base_url}{LIST_PRODUCT_CATEGORIES_ENDPOINT}", validate_certs)
+            product_categories = perform_search(session, "{0}{1}".format(api_base_url, LIST_PRODUCT_CATEGORIES_ENDPOINT), validate_certs)
         except Exception as err:
             module.fail_json(msg="Error Listing Available Product Categories: %s" % (to_native(err)))
 
         if product_category not in product_categories:
-            module.fail_json(msg=f"'{product_category}' is not a valid Product Category")
+            module.fail_json(msg="'{0}' is not a valid Product Category".format(product_category))
 
     # Search for Products
     search_results = []
 
     search_params = generate_search_params(product_category, product_id, product_type, product_version)
 
     try:
-        search_results = perform_search(session, f"{api_base_url}{SEARCH_ENDPOINT}", validate_certs, search_params)
+        search_results = perform_search(session, "{0}{1}".format(api_base_url, SEARCH_ENDPOINT), validate_certs, search_params)
     except Exception as err:
         module.fail_json(msg="Error Searching for Products: %s" % (to_native(err)))