Repository: SFDO-Tooling/CumulusCI
Pull request #3031 (2022-01-08 00:11:23)

update_data task for CumulusCI
# Changes

CumulusCI can now do bulk updates for Salesforce sObjects.

# Notes

This is ready for review. Tests will not pass until https://github.com/SFDO-Tooling/Snowfakery/pull/586 is merged and a pre-release is made, but the code itself is ready for review.



Patch:
diff --git a/cumulusci/core/tasks.py b/cumulusci/core/tasks.py
index fc64be6ca..2965b0ab8 100644
--- a/cumulusci/core/tasks.py
+++ b/cumulusci/core/tasks.py
@@ -272,3 +272,59 @@ class BaseSalesforceTask(BaseTask):
     def _update_credentials(self):
         with self.org_config.save_if_changed():
             self.org_config.refresh_oauth_token(self.project_config.keychain)
+
+    def _validate_and_inject_namespace_prefixes(
+        self,
+        should_inject_namespaces: bool,
+        sobjects_to_validate: list,
+        operation_to_validate: str,
+    ):
+        """Perform namespace injection and ensure that we can successfully access all of the selected objects."""
+
+        global_describe = {
+            entry["name"]: entry
+            for entry in self.org_config.salesforce_client.describe()["sobjects"]
+        }
+
+        # Namespace injection
+        if should_inject_namespaces and self.project_config.project__package__namespace:
+
+            def inject(element: str):
+                return f"{self.project_config.project__package__namespace}__{element}"
+
+            sobjects = [
+                self.fixup_sobject_name(inject, sobject, global_describe)
+                for sobject in sobjects_to_validate
+            ]
+        else:
+            sobjects = sobjects_to_validate
+
+        # Validate CRUD
+        non_accessible_objects = [
+            s
+            for s in sobjects
+            if not (s in global_describe and global_describe[s][operation_to_validate])
+        ]
+        if non_accessible_objects:
+            raise TaskOptionsError(
+                f"The objects {', '.join(non_accessible_objects)} are not present or not {operation_to_validate}."
+            )
+        return sobjects
+
+    def fixup_sobject_name(self, injection_func, sobject, global_describe):
+        def _is_injectable(element: str) -> bool:
+            return element.count("__") == 1
+
+        if not _is_injectable(sobject):
+            return sobject
+
+        injected = injection_func(sobject)
+        if sobject in global_describe and injected in global_describe:
+            self.logger.warning(
+                f"Both {sobject} and {injected} are present in the target org. Using {sobject}."
+            )
+
+        if sobject not in global_describe and injected in global_describe:
+            return injected
+        else:
+            return sobject
diff --git a/cumulusci/cumulusci.yml b/cumulusci/cumulusci.yml
index f6431552d..44db3b130 100644
--- a/cumulusci/cumulusci.yml
+++ b/cumulusci/cumulusci.yml
@@ -187,6 +187,11 @@ tasks:
         description: Query existing data for a specific sObject and perform a Bulk API delete of all matching records.
         class_path: cumulusci.tasks.bulkdata.DeleteData
         group: Data Operations
+    update_data:
+        description: Update records of an sObject matching a where-clause.
+        class_path: cumulusci.tasks.bulkdata.update_data.UpdateData
+        group: Data Operations
+
     deploy:
         description: Deploys the src directory of the repository to the org
         class_path: cumulusci.tasks.salesforce.Deploy
diff --git a/cumulusci/tasks/bulkdata/delete.py b/cumulusci/tasks/bulkdata/delete.py
index e6505d07a..8500a208f 100644
--- a/cumulusci/tasks/bulkdata/delete.py
+++ b/cumulusci/tasks/bulkdata/delete.py
@@ -78,55 +78,14 @@ class DeleteData(BaseSalesforceApiTask):
         if self.options["hardDelete"] and self.options["api"] is DataApi.REST:
             raise TaskOptionsError("The hardDelete option requires Bulk API.")
 
