Linting

Author: durera

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2025-12-12T20:28:52Z",
+  "generated_at": "2025-12-15T15:57:18Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -82,15 +82,15 @@
         "hashed_secret": "053f5ed451647be0bbb6f67b80d6726808cad97e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 42,
+        "line_number": 44,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "4f75456d6c1887d41ed176f7ad3e2cfff3fdfd91",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 51,
+        "line_number": 53,
         "type": "Secret Keyword",
         "verified_result": null
       }

bin/mas-devops-create-initial-users-for-saas

@@ -10,21 +10,18 @@
 #
 # *****************************************************************************
 
+from mas.devops.users import MASUserUtils
+from botocore.exceptions import ClientError
+import boto3
+import sys
+import json
+import yaml
 from kubernetes import client, config
 from kubernetes.config.config_exception import ConfigException
 import argparse
 import logging
 import urllib3
 urllib3.disable_warnings()
-import yaml
-import json
-import sys
-
-import boto3
-from botocore.exceptions import ClientError
-
-from mas.devops.users import MASUserUtils
-
 
 
 if __name__ == "__main__":
@@ -38,7 +35,6 @@ if __name__ == "__main__":
     parser.add_argument("--admin-dashboard-port", required=False, default=443)
     parser.add_argument("--manage-api-port", required=False, default=443)
 
-
     group = parser.add_mutually_exclusive_group(required=True)
     group.add_argument("--initial-users-yaml-file")
     group.add_argument("--initial-users-secret-name")
@@ -66,7 +62,6 @@ if __name__ == "__main__":
     admin_dashboard_port = args.admin_dashboard_port
     manage_api_port = args.manage_api_port
 
-
     logger.info("Configuration:")
     logger.info("--------------")
     logger.info(f"mas_instance_id:           {mas_instance_id}")
@@ -88,7 +83,6 @@ if __name__ == "__main__":
         config.load_kube_config()
         logger.debug("Loaded kubeconfig file")
 
-
     user_utils = MASUserUtils(mas_instance_id, mas_workspace_id, client.api_client.ApiClient(), coreapi_port=coreapi_port, admin_dashboard_port=admin_dashboard_port, manage_api_port=manage_api_port)
 
     if initial_users_secret_name is not None:
@@ -100,7 +94,7 @@ if __name__ == "__main__":
             service_name='secretsmanager',
         )
         try:
-            initial_users_secret = aws_sm_client.get_secret_value( # pragma: allowlist secret
+            initial_users_secret = aws_sm_client.get_secret_value(  # pragma: allowlist secret
                 SecretId=initial_users_secret_name
             )
         except ClientError as e:
@@ -109,16 +103,15 @@ if __name__ == "__main__":
                 sys.exit(0)
 
             raise Exception(f"Failed to fetch secret {initial_users_secret_name}: {str(e)}")
-        
+
         secret_json = json.loads(initial_users_secret['SecretString'])
         initial_users = user_utils.parse_initial_users_from_aws_secret_json(secret_json)
     elif initial_users_yaml_file is not None:
         with open(initial_users_yaml_file, 'r') as file:
             initial_users = yaml.safe_load(file)
     else:
         raise Exception("Something unexpected happened")
-    
-    
+
     result = user_utils.create_initial_users_for_saas(initial_users)
 
     # if user details were sourced from an AWS SM secret, remove the completed entries from the secret
@@ -133,14 +126,13 @@ if __name__ == "__main__":
         if has_updates:
             logger.info(f"Updating secret {initial_users_secret_name}")
             try:
-                aws_sm_client.update_secret( # pragma: allowlist secret
+                aws_sm_client.update_secret(  # pragma: allowlist secret
                     SecretId=initial_users_secret_name,
                     SecretString=json.dumps(secret_json)
                 )
             except ClientError as e:
                 raise Exception(f"Failed to update secret {initial_users_secret_name}: {str(e)}")
-            
 
     if len(result["failed"]) > 0:
-        failed_user_ids = list(map(lambda u : u["email"], result["failed"]))
-        raise Exception(f"Sync failed for the following user IDs {failed_user_ids}")
+        failed_user_ids = list(map(lambda u: u["email"], result["failed"]))
+        raise Exception(f"Sync failed for the following user IDs {failed_user_ids}")

bin/mas-devops-notify-slack

@@ -33,7 +33,7 @@ def _getToolchainLink() -> str:
     return ""
 
 
-def notifyProvisionFyre(channels: list[str], rc: int, additionalMsg: str=None) -> bool:
+def notifyProvisionFyre(channels: list[str], rc: int, additionalMsg: str = None) -> bool:
     """Send Slack notification about Fyre OCP cluster provisioning status."""
     name = _getClusterName()
     toolchainLink = _getToolchainLink()
@@ -60,14 +60,14 @@ def notifyProvisionFyre(channels: list[str], rc: int, additionalMsg: str=None) -
             SlackUtil.buildHeader(f":glyph-fail: Your IBM DevIT Fyre OCP cluster ({name}) failed to deploy"),
             SlackUtil.buildSection(f"<https://beta.fyre.ibm.com/development/vms|Fyre Dashboard>{toolchainLink}")
         ]
-    
+
     response = SlackUtil.postMessageBlocks(channels, message)
     if isinstance(response, list):
         return all([res.data.get("ok", False) for res in response])
     return response.data.get("ok", False)
 
 
-def notifyProvisionRoks(channels: list[str], rc: int, additionalMsg: str=None) -> bool:
+def notifyProvisionRoks(channels: list[str], rc: int, additionalMsg: str = None) -> bool:
     """Send Slack notification about ROKS cluster provisioning status."""
     name = _getClusterName()
     toolchainLink = _getToolchainLink()
@@ -103,18 +103,18 @@ if __name__ == "__main__":
     SLACK_CHANNEL = os.getenv("SLACK_CHANNEL", "")
     if SLACK_TOKEN == "" or SLACK_CHANNEL == "":
         sys.exit(0)
-    
+
     # Parse comma-separated channel list
     channelList = [ch.strip() for ch in SLACK_CHANNEL.split(",")]
-    
+
     # Initialize the properties we need
     parser = argparse.ArgumentParser()
 
     # Primary Options
     parser.add_argument("--action", required=True)
     parser.add_argument("--rc", required=True, type=int)
     parser.add_argument("--msg", required=False, default=None)
-    
+
     args, unknown = parser.parse_known_args()
 
     if args.action == "ocp-provision-fyre":

bin/mas-devops-saas-job-cleaner

@@ -46,7 +46,6 @@ if __name__ == "__main__":
     ch.setFormatter(chFormatter)
     logger.addHandler(ch)
 
-
     limit = args.limit
     label = args.label
     dry_run = args.dry_run

src/mas/devops/slack.py

@@ -39,7 +39,7 @@ def client(cls) -> WebClient:
 
     # Post message to Slack
     # -----------------------------------------------------------------------------
-    def postMessageBlocks(cls, channelList: str | list[str], messageBlocks: list, threadId: str=None) -> SlackResponse | list[SlackResponse]:
+    def postMessageBlocks(cls, channelList: str | list[str], messageBlocks: list, threadId: str = None) -> SlackResponse | list[SlackResponse]:
         responses: list[SlackResponse] = []
 
         if isinstance(channelList, str):
@@ -84,7 +84,7 @@ def postMessageBlocks(cls, channelList: str | list[str], messageBlocks: list, th
 
         return responses if len(responses) > 1 else responses[0]
 
-    def postMessageText(cls, channelList: str | list[str], message: str, attachments=None, threadId: str=None) -> SlackResponse | list[SlackResponse]:
+    def postMessageText(cls, channelList: str | list[str], message: str, attachments=None, threadId: str = None) -> SlackResponse | list[SlackResponse]:
         responses: list[SlackResponse] = []
 
         if isinstance(channelList, str):