Add new tasks get_assignable_permission_sets and get_assignable_licenses

.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

@@ -45,8 +45,6 @@ jobs:
               uses: actions/setup-python@v5
               with:
                   python-version: 3.11
-                  cache: pip
-                  cache-dependency-path: "pyproject.toml"
             - 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,42 @@
 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 +67,20 @@ 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,6 +1,8 @@
 from unittest.mock import Mock
 
 from cumulusci.tasks.preflight.licenses import (
+    GetAssignableLicenses,
+    GetAssignablePermissionSets,
     GetAvailableLicenses,
     GetAvailablePermissionSetLicenses,
     GetAvailablePermissionSets,
@@ -16,17 +18,56 @@ def test_license_preflight(self):
         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"]
 
+    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": [
+                {
+                    "Id": "L1",
+                    "LicenseDefinitionKey": "TEST1",
+                    "TotalLicenses": 100,
+                    "UsedLicenses": 90,
+                },
+                {
+                    "Id": "L2",
+                    "LicenseDefinitionKey": "TEST2",
+                    "TotalLicenses": 100,
+                    "UsedLicenses": 100,
+                },
+            ],
+        }
+
+        task()
+        task._init_api.return_value.query.assert_called_once_with(
+            "SELECT Id, LicenseDefinitionKey, TotalLicenses, UsedLicenses FROM UserLicense WHERE Status = 'Active'"
+        )
+        # Only TEST1 assignable licenses
+        assert task.return_values == ["TEST1"]
+
     def test_psl_preflight(self):
         task = create_task(GetAvailablePermissionSetLicenses, {})
         task._init_api = Mock()
@@ -93,3 +134,41 @@ 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"]