-    @staticmethod
-    def _is_injectable(element: str) -> bool:
-        return element.count("__") == 1
-
     def _validate_and_inject_namespace(self):
         """Perform namespace injection and ensure that we can successfully delete all of the selected objects."""
 
-        global_describe = {
-            entry["name"]: entry
-            for entry in self.org_config.salesforce_client.describe()["sobjects"]
-        }
-
-        # Namespace injection
-        if (
-            self.options["inject_namespaces"]
-            and self.project_config.project__package__namespace
-        ):
-
-            def inject(element: str):
-                return f"{self.project_config.project__package__namespace}__{element}"
-
-            self.sobjects = []
-            for sobject in self.options["objects"]:
-                if self._is_injectable(sobject):
-                    injected = inject(sobject)
-                    if sobject in global_describe and injected in global_describe:
-                        self.logger.warning(
-                            f"Both {sobject} and {injected} are present in the target org. Using {sobject}."
-                        )
-
-                    if sobject not in global_describe and injected in global_describe:
-                        self.sobjects.append(injected)
-                    else:
-                        self.sobjects.append(sobject)
-                else:
-                    self.sobjects.append(sobject)
-        else:
-            self.sobjects = self.options["objects"]
-
-        # Validate CRUD
-        non_deletable_objects = [
-            s
-            for s in self.sobjects
-            if not (s in global_describe and global_describe[s]["deletable"])
-        ]
-        if non_deletable_objects:
-            raise BulkDataException(
-                f"The objects {', '.join(non_deletable_objects)} are not present or cannot be deleted."
-            )
+        self.sobjects = super()._validate_and_inject_namespace_prefixes(
+            should_inject_namespaces=self.options["inject_namespaces"],
+            sobjects_to_validate=self.options["objects"],
+            operation_to_validate="deletable",
+        )
 
     def _run_task(self):
         self._validate_and_inject_namespace()
