[patch] Merge branch 'stable' into MASR6491

Author: Bhautik Vala

src/mas/devops/ocp.py

@@ -332,3 +332,68 @@ def execInPod(core_v1_api: client.CoreV1Api, pod_name: str, namespace, command:
     logger.debug(f"stdout: \n----------------------------------------------------------------\n{stdout}\n----------------------------------------------------------------\n")
 
     return stdout
+
+
+def updateGlobalPullSecret(dynClient: DynamicClient, registryUrl: str, username: str, password: str) -> dict:
+    """
+    Update the global pull secret in openshift-config namespace with new registry credentials.
+
+    Args:
+        dynClient: OpenShift Dynamic Client
+        registryUrl: Registry URL (e.g., "myregistry.com:5000")
+        username: Registry username
+        password: Registry password
+
+    Returns:
+        dict: Updated secret information
+    """
+    import json
+    import base64
+
+    logger.info(f"Updating global pull secret with credentials for {registryUrl}")
+
+    # Get the existing pull secret
+    secretsAPI = dynClient.resources.get(api_version="v1", kind="Secret")
+    try:
+        pullSecret = secretsAPI.get(name="pull-secret", namespace="openshift-config")
+    except NotFoundError:
+        raise Exception("Global pull-secret not found in openshift-config namespace")
+
+    # Convert to dict to allow modifications
+    secretDict = pullSecret.to_dict()
+
+    # Decode the existing dockerconfigjson
+    dockerConfigJson = secretDict['data'].get(".dockerconfigjson", "")
+    dockerConfig = json.loads(base64.b64decode(dockerConfigJson).decode('utf-8'))
+
+    # Create auth string (username:password base64 encoded)
+    authString = base64.b64encode(f"{username}:{password}".encode('utf-8')).decode('utf-8')
+
+    # Add or update the registry credentials
+    if "auths" not in dockerConfig:
+        dockerConfig["auths"] = {}
+
+    dockerConfig["auths"][registryUrl] = {
+        "username": username,
+        "password": password,
+        "email": username,
+        "auth": authString
+    }
+
+    # Encode back to base64
+    updatedDockerConfig = base64.b64encode(json.dumps(dockerConfig).encode('utf-8')).decode('utf-8')
+
+    # Update the secret dict
+    secretDict['data'][".dockerconfigjson"] = updatedDockerConfig
+
+    # Apply the updated secret
+    updatedSecret = secretsAPI.apply(body=secretDict, namespace="openshift-config")
+
+    logger.info(f"Successfully updated global pull secret with credentials for {registryUrl}")
+
+    return {
+        "name": updatedSecret.metadata.name,
+        "namespace": updatedSecret.metadata.namespace,
+        "registry": registryUrl,
+        "changed": True
+    }

src/mas/devops/users.py

@@ -759,17 +759,23 @@ def parse_initial_users_from_aws_secret_json(self, secret_json):
         for (email, csv) in secret_json.items():
             values = csv.split(",")
 
-            if len(values) != 3:
-                raise Exception(f"Wrong number of CSV values for {email} (expected 3 but got {len(values)})")
+            if len(values) != 3 and len(values) != 4:
+                raise Exception(f"Wrong number of CSV values for {email} (expected 3 or 4 but got {len(values)})")
 
             user_type = values[0].strip()
             given_name = values[1].strip()
             family_name = values[2].strip()
 
+            if len(values) == 4:
+                id = values[3].strip()
+            else:
+                id = email
+
             user = {
                 "email": email,
                 "given_name": given_name,
-                "family_name": family_name
+                "family_name": family_name,
+                "id": id
             }
             if user_type == "primary":
                 primary.append(user)
@@ -817,7 +823,7 @@ def create_initial_users_for_saas(self, initial_users):
         for primary_user in primary_users:
             self.logger.info("")
             try:
-                self.logger.info(f"Syncing primary user {primary_user['email']}")
+                self.logger.info(f"Syncing primary user with email {primary_user['email']}")
                 self.create_initial_user_for_saas(primary_user, "PRIMARY")
                 completed.append(primary_user)
                 self.logger.info(f"Completed sync of primary user {primary_user['email']}")
@@ -829,7 +835,7 @@ def create_initial_users_for_saas(self, initial_users):
             self.logger.info("")
             try:
                 self.logger.info("")
-                self.logger.info(f"Syncing secondary user {secondary_user['email']}")
+                self.logger.info(f"Syncing secondary user with email {secondary_user['email']}")
                 self.create_initial_user_for_saas(secondary_user, "SECONDARY")
                 completed.append(secondary_user)
                 self.logger.info(f"Completed sync of secondary user {secondary_user['email']}")
@@ -855,8 +861,13 @@ def create_initial_user_for_saas(self, user, user_type):
         user_given_name = user["given_name"]
         user_family_name = user["family_name"]
 
-        user_id = user_email
-        username = user_email
+        if "id" in user:
+            user_id = user["id"]
+        else:
+            # default to email if no id provided
+            user_id = user_email
+
+        username = user_id
         # display_name = re.search('^([^@]+)@', user_email).group(1) # local part of the email
         display_name = f"{user_given_name} {user_family_name}"
 
@@ -874,6 +885,8 @@ def create_initial_user_for_saas(self, user, user_type):
             }
             is_workspace_admin = True
             application_role = "ADMIN"
