@W-18129346 - Adding new tasks get_assignable_permission_sets and get_assignable_licenses

Author: vsbharath

.github/workflows/release_test.yml

@@ -34,7 +34,7 @@ jobs:
                   pip uninstall -y cumulusci
             - name: Store artifacts
               if: failure()
-              uses: actions/upload-artifact@v3
+              uses: actions/upload-artifact@v4
               with:
                   name: packages
                   path: dist

.github/workflows/release_test_sfdx.yml

@@ -47,6 +47,12 @@ jobs:
                   python-version: 3.11
                   cache: pip
                   cache-dependency-path: "pyproject.toml"
+            - name: Check pip cache existence
+              run: |
+                  if [ ! -d ~/.cache/pip ]; then
+                      echo "No pip cache directory found, skipping cache step."
+                  fi
+
             - name: Set up uv
               uses: SFDO-Tooling/setup-uv@main
               with:

cumulusci/cumulusci.yml

@@ -317,6 +317,10 @@ tasks:
         description: Retrieves a list of the currently available license definition keys
         class_path: cumulusci.tasks.preflight.licenses.GetAvailableLicenses
         group: Salesforce Preflight Checks
+    get_assignable_licenses:
+        description: Retrieves a list of the currently assignable license definition keys based on unused licenses
+        class_path: cumulusci.tasks.preflight.licenses.GetAssignableLicenses
+        group: Salesforce Preflight Checks
     get_available_permission_set_licenses:
         description: Retrieves a list of the currently available Permission Set License definition keys
         class_path: cumulusci.tasks.preflight.licenses.GetAvailablePermissionSetLicenses
@@ -333,6 +337,10 @@ tasks:
         description: Retrieves a list of the currently available Permission Sets
         class_path: cumulusci.tasks.preflight.licenses.GetAvailablePermissionSets
         group: Salesforce Preflight Checks
+    get_assignable_permission_sets:
+        description: Retrieves a list of the currently assignable Permission Sets based on unused associated user licenses
+        class_path: cumulusci.tasks.preflight.licenses.GetAssignablePermissionSets
+        group: Salesforce Preflight Checks
     get_existing_record_types:
         description: "Retrieves all Record Types in the org as a dict, with sObject names as keys and lists of Developer Names as values."
         class_path: cumulusci.tasks.preflight.recordtypes.CheckSObjectRecordTypes

cumulusci/tasks/preflight/licenses.py

@@ -1,18 +1,40 @@
 from cumulusci.tasks.salesforce import BaseSalesforceApiTask
 
 
-class GetAvailableLicenses(BaseSalesforceApiTask):
+class BaseUserLicenseAwareTask(BaseSalesforceApiTask):
+    def get_available_user_licenses(self, is_assignable=False):
+        """Fetch active user licenses with availability."""
+        query = "SELECT Id, LicenseDefinitionKey, TotalLicenses, UsedLicenses FROM UserLicense WHERE Status = 'Active'"
+        return {
+            lic["Id"]: lic
+            for lic in self.sf.query(query)["records"]
+            if not is_assignable or (lic["TotalLicenses"] > lic["UsedLicenses"])
+        }
+
+    def _log_list(self, title, items):
+        self.logger.info(f"{title} ({len(items)}):\n" + "\n".join(f"- {item}" for item in items))
+
+
+class GetAvailableLicenses(BaseUserLicenseAwareTask):
     def _run_task(self):
         self.return_values = [
             result["LicenseDefinitionKey"]
-            for result in self.sf.query("SELECT LicenseDefinitionKey FROM UserLicense")[
-                "records"
-            ]
+            for result in self.get_available_user_licenses().values()
         ]
         licenses = "\n".join(self.return_values)
         self.logger.info(f"Found licenses:\n{licenses}")
 
 
+class GetAssignableLicenses(BaseUserLicenseAwareTask):
+    def _run_task(self):
+        self.return_values = [
+            result["LicenseDefinitionKey"]
+            for result in self.get_available_user_licenses(is_assignable=True).values()
+        ]
+        licenses = "\n".join(self.return_values)
+        self.logger.info(f"Found assignable licenses:\n{licenses}")
+
+
 class GetAvailablePermissionSetLicenses(BaseSalesforceApiTask):
     def _run_task(self):
         query = "SELECT PermissionSetLicenseKey FROM PermissionSetLicense WHERE Status = 'Active'"
@@ -43,3 +65,18 @@ def _run_task(self):
         ]
         permsets = "\n".join(self.return_values)
         self.logger.info(f"Found Permission Sets:\n{permsets}")
+
+
+class GetAssignablePermissionSets(BaseUserLicenseAwareTask):
+    def _run_task(self):
+        license_data = self.get_available_user_licenses(is_assignable=True)
+        permsets = self.sf.query_all("SELECT LicenseId, Name FROM PermissionSet")["records"]
+        available_permsets = [
+            ps["Name"]
+            for ps in permsets
+            if not ps["LicenseId"] or ps["LicenseId"] in license_data
+        ]
+
+        self.return_values = available_permsets
+        permsets = "\n".join(self.return_values)
+        self.logger.info(f"Found assignable permission sets:\n{permsets}")

cumulusci/tasks/preflight/tests/test_licenses.py

@@ -1,8 +1,10 @@
 from unittest.mock import Mock
 
 from cumulusci.tasks.preflight.licenses import (
+    GetAssignableLicenses,
     GetAvailableLicenses,
     GetAvailablePermissionSetLicenses,
+    GetAssignablePermissionSets,
     GetAvailablePermissionSets,
     GetPermissionLicenseSetAssignments,
 )
@@ -12,20 +14,32 @@
 class TestLicensePreflights:
     def test_license_preflight(self):
         task = create_task(GetAvailableLicenses, {})
+        task.get_available_user_licenses = Mock(return_value={
+            "L1": {"LicenseDefinitionKey": "TEST1"},
+            "L3": {"LicenseDefinitionKey": "TEST3"},
+        })
+
+        task()
+        # Only TEST1 and TEST3 have available licenses
+        assert task.return_values == ["TEST1", "TEST3"]
+
+    def test_assignable_license_preflight(self):
+        task = create_task(GetAssignableLicenses, {})
         task._init_api = Mock()
         task._init_api.return_value.query.return_value = {
             "totalSize": 2,
             "records": [
-                {"LicenseDefinitionKey": "TEST1"},
-                {"LicenseDefinitionKey": "TEST2"},
+                {"Id": "L1", "LicenseDefinitionKey": "TEST1", "TotalLicenses": 100, "UsedLicenses": 90},
+                {"Id": "L2", "LicenseDefinitionKey": "TEST2", "TotalLicenses": 100, "UsedLicenses": 100},
             ],
         }
-        task()
 
+        task()
         task._init_api.return_value.query.assert_called_once_with(
-            "SELECT LicenseDefinitionKey FROM UserLicense"
+            "SELECT Id, LicenseDefinitionKey, TotalLicenses, UsedLicenses FROM UserLicense WHERE Status = 'Active'"
         )
-        assert task.return_values == ["TEST1", "TEST2"]
+        # Only TEST1 and TEST3 have available licenses
+        assert task.return_values == ["TEST1"]
 
     def test_psl_preflight(self):
         task = create_task(GetAvailablePermissionSetLicenses, {})
@@ -61,7 +75,7 @@ def test_assigned_permsetlicense_preflight(self):
                 {
                     "PermissionSetLicense": {
                         "MasterLabel": "Einstein Analytics Plus Admin",
-                        "DeveloperName": "EinsteinAnalyticsPlusAdmin",
+                        "DeveloperName": "EinsteinAnalyticsPlusAdmin"
                     },
                 },
             ],
@@ -93,3 +107,31 @@ def test_permsets_preflight(self):
             "SELECT Name FROM PermissionSet"
         )
         assert task.return_values == ["TEST1", "TEST2"]
+
+    def test_assignable_permsets_preflight(self):
+        task = create_task(GetAssignablePermissionSets, {})
+        task._init_api = Mock()
+        task._init_api.return_value.query.return_value = {
+            "totalSize": 2,
+            "records": [
+                {"Id": "L1", "LicenseDefinitionKey": "TEST1", "TotalLicenses": 100, "UsedLicenses": 90},
+                {"Id": "L2", "LicenseDefinitionKey": "TEST2", "TotalLicenses": 100, "UsedLicenses": 100},
+            ],
+        }
+        task._init_api.return_value.query_all.return_value = {
+            "totalSize": 3,
+            "records": [
+                {"LicenseId": "L1", "Name": "TEST1"},
+                {"LicenseId": "L2", "Name": "TEST2"},
+                {"LicenseId": None, "Name": "TEST3"},
+            ],
+        }
+        task()
+
+        task._init_api.return_value.query.assert_called_once_with(
+            "SELECT Id, LicenseDefinitionKey, TotalLicenses, UsedLicenses FROM UserLicense WHERE Status = 'Active'"
+        )
+        task._init_api.return_value.query_all.assert_called_once_with(
+            "SELECT LicenseId, Name FROM PermissionSet"
+        )
+        assert task.return_values == ["TEST1", "TEST3"]