@@ -180,7 +139,7 @@ class DeleteData(BaseSalesforceApiTask):
                 DataOperationStatus.ROW_FAILURE,
             ]:
                 raise BulkDataException(
-                    f"Unable to delete records for {obj}: {','.join(qs.job_result.job_errors)}"
+                    f"Unable to delete records for {obj}: {','.join(ds.job_result.job_errors)}"
                 )
 
             error_checker = RowErrorChecker(
diff --git a/cumulusci/tasks/bulkdata/update_data.py b/cumulusci/tasks/bulkdata/update_data.py
new file mode 100644
index 000000000..fa90d7117
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/update_data.py
@@ -0,0 +1,226 @@
+import csv
+from contextlib import contextmanager
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+from snowfakery import generate_data
+
+from cumulusci.core.exceptions import BulkDataException, TaskOptionsError
+from cumulusci.core.utils import (
+    process_bool_arg,
+    process_list_arg,
+    process_list_of_pairs_dict_arg,
+)
+from cumulusci.tasks.bulkdata.step import (
+    DataApi,
+    DataOperationStatus,
+    DataOperationType,
+    get_dml_operation,
+    get_query_operation,
+)
+from cumulusci.tasks.bulkdata.utils import RowErrorChecker
+from cumulusci.tasks.salesforce import BaseSalesforceApiTask
+
+
+class UpdateData(BaseSalesforceApiTask):
+    """Update records of an sObject matching a where-clause."""
+
+    task_options = {
+        "object": {
+            "description": "An SObject",
+            "required": True,
+        },
+        "where": {
+            "description": "A SOQL where-clause (without the keyword WHERE).",
+            "required": False,
+        },
+        "api": {
+            "description": "The desired Salesforce API to use, which may be 'rest', 'bulk', or "
+            "'smart' to auto-select based on record volume. The default is 'smart'.",
+            "required": False,
+        },
+        "fields": {
+            "description": "Fields to download as input to the Snowfakery recipe",
+            "required": False,
+        },
+        "recipe": {
+            "description": "Snowfakery recipe to be executed on each row",
+            "required": True,
+        },
+        "recipe_options": {
+            "required": False,
+            "description": """Pass values to override options in the format VAR1:foo,VAR2:bar
+
+            Example: --recipe_options weight:10,color:purple""",
+        },
+        "ignore_row_errors": {
+            "description": "If True, allow the operation to continue even if individual rows fail to delete."
+        },
+    }
+    row_warning_limit = 10
+
+    def _init_options(self, kwargs):
+        super()._init_options(kwargs)
+
+        self.sobject = self.options["object"]
+
+        def identifier(f):
+            return isinstance(f, str) and f.isidentifier()
+
+        if not identifier(self.sobject):
+            raise TaskOptionsError(
+                f"`object` option should be an sObject name, not {self.sobject}"
+            )
+
+        self.where = self.options.get("where")
+        self.fields = process_list_arg(self.options.get("fields", []))
+        bad_fields = [f for f in self.fields if not identifier(f)]
+        if bad_fields:
+            plural = "s" if len(bad_fields) > 1 else ""
+            verb = "are" if len(bad_fields) > 1 else "is"
+            article = "an " if len(bad_fields) == 1 else ""
+            raise TaskOptionsError(
+                f"Field{plural} `{'`,`'.join(bad_fields)}` {verb} not {article}identifier{plural}"
+            )
+
+        self.inject_namespaces = process_bool_arg(
+            self.task_options.get("inject_namespaces", True)
+        )
+        self.recipe = self.options.get("recipe")
+        self.recipe_options = process_list_of_pairs_dict_arg(
+            self.options.get("recipe_options", {})
+        )
+        self.ignore_row_errors = process_bool_arg(
+            self.options.get("ignore_row_errors", False)
+        )
+
+        try:
+            self.api = {
+                "bulk": DataApi.BULK,
+                "rest": DataApi.REST,
+                "smart": DataApi.SMART,
+            }[self.options.get("api", "smart").lower()]
+        except KeyError:
+            raise TaskOptionsError(
+                f"{self.options['api']} is not a valid value for API (valid: bulk, rest, smart)"
+            )
+
+    def _run_task(self):
+        obj = self._validate_and_inject_namespace_prefixes(
+            should_inject_namespaces=self.inject_namespaces,
+            sobjects_to_validate=[self.sobject],
+            operation_to_validate="updateable",
+        )[0]
+        fields = ["Id"] + self.fields
+        qs = self.query_objects(obj, fields)
+        if not qs:
+            self.logger.info(f"No records found, skipping update operation for {obj}")
+            return
+        with self.save_records(qs, fields) as csvfile, TemporaryDirectory() as outdir:
+            csv_out = self.generate_data(csvfile, Path(outdir))
+            with csv_out.open() as csv_out:
+                enriched_data = csv.DictReader(csv_out)
+                oid_index = enriched_data.fieldnames.index("Oid")
+                enriched_data.fieldnames[oid_index] = "Id"
+                self.load_data(qs.job_result.records_processed, enriched_data)
+
+        self.logger.info(f"Updated {self._object_description(obj)}")
+
+    def generate_data(self, csvfile: Path, outdir: Path) -> Path:
+        generate_data(
+            yaml_file=self.recipe,
+            user_options=self.recipe_options,
+            update_input_file=csvfile,
+            update_passthrough_fields=["Oid"],
+            plugin_options={
+                "org_config": self.org_config,
+                "project_config": self.project_config,
+            },
+            output_folder=outdir,
+            output_format="csv",
+        )
+        created_csv = tuple(outdir.glob("*.csv"))
+        assert len(created_csv) == 1, "CSV was not created by Snowfakery"
+        return created_csv[0]
+
+    @contextmanager
+    def save_records(self, qs, fields):  # -> Path
+        with TemporaryDirectory() as t:
+            csvfile = Path(t) / "input.csv"
+            with csvfile.open("w") as f:
+                writer = csv.writer(f)
+                snowfakery_fieldnames = ["Oid"] + fields[1:]
+                writer.writerow(snowfakery_fieldnames)
+                writer.writerows(qs.get_results())
+
+            yield csvfile
+
+    def query_objects(self, obj, fields):
+        query = f"SELECT {','.join(fields)} FROM {obj}"
+        if self.where:
+            query += f" WHERE {self.options['where']}"
+
+        qs = get_query_operation(
+            sobject=obj,
+            fields=fields,
+            api_options={},
+            context=self,
+            query=query,
+            api=DataApi.SMART,  # maybe this should always be bulk
+        )
+
+        self.logger.info(f"Querying for {obj} objects")
+        qs.query()
+        if qs.job_result.status is not DataOperationStatus.SUCCESS:
+            raise BulkDataException(
+                f"Unable to query records for {obj}: {','.join(qs.job_result.job_errors)}"
+            )
+        if not qs.job_result.records_processed:
+            return None
+        return qs
+
+    def load_data(self, row_count: int, records: csv.DictReader):
+        obj = self.sobject
+        self.logger.info(f"Loading data {self._object_description(obj)} ")
+        fieldnames = [
+            f for f in records.fieldnames if f != "id" and not f.startswith("_")
+        ]
+
+        ds = get_dml_operation(
+            sobject=obj,
+            operation=DataOperationType.UPDATE,
+            fields=fieldnames,
+            api_options={},
+            context=self,
+            api=self.api,
+            volume=row_count,
+        )
+
+        def cleanup(record):
+            return tuple(record[fieldname] for fieldname in fieldnames)
+
+        ds.start()
+        ds.load_records(cleanup(record) for record in records)
+        ds.end()
+
+        if ds.job_result.status not in [
+            DataOperationStatus.SUCCESS,
+            DataOperationStatus.ROW_FAILURE,
+        ]:
+            raise BulkDataException(
+                f"Unable to update records for {obj}: {','.join(ds.job_result.job_errors)}"
+            )
+
+        error_checker = RowErrorChecker(
+            self.logger, self.ignore_row_errors, self.row_warning_limit
+        )
+        for result in ds.get_results():
+            error_checker.check_for_row_error(result, result.id)
+        return ds
+
+    def _object_description(self, obj):
+        """Return a readable description of the object set to update."""
+        if self.where:
+            return f'{obj} objects matching "{self.options["where"]}"'
+        else:
+            return f"all {obj} objects"
diff --git a/datasets/update.recipe.yml b/datasets/update.recipe.yml
new file mode 100644
index 000000000..52e99dc3e
--- /dev/null
+++ b/datasets/update.recipe.yml
@@ -0,0 +1,10 @@
+# cci task run update_data --recipe datasets/update.recipe.yml
+#                          --object Account --fields Name --org qa
+- plugin: snowfakery.standard_plugins.Counters
+
+- object: Account
+  fields:
+      BillingStreet:
+          fake: Streetname
+      Description: ${{input.Name}} is our favorite customer
+      NumberOfEmployees: 10_000
diff --git a/docs/data.rst b/docs/data.rst
index 965512bb4..bc266a3aa 100644
--- a/docs/data.rst
+++ b/docs/data.rst
@@ -610,7 +610,55 @@ Examples
 
     cci task run delete_data -o objects Account -o hardDelete True
 
+``update_data``
+---------------
+To update records using CumulusCI, provide:
+
+* a command line or task configuration describing what to update
+* a recipe in a subset of Snowfakery syntax that says how to update it
+
+On the command line, you can run an update like
+this:
+
+``$ cci task run update_data --recipe datasets/update.recipe.yml --object Account``
+
+This command downloads every Account in the org and applies the fields from the specified
+update recipe file.
+
+You can filter the rows that you're updating like this:
+
+``$ cci task run update_data --recipe datasets/update.recipe.yml --object Account --where "name like 'AAA%'" ``
+
+The recipe for an update can be as simple as this:
+
+.. code-block::
+    - object: Account
+      fields:
+        NumberOfEmployees: 10000
+
+You can use all of the power of ``snowfakery`` to add fake data:
+
+.. code-block::
+    - object: Account
+      fields:
+        NumberOfEmployees: 10_000
+        BillingStreet:
+          fake: Streetname
+
+Using Snowfakery formulas, you can also refer to specific input fields
+like this:
+
+.. code-block::
+    - object: Account
+      fields:
+        Description: ${{input.Name}} is our favorite customer in ${{input.BillingCity}}
+
+To tell CumulusCI to extract those fields and make them
+use the ``fields`` option:
+
+``$ cci task run update_data --recipe datasets/update.recipe.yml --object Account --Fields Name,BillingCity ``
 
+You can learn more about Snowfakery syntax in the next section.
 
 Generate Fake Data
 ==================
diff --git a/requirements/prod.in b/requirements/prod.in
index f18c5aa69..b7e629ed1 100644
--- a/requirements/prod.in
+++ b/requirements/prod.in
@@ -27,6 +27,6 @@ salesforce-bulk
 sarge
 selenium<4
 simple-salesforce==1.11.4
-snowfakery
+snowfakery==2.5.1.dev0
 SQLAlchemy
 xmltodict
diff --git a/requirements/prod.txt b/requirements/prod.txt
index 67307ef32..f07c35cbe 100644
--- a/requirements/prod.txt
+++ b/requirements/prod.txt
@@ -1,5 +1,5 @@
 #
-# This file is autogenerated by pip-compile with python 3.8
+# This file is autogenerated by pip-compile with python 3.9
 # To update, run:
 #
 #    pip-compile requirements/prod.in
@@ -151,7 +151,7 @@ six==1.16.0
     #   python-dateutil
     #   salesforce-bulk
     #   snowfakery
-snowfakery==2.5.0
+snowfakery==2.5.1.dev0
     # via -r requirements/prod.in
 sqlalchemy==1.4.31
     # via


Test patch:
diff --git a/cumulusci/tasks/bulkdata/tests/test_delete.py b/cumulusci/tasks/bulkdata/tests/test_delete.py
index 5b0cb52d2..4c799e89e 100644
--- a/cumulusci/tasks/bulkdata/tests/test_delete.py
+++ b/cumulusci/tasks/bulkdata/tests/test_delete.py
@@ -328,7 +328,7 @@ class TestDeleteData(unittest.TestCase):
         )
         # ApexTestQueueItem is not deletable
 
-        with pytest.raises(BulkDataException):
+        with pytest.raises(TaskOptionsError):
             task._validate_and_inject_namespace()
 
     def test_validate_and_inject_namespace__packaged(self):
diff --git a/cumulusci/tasks/bulkdata/tests/test_updates.py b/cumulusci/tasks/bulkdata/tests/test_updates.py
new file mode 100644
index 000000000..342265a60
--- /dev/null
+++ b/cumulusci/tasks/bulkdata/tests/test_updates.py
@@ -0,0 +1,404 @@
+import typing as T
+from functools import wraps
+from unittest import mock
+from unittest.mock import call
+
+import pytest
+
+from cumulusci.core import exceptions as exc
+from cumulusci.tasks.bulkdata.step import (
+    DataApi,
+    DataOperationJobResult,
+    DataOperationResult,
+    DataOperationStatus,
+    DataOperationType,
+)
+from cumulusci.tasks.bulkdata.update_data import UpdateData
+
+
+def _hashify_operation(kwargs):
+    return frozenset(
+        {
+            **kwargs,
+            "context": None,
+            "fields": frozenset(kwargs["fields"]),
+            "api_options": frozenset(kwargs.get("api_options", {}).items()),
+        }.items()
+    )
+
+
+def noop(*args, **kwargs):
+    pass
+
+
+class MockBulkAPIResponses:
+    def activate(self, func):
+        @wraps(func)
+        def wrapper(*args, **kwds):
+            self.mock_bulk_API_responses_context = MockBulkAPIResponsesContext()
+            with mock.patch(
+                "cumulusci.tasks.bulkdata.update_data.get_query_operation",
+                self.mock_bulk_API_responses_context.get_query_operation,
+            ), mock.patch(
+                "cumulusci.tasks.bulkdata.update_data.get_dml_operation",
+                self.mock_bulk_API_responses_context.get_dml_operation,
+            ):
+                try:
+                    ret = func(*args, **kwds)
+                finally:
+                    self.mock_bulk_API_responses_context = None
+            return ret
+
+        return wrapper
+
+    @property
+    def add_query_operation(self):
+        return self.mock_bulk_API_responses_context.add_query_operation
+
+    @property
+    def add_dml_operation(self):
+        return self.mock_bulk_API_responses_context.add_dml_operation
+
+
+class FakeJobOperationResults(T.NamedTuple):
+    results: list
+    job_result_status: DataOperationStatus
+    loader_callback: T.Callable = noop
+
+
+class MockBulkAPIResponsesContext:
+    def __init__(self):
+        self.query_operations = {}
+        self.dml_operations = {}
+
+    def add_query_operation(
+        self, *, results, job_result_status=DataOperationStatus.SUCCESS, **expected
+    ):
+        self.query_operations[_hashify_operation(expected)] = FakeJobOperationResults(
+            results,
+            job_result_status,
+        )
+
+    def add_dml_operation(
+        self,
+        *,
+        results,
+        loader_callback=noop,
+        job_result_status=DataOperationStatus.SUCCESS,
+        **expected,
+    ):
+        val = FakeJobOperationResults(results, job_result_status, loader_callback)
+        self.dml_operations[_hashify_operation(expected)] = val
+
+    def get_query_operation(self, **given):
+        try:
+            results = self.query_operations[_hashify_operation(given)]
+        except KeyError:
+            raise KeyError(f"Cannot find response matching {given}")
+        return FakeOperationResult(results)
+
+    def get_dml_operation(self, **given):
+        try:
+            results = self.dml_operations[_hashify_operation(given)]
+        except KeyError:
+            raise KeyError(f"Cannot find response matching {given}")
+        return FakeDMLOperationResult(results, fields=given["fields"])
+
+
+class FakeOperationResult:
+    def __init__(self, results: FakeJobOperationResults):
+        self.results = results
+
+    def yield_per(self, number):  # TODO
+        return self.results.results
+
+    def count(self):
+        return len(self.results.results)
+
+    def query(self):
+        pass
+
+    @property
+    def job_result(self):
+        return DataOperationJobResult(
+            self.results.job_result_status, [], len(self.results.results), 0
+        )
+
+    def get_results(self):
+        return self.results.results
+
+
+class FakeDMLOperationResult(FakeOperationResult):
+    def __init__(self, results: FakeJobOperationResults, fields=None):
+        self.results = results
+        self.fields = fields
+
+    def start(self):
+        pass
+
+    def load_records(self, records, *args, **kwargs):
+        for record in records:
+            self.results.loader_callback(dict(zip(self.fields, record)))
+
+    def end(self):
+        pass
+
+
+def _fake_val_and_inject_ns(
+    should_inject_namespaces: bool,
+    sobjects_to_validate: list,
+    operation_to_validate: str,
+):
+    return sobjects_to_validate
+
+
+bulkapi_responses = MockBulkAPIResponses()
+
+
+class TestUpdates:
+    def test_options__bad_sobject(self, create_task):
+        with pytest.raises(exc.TaskOptionsError) as e:
+            task = create_task(UpdateData, {"object": "a b c"})
+            task()
+        assert "a b c" in str(e.value), e.value
+
+    def test_options__bad_field(self, create_task):
+        with pytest.raises(exc.TaskOptionsError) as e:
+            task = create_task(UpdateData, {"object": "Account", "fields": ["a b c"]})
+            task()
+        assert "a b c" in str(e.value), e.value
+
+    def test_options__bad_fields(self, create_task):
+        with pytest.raises(exc.TaskOptionsError) as e:
+            task = create_task(
+                UpdateData, {"object": "Account", "fields": ["a b c", "1 2 3"]}
+            )
+            task()
+        assert "a b c" in str(e.value), e.value
+
+    def test_options__bad_api(self, create_task):
+        with pytest.raises(exc.TaskOptionsError) as e:
+            task = create_task(UpdateData, {"object": "Account", "api": "XML-RPC"})
+            task()
+        assert "XML-RPC" in str(e.value), e.value
+
+    @bulkapi_responses.activate
+    def test_simple_task(self, create_task):
+
+        bulkapi_responses.add_query_operation(
+            sobject="Account",
+            fields=["Id", "Name"],
+            query="SELECT Id,Name FROM Account",
+            api=DataApi.SMART,
+            results=[
+                ["OID000BLAH", "Mark Benihoff"],
+                ["OID000BLAH2", "Parkour Hairish"],
+            ],
+        )
+        loader = mock.Mock()
+        bulkapi_responses.add_dml_operation(
+            sobject="Account",
+            operation=DataOperationType.UPDATE,
+            fields=["BillingStreet", "Description", "NumberOfEmployees", "Id"],
+            api_options={},
+            api=DataApi.SMART,
+            volume=2,
+            results=[
+                DataOperationResult("OID000BLAH", success=True, error=""),
+                DataOperationResult("OID000BLAH2", success=True, error=""),
+            ],
+            loader_callback=loader,
+        )
+
+        task = create_task(
+            UpdateData,
+            {
+                "object": "Account",
+                "recipe": "datasets/update.recipe.yml",
+                "fields": ["Name"],
+            },
+        )
+
+        task._validate_and_inject_namespace_prefixes = _fake_val_and_inject_ns
+        task()
+        assert loader.mock_calls == [
+            call(
+                {
+                    "BillingStreet": mock.ANY,
+                    "Description": "Mark Benihoff is our favorite customer",
+                    "NumberOfEmployees": "10000",
+                    "Id": "OID000BLAH",
+                }
+            ),
+            call(
+                {
+                    "BillingStreet": mock.ANY,
+                    "Description": "Parkour Hairish is our favorite customer",
+                    "NumberOfEmployees": "10000",
+                    "Id": "OID000BLAH2",
+                }
+            ),
+        ], loader.mock_calls
+
+    @bulkapi_responses.activate
+    def test_with_no_query_results(self, create_task):
+        bulkapi_responses.add_query_operation(
+            sobject="Account",
+            fields=["Id", "Name"],
+            query="SELECT Id,Name FROM Account",
+            api=DataApi.SMART,
+            results=[],
+        )
+
+        task = create_task(
+            UpdateData,
+            {
+                "object": "Account",
+                "recipe": "datasets/update.recipe.yml",
+                "fields": ["Name"],
+            },
+        )
+
+        task._validate_and_inject_namespace_prefixes = _fake_val_and_inject_ns
+        task.logger = mock.Mock()
+        task()
+        last_message = task.logger.mock_calls[-1].args[0]
+        assert last_message.startswith("No records found"), last_message
+
+    @bulkapi_responses.activate
+    def test_with_where_clause(self, create_task):
+        bulkapi_responses.add_query_operation(
+            sobject="Account",
+            fields=["Id", "Name"],
+            query="SELECT Id,Name FROM Account WHERE Name Like '%ark%'",
+            api=DataApi.SMART,
+            results=[
+                ["OID000BLAH", "Mark Benihoff"],
+                ["OID000BLAH2", "Parkour Hairish"],
+            ],
+        )
+        bulkapi_responses.add_dml_operation(
+            sobject="Account",
+            operation=DataOperationType.UPDATE,
+            fields=["BillingStreet", "Description", "NumberOfEmployees", "Id"],
+            api_options={},
+            api=DataApi.SMART,
+            volume=2,
+            results=[
+                DataOperationResult("OID000BLAH", success=True, error=""),
+                DataOperationResult("OID000BLAH2", success=True, error=""),
+            ],
+        )
+        task = create_task(
+            UpdateData,
+            {
+                "object": "Account",
+                "recipe": "datasets/update.recipe.yml",
+                "fields": ["Name"],
+                "where": "Name Like '%ark%'",
+            },
+        )
+        task._validate_and_inject_namespace_prefixes = _fake_val_and_inject_ns
+        task.logger = mock.Mock()
+        task()
+        last_message = task.logger.mock_calls[-1].args[0]
+        assert last_message.startswith(
+            "Updated Account objects matching \"Name Like '%ark%'\""
+        ), last_message
+
+    @bulkapi_responses.activate
+    def test_query_failed(self, create_task):
+        bulkapi_responses.add_query_operation(
+            sobject="Account",
+            fields=["Id", "Name"],
+            query="SELECT Id,Name FROM Account",
+            api=DataApi.SMART,
+            results=[],
+            job_result_status=DataOperationStatus.JOB_FAILURE,
+        )
+        loader = mock.Mock()
+        bulkapi_responses.add_dml_operation(
+            sobject="Account",
+            operation=DataOperationType.UPDATE,
+            fields=["BillingStreet", "Description", "NumberOfEmployees", "Id"],
+            api_options={},
+            api=DataApi.SMART,
+            volume=2,
+            results=[
+                DataOperationResult("OID000BLAH", success=True, error=""),
+                DataOperationResult("OID000BLAH2", success=True, error=""),
+            ],
+            loader_callback=loader,
+        )
+
+        task = create_task(
+            UpdateData,
+            {
+                "object": "Account",
+                "recipe": "datasets/update.recipe.yml",
+                "fields": ["Name"],
+            },
+        )
+
+        task._validate_and_inject_namespace_prefixes = _fake_val_and_inject_ns
+        with pytest.raises(exc.BulkDataException) as e:
+            task()
+
+        assert "Unable to query records" in str(e.value)
+        assert "Account" in str(e.value)
+
+    @bulkapi_responses.activate
+    def test_update_query_failed(self, create_task):
+        bulkapi_responses.add_query_operation(
+            sobject="Account",
+            fields=["Id", "Name"],
+            query="SELECT Id,Name FROM Account",
+            api=DataApi.SMART,
+            results=[
+                ["OID000BLAH", "Mark Benihoff"],
+                ["OID000BLAH2", "Parkour Hairish"],
+            ],
+        )
+        loader = mock.Mock()
+        bulkapi_responses.add_dml_operation(
+            sobject="Account",
+            operation=DataOperationType.UPDATE,
+            fields=["BillingStreet", "Description", "NumberOfEmployees", "Id"],
+            api_options={},
+            api=DataApi.SMART,
+            volume=2,
+            results=[],
+            loader_callback=loader,
+            job_result_status=DataOperationStatus.JOB_FAILURE,
+        )
+
+        task = create_task(
+            UpdateData,
+            {
+                "object": "Account",
+                "recipe": "datasets/update.recipe.yml",
+                "fields": ["Name"],
+            },
+        )
+
+        task._validate_and_inject_namespace_prefixes = _fake_val_and_inject_ns
+        with pytest.raises(exc.BulkDataException) as e:
+            task()
+        assert "Unable to update records" in str(e.value)
+
+
+class TestUpdatesIntegrationTests:
+    @pytest.mark.needs_org()
+    def test_updates_task(self, create_task, cumulusci_test_repo_root):
+        task = create_task(
+            UpdateData,
+            {
+                "object": "Account",
+                "recipe": "datasets/update.recipe.yml",
+                "fields": ["Name"],
+            },
+        )
+        task.logger = mock.Mock()
+        task()
+        last_message = task.logger.mock_calls[-1].args[0]
+        assert "Updated all Account" in str(last_message), str(last_message)