+            facilities_role = "PREMIUM"
+            manage_role = "MANAGEUSER"
             # TODO: check which security groups primary users should be members of
             manage_security_groups = ["MAXADMIN"]
         elif user_type == "SECONDARY":
@@ -889,6 +902,8 @@ def create_initial_user_for_saas(self, user, user_type):
             }
             is_workspace_admin = False
             application_role = "USER"
+            facilities_role = "BASE"
+            manage_role = "MANAGEUSER"
             # TODO: check which security groups secondary users should be members of
             manage_security_groups = []
         else:
@@ -906,6 +921,8 @@ def create_initial_user_for_saas(self, user, user_type):
                     "primary": True
                 }
             ],
+            "phoneNumbers": [],
+            "addresses": [],
             "displayName": display_name,
             "issuer": "local",
             "permissions": permissions,
@@ -921,8 +938,9 @@ def create_initial_user_for_saas(self, user, user_type):
         for mas_application_id in self.mas_workspace_application_ids:
             self.await_mas_application_availability(mas_application_id)
             if mas_application_id == "manage":
-                # special case for manage; role is always "MANAGEUSER"
-                role = "MANAGEUSER"
+                role = manage_role
+            elif mas_application_id == "facilities":
+                role = facilities_role
             else:
                 # otherwise grant the user the appropriate role for their user_type
                 role = application_role

test/src/test_users.py

@@ -1573,7 +1573,8 @@ def test_parse_initial_users_from_aws_secret_json(user_utils):
         {
             "user1@example.com": "primary,joe,bloggs",
             "user2@example.com": "  primary     ,   ben     ,   bob  ",
-            "user3@example.com": "secondary     ,bill,  bibb"
+            "user3@example.com": "secondary     ,bill,  bibb",
+            "user4@example.com": "primary     ,bab,  bub,user4"
         }
     )
 
@@ -1583,19 +1584,28 @@ def test_parse_initial_users_from_aws_secret_json(user_utils):
                 {
                     "email": "user1@example.com",
                     "given_name": "joe",
-                    "family_name": "bloggs"
+                    "family_name": "bloggs",
+                    "id": "user1@example.com",
                 },
                 {
                     "email": "user2@example.com",
                     "given_name": "ben",
-                    "family_name": "bob"
+                    "family_name": "bob",
+                    "id": "user2@example.com",
+                },
+                {
+                    "email": "user4@example.com",
+                    "given_name": "bab",
+                    "family_name": "bub",
+                    "id": "user4",
                 }
             ],
             "secondary": [
                 {
                     "email": "user3@example.com",
                     "given_name": "bill",
-                    "family_name": "bibb"
+                    "family_name": "bibb",
+                    "id": "user3@example.com",
                 }
             ]
         }
@@ -1607,7 +1617,7 @@ def test_parse_initial_users_from_aws_secret_json(user_utils):
         user_utils.parse_initial_users_from_aws_secret_json({
             "user1@example.com": "primary"
         })
