Repository: SFDO-Tooling/CumulusCI
Pull request #3672 (2023-10-09 11:46:04)

Task to retrieve a complete Profile from an org
[W-8932343](https://gus.lightning.force.com/lightning/r/ADM_Work__c/a07B00000097PhOIAU/view) - Build a task to retrieve a complete Profile from an org

Permissionable Entities Retrieved:
1. CustomApplication
2. ApexClass
3. CustomPermission
4. CustomObject
5. ExternalDataSource
6. Flow
7. ApexPage
8. CustomTab

Issues:
1. I'm able to retrieve flows within the same package.xml, but they are not visible under the Profile metadata. **(Added functionality to explicitly add them to the profile-meta.xml file)**
2. While retrieving apps, those without a StartUrl in the AppMenuItem table are not being included in the retrieval process. This seems to be a package.xml issue, not something within our control.
3. Two sObjects, AIRecordInsight and Document, are not being successfully retrieved even when specified in the package.xml. **(Edit: However, they show up under the ObjectPermissions in the profile-meta.xml file, so no issues)**



Patch:
diff --git a/cumulusci/cumulusci.yml b/cumulusci/cumulusci.yml
index 6a3bf9842..301fe6adf 100644
--- a/cumulusci/cumulusci.yml
+++ b/cumulusci/cumulusci.yml
@@ -200,6 +200,10 @@ tasks:
         options:
             license: Salesforce
         group: Salesforce Metadata
+    retrieve_profile:
+        description: Given a list of profiles, the task retrieves all complete profiles along with their associated dependencies for all permissionable entities - ApexClass, ApexPage, CustomApplications, CustomObjects, CustomPermissions, CustomTabs, ExternalDataSources and Flows
+        class_path: cumulusci.tasks.salesforce.retrieve_profile.RetrieveProfile
+        group: Salesforce Metadata
     delete_data:
         description: Query existing data for a specific sObject and perform a Bulk API delete of all matching records.
         class_path: cumulusci.tasks.bulkdata.DeleteData
diff --git a/cumulusci/salesforce_api/retrieve_profile_api.py b/cumulusci/salesforce_api/retrieve_profile_api.py
new file mode 100644
index 000000000..4b100ae54
--- /dev/null
+++ b/cumulusci/salesforce_api/retrieve_profile_api.py
@@ -0,0 +1,295 @@
+import json
+from typing import Dict, List, Union
+
+import salesforce_bulk
+from requests.exceptions import ConnectionError
+from simple_salesforce.exceptions import SalesforceGeneralError
+
+from cumulusci.tasks.salesforce.BaseSalesforceApiTask import BaseSalesforceApiTask
+from cumulusci.utils.parallel.queries_in_parallel.run_queries_in_parallel import (
+    RunParallelQueries,
+)
+
+
+class MetadataInfo:
+    def __init__(
+        self, columns: List[str], table_name: str, package_xml_name: str, id_field: str
+    ):
+        self.columns = columns
+        self.table_name = table_name
+        self.package_xml_name = package_xml_name
+        self.id_field = id_field
+
+
+APEXCLASS = MetadataInfo(
+    columns=["Name", "NamespacePrefix"],
+    table_name="ApexClass",
+    package_xml_name="ApexClass",
+    id_field="Id",
+)
+APEXPAGE = MetadataInfo(
+    columns=["Name", "NamespacePrefix"],
+    table_name="ApexPage",
+    package_xml_name="ApexPage",
+    id_field="Id",
+)
+CUSTOMPERMISSION = MetadataInfo(
+    columns=["DeveloperName", "NamespacePrefix"],
+    table_name="CustomPermission",
+    package_xml_name="CustomPermission",
+    id_field="Id",
+)
+TABSET = MetadataInfo(
+    columns=["Name", "NamespacePrefix"],
+    table_name="AppMenuItem",
+    package_xml_name="CustomApplication",
+    id_field="ApplicationId",
+)
+CONNECTEDAPPLICATION = MetadataInfo(
+    columns=["Name", "NamespacePrefix"],
+    table_name="AppMenuItem",
+    package_xml_name="CustomApplication",
+    id_field="ApplicationId",
+)
+EXTERNALDATASOURCE = MetadataInfo(
+    columns=["DeveloperName", "NamespacePrefix"],
+    table_name="ExternalDataSource",
+    package_xml_name="ExternalDataSource",
+    id_field="Id",
+)
+FLOWDEFINITION = MetadataInfo(
+    columns=["ApiName"],
+    table_name="FlowDefinitionView",
+    package_xml_name="Flow",
+    id_field="Id",
+)
+
+SETUP_ENTITY_TYPES = {
+    "ApexClass": APEXCLASS,
+    "ApexPage": APEXPAGE,
+    "CustomPermission": CUSTOMPERMISSION,
+    "TabSet": TABSET,
+    "ConnectedApplication": CONNECTEDAPPLICATION,
+    "ExternalDataSource": EXTERNALDATASOURCE,
+    "FlowDefinition": FLOWDEFINITION,
+}
+SETUP_ENTITY_QUERY_NAME = "setupEntityAccess"
+
+SOBJECT_TYPE = "CustomObject"
+SOBJECT_QUERY_NAME = "sObject"
+
+CUSTOM_TAB_TYPE = "CustomTab"
+CUSTOM_TAB_QUERY_NAME = "customTab"
+
+PROFILE_FLOW_QUERY_NAME = "profileFlow"
+
+
+class RetrieveProfileApi(BaseSalesforceApiTask):
+    def _init_task(self):
+        super(RetrieveProfileApi, self)._init_task()
+        self.api_version = self.org_config.latest_api_version
+
+    def _retrieve_existing_profiles(self, profiles: List[str]):
+        query = self._build_query(["Name"], "Profile", {"Name": profiles})
+        result = self._run_query(query)
+
+        existing_profiles = []
+        for data in result["records"]:
+            existing_profiles.append(data["Name"])
+
+        return existing_profiles
+
+    def _run_query(self, query):
+        try:
+            result = self.sf.query(query)
+        except (SalesforceGeneralError, ConnectionError):
+            result = self._run_bulk_query(query)
+        return result
+
+    def _run_bulk_query(self, query):
+        table_name = self._extract_table_name_from_query(query)
+        job = self.bulk.create_query_job(table_name, contentType="JSON")
+
+        batch = self.bulk.query(job, query)
+        self.bulk.wait_for_batch(job, batch)
+        self.bulk.close_job(job)
+        results = self.bulk.get_all_results_for_query_batch(batch)
+        for result in results:
+            result = json.load(salesforce_bulk.util.IteratorBytesIO(result))
+            return {"records": result}
+
+    def _extract_table_name_from_query(self, query):
+        from_index = query.upper().find("FROM")
+        if from_index != -1:
+            query = query[from_index + len("FROM") :].strip()
+            table_name = query.split()[0]
+            return table_name
+        else:
+            raise ValueError("Invalid query format. 'FROM' clause not found.")
+
+    def _build_query(
+        self,
+        columns: List[str],
+        table_name: str,
+        where: Dict[str, Union[str, List[str]]] = None,
+    ) -> str:
+        if not columns:
+            raise ValueError("Columns list cannot be empty")
+        if not table_name:
+            raise ValueError("Table name cannot be empty")
+
+        select_clause = ", ".join(columns)
+
+        where_clause = ""
+        if where:
+            where_list = []
+            for key, value in where.items():
+                if isinstance(value, list):
+                    value = ", ".join([f"'{item}'" for item in value])
+                    condition = f"{key} IN ({value})"
+                else:
+                    condition = f"{key} = '{value}'"
+                where_list.append(condition)
+            where_clause = "WHERE " + " AND ".join(where_list)
+
+        query = f"SELECT {select_clause} FROM {table_name} {where_clause}".strip()
+        return query
+
+    def _queries_retrieve_permissions(self, profiles: List[str]):
+        queries = {}
+
+        # Setup Entity Access query
+        setupEntityAccess_query = self._build_query(
+            ["SetupEntityId", "SetupEntityType"],
+            "SetupEntityAccess",
+            {"Parent.Profile.Name": profiles},
+        )
+        queries[SETUP_ENTITY_QUERY_NAME] = setupEntityAccess_query
+
+        # sObject query
+        sObject_query = self._build_query(
+            ["SObjectType"], "ObjectPermissions", {"Parent.Profile.Name": profiles}
+        )
+        queries[SOBJECT_QUERY_NAME] = sObject_query
+
+        # Custom Tab query
+        customTab_query = self._build_query(
+            ["Name"], "PermissionSetTabSetting", {"Parent.Profile.Name": profiles}
+        )
+        queries[CUSTOM_TAB_QUERY_NAME] = customTab_query
+
+        # Matching Profile Name and Flow query
+        profileFlow_query = self._build_query(
+            ["SetupEntityId", "Parent.Profile.Name"],
+            "SetupEntityAccess",
+            {"Parent.Profile.Name": profiles, "SetupEntityType": "FlowDefinition"},
+        )
+        queries[PROFILE_FLOW_QUERY_NAME] = profileFlow_query
+
+        return queries
+
+    def _process_all_results(self, result: dict):
+        permissionable_entities = {}
+        entities, result_setupEntityAccess = self._process_setupEntityAccess_results(
+            result[SETUP_ENTITY_QUERY_NAME]
+        )
+        permissionable_entities.update(entities)
+        permissionable_entities.update(
+            self._process_sObject_results(result[SOBJECT_QUERY_NAME])
+        )
+        permissionable_entities.update(
+            self._process_customTab_results(result[CUSTOM_TAB_QUERY_NAME])
+        )
+
+        # Process the profile and flows
+        profile_flow = self._match_profiles_and_flows(
+            result[PROFILE_FLOW_QUERY_NAME],
+            result_setupEntityAccess.get("FlowDefinition", []),
+        )
+
+        return permissionable_entities, profile_flow
+
+    # Retrieve all the permissionable entitites for a set of profiles
+    def _retrieve_permissionable_entities(self, profiles: List[str]):
+        # Logs
+        self.logger.info("Querying for all permissionable entities:")
+        self.logger.info("Pending")
+
+        # Run all queries
+        result = RunParallelQueries._run_queries_in_parallel(
+            self._queries_retrieve_permissions(profiles), self._run_query
+        )
+
+        # Process the results
+        permissionable_entities, profile_flow = self._process_all_results(result)
+
+        # Logs
+        self.logger.info("[Done]\n")
+
+        return permissionable_entities, profile_flow
+
+    def _process_setupEntityAccess_results(self, result_list: List[dict]):
+        setupEntityAccess_dict = {
+            entity_type: [] for entity_type in SETUP_ENTITY_TYPES.keys()
+        }
+
+        for data in result_list:
+            entity_type = data["SetupEntityType"]
+            entity_id = data["SetupEntityId"]
+            if entity_type in SETUP_ENTITY_TYPES.keys():
+                setupEntityAccess_dict[entity_type].append(entity_id)
+
+        queries = {}
+        for entity_type, query_values in SETUP_ENTITY_TYPES.items():
+            if query_values and len(setupEntityAccess_dict[entity_type]) != 0:
+                queries[entity_type] = self._build_query(
+                    query_values.columns,
+                    query_values.table_name,
+                    {query_values.id_field: setupEntityAccess_dict[entity_type]},
+                )
+
+        result = RunParallelQueries._run_queries_in_parallel(queries, self._run_query)
+
+        extracted_values = {}
+        for entity_type, data in SETUP_ENTITY_TYPES.items():
+            if entity_type in result and data:
+                extracted_values.setdefault(data.package_xml_name, [])
+                for item in result[entity_type]:
+                    if (
+                        "NamespacePrefix" in item
+                        and item["NamespacePrefix"] is not None
+                    ):
+                        extracted_values[data.package_xml_name].append(
+                            f'{item["NamespacePrefix"]}__{item[data.columns[0]]}'
+                        )
+                    else:
+                        extracted_values[data.package_xml_name].append(
+                            item[data.columns[0]]
+                        )
+
+        return extracted_values, result
+
+    def _process_sObject_results(self, result_list: List[dict]):
+        sObject_list = [data["SobjectType"] for data in result_list]
+        return {SOBJECT_TYPE: sObject_list}
+
+    def _process_customTab_results(self, result_list: List[dict]):
+        customTab_list = [data["Name"] for data in result_list]
+        return {CUSTOM_TAB_TYPE: customTab_list}
+
+    def _match_profiles_and_flows(
+        self, result_profiles: List[dict], result_flows: List[dict]
+    ):
+        profile_mapping = {}
+        for item in result_profiles:
+            setup_entity_id = item["SetupEntityId"]
+            profile_name = item["Parent"]["Profile"]["Name"]
+            profile_mapping.setdefault(setup_entity_id, []).append(profile_name)
+
+        result_dict = {}
+        for flow in result_flows:
+            setup_entity_id = flow["attributes"]["url"].split("/")[-1]
+            for profile_name in profile_mapping.get(setup_entity_id, []):
+                result_dict.setdefault(profile_name, []).append(flow["ApiName"])
+
+        return result_dict
diff --git a/cumulusci/tasks/salesforce/retrieve_profile.py b/cumulusci/tasks/salesforce/retrieve_profile.py
new file mode 100644
index 000000000..b9bb3ece8
--- /dev/null
+++ b/cumulusci/tasks/salesforce/retrieve_profile.py
@@ -0,0 +1,172 @@
+import os
+
+from cumulusci.core.utils import process_bool_arg, process_list_arg
+from cumulusci.salesforce_api.metadata import ApiRetrieveUnpackaged
+from cumulusci.salesforce_api.retrieve_profile_api import RetrieveProfileApi
+from cumulusci.tasks.salesforce.BaseSalesforceMetadataApiTask import (
+    BaseSalesforceMetadataApiTask,
+)
+
+
+class RetrieveProfile(BaseSalesforceMetadataApiTask):
+    api_class = ApiRetrieveUnpackaged
+    task_options = {
+        "profiles": {
+            "description": "List of profile API names that you want to retrieve",
+            "required": True,
+        },
+        "path": {
+            "description": "Target folder path. By default, it uses force-app/main/default",
+        },
+        "strict_mode": {
+            "description": "When set to False, enables leniency by ignoring missing profiles when provided with a list of profiles."
+            " When set to True, enforces strict validation, causing a failure if any profile is not present in the list."
+            " Default is True",
+            "required": False,
+        },
+    }
+
+    def _init_options(self, kwargs):
+        super(RetrieveProfile, self)._init_options(kwargs)
+        self.api_version = self.org_config.latest_api_version
+        self.profiles = process_list_arg(self.options["profiles"])
+        if not self.profiles:
+            raise ValueError("At least one profile must be specified.")
+
+        self.extract_dir = self.options.get("path", "force-app/default/main")
+
+        if not os.path.exists(self.extract_dir):
+            raise FileNotFoundError(
+                f"The extract directory '{self.extract_dir}' does not exist."
+            )
+
+        if not os.path.isdir(self.extract_dir):
+            raise NotADirectoryError(f"'{self.extract_dir}' is not a directory.")
+
+        self.strictMode = process_bool_arg(self.options.get("strict_mode", True))
+
+    def _check_existing_profiles(self, retrieve_profile_api_task):
+        # Check for existing profiles
+        self.existing_profiles = retrieve_profile_api_task._retrieve_existing_profiles(
+            self.profiles
+        )
+        self.missing_profiles = set(self.profiles) - set(self.existing_profiles)
+
+        # Handle for strictMode
+        if self.missing_profiles:
+            self.logger.warning(
+                f"The following profiles were not found or could not be retrieved: '{', '.join(self.missing_profiles)}'\n"
+            )
+            if self.strictMode:
+                raise RuntimeError(
+                    "Operation failed due to missing profiles. Set strictMode to False if you want to ignore missing profiles."
+                )
+
+        # Handle for no existing profiles
+        if not self.existing_profiles:
+            raise RuntimeError("None of the profiles given were found.")
+
+    def add_flow_accesses(self, profile_content, flows):
+        # Find the position of the closing </Profile> tag
+        profile_end_position = profile_content.find("</Profile>")
+
+        if profile_end_position != -1:
+            flow_accesses_xml = "".join(
+                [
+                    f"    <flowAccesses>\n"
+                    f"        <enabled>true</enabled>\n"
+                    f"        <flow>{flow}</flow>\n"
+                    f"    </flowAccesses>\n"
+                    for flow in flows
+                ]
+            )
+            modified_content = (
+                profile_content[:profile_end_position]
+                + flow_accesses_xml
+                + profile_content[profile_end_position:]
+            )
+            return modified_content
+
+        return profile_content
+
+    def save_profile_file(self, extract_dir, filename, content):
+        profile_path = os.path.join(extract_dir, filename)
+        os.makedirs(os.path.dirname(profile_path), exist_ok=True)
+        with open(profile_path, "w", encoding="utf-8") as updated_profile_file:
+            updated_profile_file.write(content)
+
+    def _run_task(self):
+        self.retrieve_profile_api_task = RetrieveProfileApi(
+            project_config=self.project_config,
+            task_config=self.task_config,
+            org_config=self.org_config,
+        )
+        self.retrieve_profile_api_task._init_task()
+        self._check_existing_profiles(self.retrieve_profile_api_task)
+        (
+            permissionable_entities,
+            profile_flows,
+        ) = self.retrieve_profile_api_task._retrieve_permissionable_entities(
+            self.existing_profiles
+        )
+        entities_to_be_retrieved = {
+            **permissionable_entities,
+            **{"Profile": self.existing_profiles},
+        }
+
+        self.package_xml = self._create_package_xml(
+            entities_to_be_retrieved, self.api_version
+        )
+        api = self._get_api()
+        zip_result = api()
+
+        for file_info in zip_result.infolist():
+            if file_info.filename.startswith(
+                "profiles/"
+            ) and file_info.filename.endswith(".profile"):
+                with zip_result.open(file_info) as profile_file:
+                    profile_content = profile_file.read().decode("utf-8")
+                    profile_name = os.path.splitext(
+                        os.path.basename(file_info.filename)
+                    )[0]
+
+                    if profile_name in profile_flows:
+                        profile_content = self.add_flow_accesses(
+                            profile_content, profile_flows[profile_name]
+                        )
+
+                    self.save_profile_file(
+                        self.extract_dir, file_info.filename, profile_content
+                    )
+
+        # zip_result.extractall('./unpackaged')
+
+        self.logger.info(
+            f"Profiles {', '.join(self.existing_profiles)} unzipped into folder '{self.extract_dir}'"
+        )
+
+    def _get_api(self):
+        # Logs
+        self.logger.info("Retrieving all entities from org:")
+
+        return self.api_class(
+            self,
+            api_version=self.api_version,
+            package_xml=self.package_xml,
+        )
+
+    def _create_package_xml(self, input_dict: dict, api_version: str):
+        package_xml = '<?xml version="1.0" encoding="UTF-8"?>\n'
+        package_xml += '<Package xmlns="http://soap.sforce.com/2006/04/metadata">\n'
+
+        for name, members in input_dict.items():
+            package_xml += "    <types>\n"
+            for member in members:
+                package_xml += f"        <members>{member}</members>\n"
+            package_xml += f"        <name>{name}</name>\n"
+            package_xml += "    </types>\n"
+
+        package_xml += f"    <version>{api_version}</version>\n"
+        package_xml += "</Package>\n"
+
+        return package_xml
diff --git a/cumulusci/utils/parallel/queries_in_parallel/run_queries_in_parallel.py b/cumulusci/utils/parallel/queries_in_parallel/run_queries_in_parallel.py
new file mode 100644
index 000000000..2db64811b
--- /dev/null
+++ b/cumulusci/utils/parallel/queries_in_parallel/run_queries_in_parallel.py
@@ -0,0 +1,29 @@
+from concurrent.futures import ThreadPoolExecutor
+from typing import Callable, Dict
+
+
+class RunParallelQueries:
+    @staticmethod
+    def _run_queries_in_parallel(
+        queries: Dict[str, str], run_query: Callable[[str], dict], num_threads: int = 4
+    ) -> Dict[str, list]:
+        """Accepts a set of queries structured as {'query_name': 'query'}
+        and a run_query function that runs a particular query. Runs queries in parallel and returns the queries"""
+        results_dict = {}
+
+        with ThreadPoolExecutor(max_workers=num_threads) as executor:
+            futures = {
+                query_name: executor.submit(run_query, query)
+                for query_name, query in queries.items()
+            }
+
+        for query_name, future in futures.items():
+            try:
+                query_result = future.result()
+                results_dict[query_name] = query_result["records"]
+            except Exception as e:
+                raise Exception(f"Error executing query '{query_name}': {type(e)}: {e}")
+            else:
+                queries.pop(query_name, None)
+
+        return results_dict


Test patch:
diff --git a/cumulusci/salesforce_api/tests/test_retrieve_profile_api.py b/cumulusci/salesforce_api/tests/test_retrieve_profile_api.py
new file mode 100644
index 000000000..177e1b1b0
--- /dev/null
+++ b/cumulusci/salesforce_api/tests/test_retrieve_profile_api.py
@@ -0,0 +1,335 @@
+from io import StringIO
+from unittest.mock import MagicMock, patch
+
+import pytest
+from requests.exceptions import ConnectionError
+
+from cumulusci.salesforce_api.retrieve_profile_api import RetrieveProfileApi
+from cumulusci.tasks.salesforce.BaseSalesforceApiTask import BaseSalesforceApiTask
+from cumulusci.utils.parallel.queries_in_parallel.run_queries_in_parallel import (
+    RunParallelQueries,
+)
+
+
+@pytest.fixture
+def retrieve_profile_api_instance():
+    sf_mock = MagicMock()
+    bulk_mock = MagicMock()
+    project_config = MagicMock()
+    task_config = MagicMock()
+    org_config = MagicMock()
+    org_config.latest_api_version = "58.0"
+    sf_mock.query.return_value = {"records": []}
+    api = RetrieveProfileApi(
+        project_config=project_config, org_config=org_config, task_config=task_config
+    )
+    api.sf = sf_mock
+    api.bulk = bulk_mock
+    return api
+
+
+def test_init_task(retrieve_profile_api_instance):
+    with patch.object(BaseSalesforceApiTask, "_init_task"):
+        retrieve_profile_api_instance._init_task()
+
+    assert retrieve_profile_api_instance.api_version == "58.0"
+
+
+def test_retrieve_existing_profiles(retrieve_profile_api_instance):
+    profiles = ["Profile1", "Profile2"]
+    result = {"records": [{"Name": "Profile1"}]}
+    with patch.object(
+        RetrieveProfileApi, "_build_query", return_value="some_query"
+    ), patch.object(RetrieveProfileApi, "_run_query", return_value=result):
+        existing_profiles = retrieve_profile_api_instance._retrieve_existing_profiles(
+            profiles
+        )
+
+    assert "Profile1" in existing_profiles
+    assert "Profile2" not in existing_profiles
+
+
+def test_run_query_sf(retrieve_profile_api_instance):
+    query = "SELECT Id FROM Account"
+    result_data = {"records": [{"Id": "001abc"}]}
+    retrieve_profile_api_instance.sf.query.return_value = result_data
+
+    result = retrieve_profile_api_instance._run_query(query)
+    assert result == result_data
+
+
+def test_run_query_bulk(retrieve_profile_api_instance):
+    query = "SELECT Id FROM Account"
+    result_data = {"records": [{"Id": "001abc"}]}
+    retrieve_profile_api_instance.sf.query.side_effect = ConnectionError
+
+    retrieve_profile_api_instance.bulk.create_query_job.return_value = "job_id"
+    retrieve_profile_api_instance.bulk.query.return_value = "batch_id"
+    retrieve_profile_api_instance.bulk.wait_for_batch.return_value = None
+    retrieve_profile_api_instance.bulk.get_all_results_for_query_batch.return_value = [
+        "some_value"
+    ]
+
+    with patch(
+        "salesforce_bulk.util.IteratorBytesIO",
+        return_value=StringIO('[{"Id": "001abc"}]'),
+    ):
+        result = retrieve_profile_api_instance._run_query(query)
+
+    assert result == result_data
+
+
+def test_extract_table_name_from_query_valid(retrieve_profile_api_instance):
+    query = "SELECT Id FROM Account"
+    table_name = retrieve_profile_api_instance._extract_table_name_from_query(query)
+    assert table_name == "Account"
+
+
+def test_extract_table_name_from_query_invalid(retrieve_profile_api_instance):
+    query = "SELECT Id"
+    with pytest.raises(ValueError):
+        retrieve_profile_api_instance._extract_table_name_from_query(query)
+
+
+def test_build_query_basic(retrieve_profile_api_instance):
+    columns = ["Id", "Name"]
+    table_name = "Account"
+    expected_query = "SELECT Id, Name FROM Account"
+    result_query = retrieve_profile_api_instance._build_query(columns, table_name)
+    assert result_query == expected_query
+
+
+def test_build_query_with_where(retrieve_profile_api_instance):
+    columns = ["Id", "Name"]
+    table_name = "Account"
+    where = {"Type": "Customer", "Status": ["Active", "Inactive"]}
+    expected_query = "SELECT Id, Name FROM Account WHERE Type = 'Customer' AND Status IN ('Active', 'Inactive')"
+    result_query = retrieve_profile_api_instance._build_query(
+        columns, table_name, where
+    )
+    assert result_query == expected_query
+
+
+def test_build_query_empty_columns(retrieve_profile_api_instance):
+    columns = []
+    table_name = "Account"
+    with pytest.raises(ValueError):
+        retrieve_profile_api_instance._build_query(columns, table_name)
+
+
+def test_build_query_empty_table_name(retrieve_profile_api_instance):
+    columns = ["Id", "Name"]
+    table_name = ""
+    with pytest.raises(ValueError):
+        retrieve_profile_api_instance._build_query(columns, table_name)
+
+
+def test_process_sObject_results(retrieve_profile_api_instance):
+    result_list = [
+        {"SobjectType": "Account"},
+        {"SobjectType": "Contact"},
+        {"SobjectType": "Opportunity"},
+    ]
+    expected_result = {"CustomObject": ["Account", "Contact", "Opportunity"]}
+    result = retrieve_profile_api_instance._process_sObject_results(result_list)
+    assert result == expected_result
+
+
+def test_process_sObject_results_missing_key(retrieve_profile_api_instance):
+    result_list = [{"ObjectType": "Account"}]
+    with pytest.raises(KeyError):
+        retrieve_profile_api_instance._process_sObject_results(result_list)
+
+
+def test_process_customTab_results(retrieve_profile_api_instance):
+    result_list = [
+        {"Name": "CustomTab1"},
+        {"Name": "CustomTab2"},
+        {"Name": "CustomTab3"},
+    ]
+    expected_result = {"CustomTab": ["CustomTab1", "CustomTab2", "CustomTab3"]}
+    result = retrieve_profile_api_instance._process_customTab_results(result_list)
+    assert result == expected_result
+
+
+def test_process_customTab_results_missing_key(retrieve_profile_api_instance):
+    result_list = [{"TabName": "CustomTab1"}]
+    with pytest.raises(KeyError):
+        retrieve_profile_api_instance._process_customTab_results(result_list)
+
+
+def test_process_setupEntityAccess_results(retrieve_profile_api_instance):
+    result_list = [
+        {"SetupEntityType": "ApexClass", "SetupEntityId": "001abc"},
+        {"SetupEntityType": "ApexPage", "SetupEntityId": "002def"},
+        {"SetupEntityType": "CustomPermission", "SetupEntityId": "003ghi"},
+    ]
+
+    queries_result = {
+        "ApexClass": [
+            {"Id": "001abc", "Name": "TestApexClass", "NamespacePrefix": "apex"}
+        ],
+        "ApexPage": [{"Id": "002def", "Name": "TestApexPage"}],
+        "CustomPermission": [],
+    }
+    with patch.object(
+        RetrieveProfileApi, "_build_query", return_value="SELECT Id, Name FROM Table"
+    ) as mock_build_query, patch.object(
+        RunParallelQueries,
+        "_run_queries_in_parallel",
+        return_value=queries_result,
+    ) as mock_run_queries:
+
+        (
+            entities,
+            result,
+        ) = retrieve_profile_api_instance._process_setupEntityAccess_results(
+            result_list
+        )
+        mock_build_query.assert_any_call(
+            ["Name", "NamespacePrefix"], "ApexClass", {"Id": ["001abc"]}
+        )
+        mock_build_query.assert_any_call(
+            ["Name", "NamespacePrefix"], "ApexPage", {"Id": ["002def"]}
+        )
+        mock_build_query.assert_any_call(
+            ["DeveloperName", "NamespacePrefix"], "CustomPermission", {"Id": ["003ghi"]}
+        )
+        mock_run_queries.assert_called_with(
+            {
+                "ApexClass": "SELECT Id, Name FROM Table",
+                "ApexPage": "SELECT Id, Name FROM Table",
+                "CustomPermission": "SELECT Id, Name FROM Table",
+            },
+            retrieve_profile_api_instance._run_query,
+        )
+
+    expected_entities = {
+        "ApexClass": ["apex__TestApexClass"],
+        "ApexPage": ["TestApexPage"],
+        "CustomPermission": [],
+    }
+    assert entities == expected_entities
+    assert result == queries_result
+
+
+def test_process_all_results(retrieve_profile_api_instance):
+    result_dict = {
+        "setupEntityAccess": "some_result",
+        "sObject": "some_result",
+        "customTab": "some_result",
+        "profileFlow": "some_result",
+    }
+    with patch.object(
+        RetrieveProfileApi,
+        "_process_setupEntityAccess_results",
+        return_value=(
+            {
+                "ApexClass": ["TestApexClass"],
+                "ApexPage": ["TestApexPage"],
+                "FlowDefinition": ["TestFlow"],
+            },
+            {"FlowDefinition": ["some_result"]},
+        ),
+    ), patch.object(
+        RetrieveProfileApi,
+        "_process_sObject_results",
+        return_value={"CustomObject": ["TestObject"]},
+    ), patch.object(
+        RetrieveProfileApi,
+        "_process_customTab_results",
+        return_value={"CustomTab": ["TestTab"]},
+    ), patch.object(
+        RetrieveProfileApi,
+        "_match_profiles_and_flows",
+        return_value={"Profile1": ["Flow1"]},
+    ):
+        entities, profile_flow = retrieve_profile_api_instance._process_all_results(
+            result_dict
+        )
+
+    print(entities)
+    assert entities["ApexClass"] == ["TestApexClass"]
+    assert entities["ApexPage"] == ["TestApexPage"]
+    assert entities["FlowDefinition"] == ["TestFlow"]
+    assert entities["CustomObject"] == ["TestObject"]
+    assert entities["CustomTab"] == ["TestTab"]
+    assert profile_flow == {"Profile1": ["Flow1"]}
+
+
+def test_queries_retrieve_permissions(retrieve_profile_api_instance):
+    profiles = ["Profile1", "Profile2"]
+
+    with patch.object(RetrieveProfileApi, "_build_query") as mock_build_query:
+        retrieve_profile_api_instance._queries_retrieve_permissions(profiles)
+
+    mock_build_query.assert_any_call(
+        ["SetupEntityId", "SetupEntityType"],
+        "SetupEntityAccess",
+        {"Parent.Profile.Name": profiles},
+    )
+    mock_build_query.assert_any_call(
+        ["SObjectType"], "ObjectPermissions", {"Parent.Profile.Name": profiles}
+    )
+    mock_build_query.assert_any_call(
+        ["Name"], "PermissionSetTabSetting", {"Parent.Profile.Name": profiles}
+    )
+
+
+def test_retrieve_permissionable_entities(retrieve_profile_api_instance):
+    profiles = ["Profile1", "Profile2"]
+    expected_queries = {"query_name": "query"}
+    expected_result = (
+        {
+            "ApexClass": ["TestApexClass"],
+            "ApexPage": ["TestApexPage"],
+            "CustomObject": ["TestObject"],
+            "CustomTab": ["TestTab"],
+        },
+        {"Profile1": ["Flow1"]},
+    )
+
+    with patch.object(
+        RunParallelQueries, "_run_queries_in_parallel"
+    ) as mock_run_queries, patch.object(
+        RetrieveProfileApi,
+        "_queries_retrieve_permissions",
+        return_value=expected_queries,
+    ), patch.object(
+        RetrieveProfileApi, "_process_all_results", return_value=expected_result
+    ):
+
+        result = retrieve_profile_api_instance._retrieve_permissionable_entities(
+            profiles
+        )
+        mock_run_queries.assert_called_with(
+            expected_queries, retrieve_profile_api_instance._run_query
+        )
+
+        assert result == expected_result
+
+
+def test_match_profiles_and_flows(retrieve_profile_api_instance):
+    result_profiles = [
+        {"SetupEntityId": "001abc", "Parent": {"Profile": {"Name": "Profile1"}}},
+        {"SetupEntityId": "001abc", "Parent": {"Profile": {"Name": "Profile2"}}},
+        {"SetupEntityId": "002def", "Parent": {"Profile": {"Name": "Profile1"}}},
+    ]
+
+    result_flows = [
+        {"ApiName": "Flow1", "attributes": {"url": "instance_url/001abc"}},
+        {"ApiName": "Flow2", "attributes": {"url": "instance_url/002def"}},
+    ]
+
+    profile_flow = retrieve_profile_api_instance._match_profiles_and_flows(
+        result_profiles, result_flows
+    )
+
+    assert "Flow1" in profile_flow["Profile1"]
+    assert "Flow2" in profile_flow["Profile1"]
+    assert "Flow1" in profile_flow["Profile2"]
+    assert "Flow2" not in profile_flow["Profile2"]
+
+
+if __name__ == "__main__":
+    pytest.main()
diff --git a/cumulusci/tasks/salesforce/tests/test_retrieve_profile.py b/cumulusci/tasks/salesforce/tests/test_retrieve_profile.py
new file mode 100644
index 000000000..7f34e3958
--- /dev/null
+++ b/cumulusci/tasks/salesforce/tests/test_retrieve_profile.py
@@ -0,0 +1,253 @@
+import logging
+import os
+import xml.etree.ElementTree as ET
+import zipfile
+from tempfile import NamedTemporaryFile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cumulusci.core.config import TaskConfig
+from cumulusci.salesforce_api.metadata import ApiRetrieveUnpackaged
+from cumulusci.salesforce_api.retrieve_profile_api import RetrieveProfileApi
+from cumulusci.tasks.salesforce.retrieve_profile import RetrieveProfile
+
+
+@pytest.fixture
+def retrieve_profile_task(tmpdir):
+    project_config = MagicMock()
+    task_config = TaskConfig(
+        {
+            "options": {
+                "profiles": ["Profile1", "Profile2"],
+                "path": tmpdir,
+                "strict_mode": False,
+            }
+        }
+    )
+    org_config = MagicMock()
+    task = RetrieveProfile(project_config, task_config, org_config)
+    task.logger = logging.getLogger(__name__)
+    task.logger.setLevel(logging.DEBUG)
+    return task
+
+
+def test_check_existing_profiles_with_missing_profiles_and_strict_mode_enabled(tmpdir):
+    project_config = MagicMock()
+    task_config = TaskConfig(
+        {
+            "options": {
+                "profiles": ["Profile1", "Profile2"],
+                "path": tmpdir,
+                "strict_mode": True,
+            }
+        }
+    )
+    org_config = MagicMock()
+    retrieve_profile_task = RetrieveProfile(project_config, task_config, org_config)
+    with patch.object(
+        RetrieveProfileApi, "_retrieve_existing_profiles", return_value=["Profile1"]
+    ):
+        with pytest.raises(
+            RuntimeError, match="Operation failed due to missing profiles"
+        ):
+            retrieve_profile_task._check_existing_profiles(RetrieveProfileApi)
+
+
+def test_check_existing_profiles_with_no_existing_profiles(tmpdir):
+    project_config = MagicMock()
+    task_config = TaskConfig(
+        {
+            "options": {
+                "profiles": ["Profile1", "Profile2"],
+                "path": tmpdir,
+                "strict_mode": False,
+            }
+        }
+    )
+    org_config = MagicMock()
+    retrieve_profile_task = RetrieveProfile(project_config, task_config, org_config)
+    with patch.object(
+        RetrieveProfileApi, "_retrieve_existing_profiles", return_value=[]
+    ):
+        with pytest.raises(
+            RuntimeError, match="None of the profiles given were found."
+        ):
+            retrieve_profile_task._check_existing_profiles(RetrieveProfileApi)
+
+
+def test_init_options(retrieve_profile_task):
+    retrieve_profile_task._init_options(retrieve_profile_task.task_config.config)
+    assert retrieve_profile_task.profiles == ["Profile1", "Profile2"]
+    assert retrieve_profile_task.strictMode is False
+
+
+def test_init_options_raises_error_with_no_profiles():
+    project_config = MagicMock()
+    task_config = TaskConfig({"options": {"profiles": None}})
+    org_config = MagicMock()
+
+    with pytest.raises(ValueError) as exc_info:
+        RetrieveProfile(project_config, task_config, org_config)
+
+    assert str(exc_info.value) == "At least one profile must be specified."
+
+
+def test_init_options_raises_error_with_invalid_path_directory():
+    project_config = MagicMock()
+    task_config = TaskConfig(
+        {"options": {"profiles": ["Profile1"], "path": "/nonexistent/directory"}}
+    )
+    org_config = MagicMock()
+
+    with pytest.raises(FileNotFoundError) as exc_info:
+        RetrieveProfile(project_config, task_config, org_config)
+
+    expected_message = "The extract directory '/nonexistent/directory' does not exist."
+    assert str(exc_info.value) == expected_message
+
+
+def test_init_options_raises_error_with_non_directory_path(tmp_path):
+    tmpfile = tmp_path / "file.txt"
+    tmpfile.write_text("Something")
+    project_config = MagicMock()
+    task_config = TaskConfig({"options": {"profiles": ["Profile1"], "path": tmpfile}})
+    org_config = MagicMock()
+
+    with pytest.raises(NotADirectoryError) as exc_info:
+        RetrieveProfile(project_config, task_config, org_config)
+
+    expected_message = f"'{tmpfile}' is not a directory."
+    assert str(exc_info.value) == expected_message
+
+
+def create_temp_zip_file():
+    temp_zipfile = NamedTemporaryFile(delete=True)
+
+    with zipfile.ZipFile(temp_zipfile, "w", zipfile.ZIP_DEFLATED) as zipf:
+        zipf.writestr("profiles/Profile1.profile", "")
+
+    return zipfile.ZipFile(temp_zipfile, "r")
+
+
+def test_save_profile_file(retrieve_profile_task, tmpdir):
+    extract_dir = str(tmpdir)
+    filename = "TestProfile.profile"
+    content = "Profile content"
+    expected_file_path = os.path.join(extract_dir, filename)
+    retrieve_profile_task.save_profile_file(extract_dir, filename, content)
+
+    assert os.path.exists(expected_file_path)
+    with open(expected_file_path, "r", encoding="utf-8") as profile_file:
+        saved_content = profile_file.read()
+    assert saved_content == content
+
+
+def test_add_flow_accesses(retrieve_profile_task):
+    profile_content = "<Profile>\n" "    <some_tag>Hello</some_tag>\n" "</Profile>"
+    flows = ["Flow1", "Flow2"]
+    expected_content = (
+        "<Profile>\n"
+        "    <some_tag>Hello</some_tag>\n"
+        "    <flowAccesses>\n"
+        "        <enabled>true</enabled>\n"
+        "        <flow>Flow1</flow>\n"
+        "    </flowAccesses>\n"
+        "    <flowAccesses>\n"
+        "        <enabled>true</enabled>\n"
+        "        <flow>Flow2</flow>\n"
+        "    </flowAccesses>\n"
+        "</Profile>"
+    )
+    modified_content = retrieve_profile_task.add_flow_accesses(profile_content, flows)
+    assert modified_content == expected_content
+
+    # Content without the </Profile> tag
+    profile_content = "<Profile>\n" "    <some_tag>Hello</some_tag>\n"
+    modified_content = retrieve_profile_task.add_flow_accesses(profile_content, flows)
+    assert modified_content == profile_content
+
+
+def test_run_task(retrieve_profile_task, tmpdir, caplog):
+    retrieve_profile_task.extract_dir = tmpdir
+    temp_zipfile = create_temp_zip_file()
+
+    with patch.object(
+        RetrieveProfileApi,
+        "_retrieve_permissionable_entities",
+        return_value=({"ApexClass": ["TestApexClass"]}, {"Profile1": ["Flow1"]}),
+    ), patch.object(
+        RetrieveProfileApi, "_init_task", return_value="something"
+    ), patch.object(
+        ApiRetrieveUnpackaged, "__call__", return_value=temp_zipfile
+    ), patch.object(
+        RetrieveProfileApi, "_retrieve_existing_profiles", return_value=["Profile1"]
+    ):
+        retrieve_profile_task._run_task()
+
+    assert os.path.exists(tmpdir)
+    profile1_path = os.path.join(tmpdir, "profiles/Profile1.profile")
+    assert os.path.exists(profile1_path)
+
+    log_messages = [record.message for record in caplog.records]
+    assert f"Profiles Profile1 unzipped into folder '{tmpdir}'" in log_messages
+    assert (
+        f"Profiles Profile1, Profile2 unzipped into folder '{tmpdir}'"
+        not in log_messages
+    )
+    assert (
+        "The following profiles were not found or could not be retrieved: 'Profile2'\n"
+        in log_messages
+    )
+
+
+def test_get_api(retrieve_profile_task):
+    retrieve_profile_task.package_xml = ""
+    api = retrieve_profile_task._get_api()
+    assert isinstance(api, ApiRetrieveUnpackaged)
+
+
+def remove_whitespace(xml_str):
+    return "".join(line.strip() for line in xml_str.splitlines())
+
+
+def compare_xml_strings(xml_str1, xml_str2):
+    xml_str1 = remove_whitespace(xml_str1)
+    xml_str2 = remove_whitespace(xml_str2)
+
+    tree1 = ET.ElementTree(ET.fromstring(xml_str1))
+    tree2 = ET.ElementTree(ET.fromstring(xml_str2))
+
+    parsed_xml_str1 = ET.tostring(tree1.getroot(), encoding="unicode")
+    parsed_xml_str2 = ET.tostring(tree2.getroot(), encoding="unicode")
+
+    return parsed_xml_str1 == parsed_xml_str2
+
+
+def test_create_package_xml(retrieve_profile_task):
+    input_dict = {
+        "ApexClass": ["Class1", "Class2"],
+        "Profile": ["Profile1", "Profile2"],
+    }
+    package_xml = retrieve_profile_task._create_package_xml(input_dict, "58.0")
+
+    expected_package_xml = """<?xml version="1.0" encoding="UTF-8"?>
+    <Package xmlns="http://soap.sforce.com/2006/04/metadata">
+        <types>
+            <members>Class1</members>
+            <members>Class2</members>
+            <name>ApexClass</name>
+        </types>
+        <types>
+            <members>Profile1</members>
+            <members>Profile2</members>
+            <name>Profile</name>
+        </types>
+        <version>58.0</version>
+    </Package>"""
+
+    assert compare_xml_strings(package_xml, expected_package_xml)
+
+
+if __name__ == "__main__":
+    pytest.main()
diff --git a/cumulusci/utils/parallel/queries_in_parallel/tests/test_run_queries_in_parallel.py b/cumulusci/utils/parallel/queries_in_parallel/tests/test_run_queries_in_parallel.py
new file mode 100644
index 000000000..725cf8364
--- /dev/null
+++ b/cumulusci/utils/parallel/queries_in_parallel/tests/test_run_queries_in_parallel.py
@@ -0,0 +1,50 @@
+from concurrent.futures import Future, ThreadPoolExecutor
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cumulusci.utils.parallel.queries_in_parallel.run_queries_in_parallel import (
+    RunParallelQueries,
+)
+
+
+def run_query(query):
+    return {"test": query}
+
+
+def test_run_queries_in_parallel():
+    queries = {
+        "query1": "SELECT * FROM Table1",
+        "query2": "SELECT * FROM Table2",
+    }
+
+    expected_results = {
+        "query1": [{"field1": "value1"}, {"field1": "value2"}],
+        "query2": [{"field2": "value3"}, {"field2": "value4"}],
+    }
+
+    mock_future1 = MagicMock()
+    mock_future1.result.return_value = {"records": expected_results["query1"]}
+    mock_future2 = MagicMock()
+    mock_future2.result.return_value = {"records": expected_results["query2"]}
+
+    with patch.object(ThreadPoolExecutor, "submit") as mock_submit:
+        mock_submit.side_effect = [mock_future1, mock_future2]
+
+        results_dict = RunParallelQueries._run_queries_in_parallel(queries, run_query)
+
+    assert results_dict == expected_results
+
+
+def test_run_queries_in_parallel_with_exception():
+    queries = {
+        "Query1": "SELECT Id FROM Table1",
+    }
+
+    with patch.object(Future, "result", side_effect=Exception("Test exception")):
+        with pytest.raises(Exception) as excinfo:
+            RunParallelQueries._run_queries_in_parallel(queries, run_query=run_query)
+        assert (
+            "Error executing query 'Query1': <class 'Exception'>: Test exception"
+            == str(excinfo.value)
+        )