-    assert "Wrong number of CSV values for user1@example.com (expected 3 but got 1)" == str(excinfo.value)
+    assert "Wrong number of CSV values for user1@example.com (expected 3 or 4 but got 1)" == str(excinfo.value)
 
     with pytest.raises(Exception) as excinfo:
         user_utils.parse_initial_users_from_aws_secret_json({
@@ -1642,32 +1652,64 @@ def test_create_initial_user_for_saas_unsupported_type(user_utils):
 # Assisted by watsonx Code Assistant
 
 
-@pytest.mark.parametrize("user_type, permissions, entitlement, is_workspace_admin, application_role, manage_security_groups", [
+@pytest.mark.parametrize("user_type, user_id, user_email, permissions, entitlement, is_workspace_admin, application_role, manage_role, facilities_role, manage_security_groups", [
     (
         "PRIMARY",
+        None,
+        "bill.bob@acme.com",
         {"systemAdmin": False, "userAdmin": True, "apikeyAdmin": False},
         {"application": "PREMIUM", "admin": "ADMIN_BASE", "alwaysReserveLicense": True},
         True,
         "ADMIN",
+        "MANAGEUSER",
+        "PREMIUM",
         ["MAXADMIN"]
     ),
+    (
+        "PRIMARY",
+        "billbob",
+        "bill.bob@acme.com",
+        {"systemAdmin": False, "userAdmin": True, "apikeyAdmin": False},
+        {"application": "PREMIUM", "admin": "ADMIN_BASE", "alwaysReserveLicense": True},
+        True,
+        "ADMIN",
+        "MANAGEUSER",
+        "PREMIUM",
+        ["MAXADMIN"]
+    ),
+    (
+        "SECONDARY",
+        None,
+        "bab.bon@acme.com",
+        {"systemAdmin": False, "userAdmin": False, "apikeyAdmin": False},
+        {"application": "BASE", "admin": "NONE", "alwaysReserveLicense": True},
+        False,
+        "USER",
+        "MANAGEUSER",
+        "BASE",
+        []
+    ),
     (
         "SECONDARY",
+        "babbon",
+        "bab.bon@acme.com",
         {"systemAdmin": False, "userAdmin": False, "apikeyAdmin": False},
         {"application": "BASE", "admin": "NONE", "alwaysReserveLicense": True},
         False,
         "USER",
+        "MANAGEUSER",
+        "BASE",
         []
     )
 ])
 def test_create_initial_user_for_saas(
-    user_type, permissions, entitlement, is_workspace_admin, application_role, manage_security_groups,
+    user_type, user_id, user_email, permissions, entitlement, is_workspace_admin, application_role, manage_role, facilities_role, manage_security_groups,
     user_utils, requests_mock
 ):
     user_utils.get_or_create_user = MagicMock()
     user_utils.link_user_to_local_idp = MagicMock()
     user_utils.add_user_to_workspace = MagicMock()
-    mas_workspace_application_ids = ["manage", "iot"]
+    mas_workspace_application_ids = ["manage", "iot", "facilities"]
     user_utils.get_mas_applications_in_workspace = MagicMock(return_value=map(lambda x: {"id": x}, mas_workspace_application_ids))
     user_utils.await_mas_application_availability = MagicMock()
     user_utils.set_user_application_permission = MagicMock()
@@ -1676,20 +1718,24 @@ def test_create_initial_user_for_saas(
     user_utils.create_or_get_manage_api_key_for_user = MagicMock(return_value=manage_api_key)
     user_utils.add_user_to_manage_group = MagicMock()
 
-    user_email = "bill.bob@acme.com"
     user_given_name = "billy"
     user_family_name = "bobby"
-    user_id = user_email
-    username = user_email
     display_name = f"{user_given_name} {user_family_name}"
 
-    user_utils.create_initial_user_for_saas({
+    initial_users = {
         "email": user_email,
         "given_name": user_given_name,
         "family_name": user_family_name
-    },
-        user_type
-    )
+    }
+
+    if user_id is None:
+        user_id = user_email
+    else:
+        initial_users["id"] = user_id
+
+    username = user_id
+
+    user_utils.create_initial_user_for_saas(initial_users, user_type)
 
     user_utils.get_or_create_user.assert_called_once_with({
         "id": user_id,
@@ -1703,6 +1749,8 @@ def test_create_initial_user_for_saas(
                 "primary": True
             }
         ],
+        "phoneNumbers": [],
+        "addresses": [],
         "displayName": display_name,
         "issuer": "local",
         "permissions": permissions,
@@ -1714,12 +1762,14 @@ def test_create_initial_user_for_saas(
     user_utils.add_user_to_workspace.assert_called_once_with(user_id, is_workspace_admin=is_workspace_admin)
     user_utils.await_mas_application_availability.assert_has_calls([call("manage"), call("iot")])
     user_utils.set_user_application_permission.assert_has_calls([
-        call(user_id, "manage", "MANAGEUSER"),
+        call(user_id, "manage", manage_role),
         call(user_id, "iot", application_role),
+        call(user_id, "facilities", facilities_role),
     ])
     user_utils.check_user_sync.assert_has_calls([
         call(user_id, "manage"),
-        call(user_id, "iot")
+        call(user_id, "iot"),
+        call(user_id, "facilities")
     ])
 
     if len(manage_security_groups) > 0: