Repository: SFDO-Tooling/CumulusCI
Pull request #3119 (2022-03-18 16:11:46)

Refactor two bulkdata utils as functions
In upcoming work I'm doing, I'm trying to reduce our dependence on LoadData methods and use more helper functions 

My functions require certain helper features to be available as functions, not methods.

The methods remain in place for backwards compatibility, but I could also rewrite code to use them instead of the Mixin methods if that's the consensus.

Note that a reference to session.flush() was removed but I believe it was extraneous to start with.


Patch:
diff --git a/cumulusci/tasks/bulkdata/extract.py b/cumulusci/tasks/bulkdata/extract.py
index 5f02d1fe1..7ffd594f7 100644
--- a/cumulusci/tasks/bulkdata/extract.py
+++ b/cumulusci/tasks/bulkdata/extract.py
@@ -17,7 +17,13 @@ from cumulusci.tasks.bulkdata.step import (
     DataOperationType,
     get_query_operation,
 )
-from cumulusci.tasks.bulkdata.utils import SqlAlchemyMixin, consume, create_table
+from cumulusci.tasks.bulkdata.utils import (
+    SqlAlchemyMixin,
+    consume,
+    create_table,
+    sql_bulk_insert_from_records,
+    sql_bulk_insert_from_records_incremental,
+)
 from cumulusci.tasks.salesforce import BaseSalesforceApiTask
 from cumulusci.utils import log_progress
 
@@ -209,9 +215,9 @@ class ExtractData(SqlAlchemyMixin, BaseSalesforceApiTask):
             record_iterator = (strip_name_field(record) for record in record_iterator)
 
         if mapping.get_oid_as_pk():
-            self._sql_bulk_insert_from_records(
+            sql_bulk_insert_from_records(
                 connection=conn,
-                table=mapping.table,
+                table=self.metadata.tables[mapping.table],
                 columns=columns,
                 record_iterable=record_iterator,
             )
@@ -222,15 +228,15 @@ class ExtractData(SqlAlchemyMixin, BaseSalesforceApiTask):
             f_values = (row[1:] for row in values)
             f_ids = (row[:1] for row in ids)
 
-            values_chunks = self._sql_bulk_insert_from_records_incremental(
+            values_chunks = sql_bulk_insert_from_records_incremental(
                 connection=conn,
-                table=mapping.table,
+                table=self.metadata.tables[mapping.table],
                 columns=columns[1:],  # Strip off the Id column
                 record_iterable=f_values,
             )
-            ids_chunks = self._sql_bulk_insert_from_records_incremental(
+            ids_chunks = sql_bulk_insert_from_records_incremental(
                 connection=conn,
-                table=mapping.get_sf_id_table(),
+                table=self.metadata.tables[mapping.get_sf_id_table()],
                 columns=["sf_id"],
                 record_iterable=f_ids,
             )
diff --git a/cumulusci/tasks/bulkdata/load.py b/cumulusci/tasks/bulkdata/load.py
index 0172194be..bcdbf07dc 100644
--- a/cumulusci/tasks/bulkdata/load.py
+++ b/cumulusci/tasks/bulkdata/load.py
@@ -34,7 +34,11 @@ from cumulusci.tasks.bulkdata.step import (
     DataOperationType,
     get_dml_operation,
 )
-from cumulusci.tasks.bulkdata.utils import RowErrorChecker, SqlAlchemyMixin
+from cumulusci.tasks.bulkdata.utils import (
+    RowErrorChecker,
+    SqlAlchemyMixin,
+    sql_bulk_insert_from_records,
+)
 from cumulusci.tasks.salesforce import BaseSalesforceApiTask
 
 
@@ -364,9 +368,9 @@ class LoadData(SqlAlchemyMixin, BaseSalesforceApiTask):
         if is_insert_or_upsert and (
             step.job_result.records_processed - step.job_result.total_row_errors
         ):
-            self._sql_bulk_insert_from_records(
+            sql_bulk_insert_from_records(
                 connection=conn,
-                table=id_table_name,
+                table=self.metadata.tables[id_table_name],
                 columns=("id", "sf_id"),
                 record_iterable=results_generator,
             )
@@ -385,9 +389,9 @@ class LoadData(SqlAlchemyMixin, BaseSalesforceApiTask):
         ):
             account_id_lookup = mapping.lookups.get("AccountId")
             if account_id_lookup:
-                self._sql_bulk_insert_from_records(
+                sql_bulk_insert_from_records(
                     connection=conn,
-                    table=id_table_name,
+                    table=self.metadata.tables[id_table_name],
                     columns=("id", "sf_id"),
                     record_iterable=self._generate_contact_id_map_for_person_accounts(
                         mapping, account_id_lookup, conn
diff --git a/cumulusci/tasks/bulkdata/utils.py b/cumulusci/tasks/bulkdata/utils.py
index 2de789163..796ebc523 100644
--- a/cumulusci/tasks/bulkdata/utils.py
+++ b/cumulusci/tasks/bulkdata/utils.py
@@ -1,11 +1,13 @@
 import collections
 import logging
 import tempfile
+import typing as T
 from contextlib import contextmanager, nullcontext
 from pathlib import Path
 
 from simple_salesforce import Salesforce
 from sqlalchemy import Column, Integer, MetaData, Table, Unicode
+from sqlalchemy.engine.base import Connection
 from sqlalchemy.orm import Session, mapper
 
 from cumulusci.core.exceptions import BulkDataException
@@ -20,33 +22,6 @@ class SqlAlchemyMixin:
     session: Session
     sf: Salesforce
 
-    def _sql_bulk_insert_from_records(
-        self, *, connection, table, columns, record_iterable
-    ):
-        """Persist records from the given generator into the local database."""
-        consume(
-            self._sql_bulk_insert_from_records_incremental(
-                connection=connection,
-                table=table,
-                columns=columns,
-                record_iterable=record_iterable,
-            )
-        )
-
-    def _sql_bulk_insert_from_records_incremental(
-        self, *, connection, table, columns, record_iterable
-    ):
-        """Generator that persists batches of records from the given generator into the local database
-
-        Yields after every batch."""
-        table = self.metadata.tables[table]
-        dict_iterable = (dict(zip(columns, row)) for row in record_iterable)
-        for group in iterate_in_chunks(10000, dict_iterable):
-            with connection.begin():
-                connection.execute(table.insert(), group)
-            self.session.flush()
-            yield
-
     def _create_record_type_table(self, table_name):
         """Create a table to store mapping between Record Type Ids and Developer Names."""
         rt_map_model_name = f"{table_name}Model"
@@ -58,7 +33,7 @@ class SqlAlchemyMixin:
         rt_map_table = Table(table_name, self.metadata, *rt_map_fields)
         mapper(self.models[table_name], rt_map_table)
 
-    def _extract_record_types(self, sobject, table, conn):
+    def _extract_record_types(self, sobject, tablename: str, conn):
         """Query for Record Type information and persist it in the database."""
         self.logger.info(f"Extracting Record Types for {sobject}")
         query = (
@@ -68,9 +43,9 @@ class SqlAlchemyMixin:
         result = self.sf.query(query)
 
         if result["totalSize"]:
-            self._sql_bulk_insert_from_records(
+            sql_bulk_insert_from_records(
                 connection=conn,
-                table=table,
+                table=self.metadata.tables[tablename],
                 columns=["record_type_id", "developer_name"],
                 record_iterable=(
                     [rt["Id"], rt["DeveloperName"]] for rt in result["records"]
@@ -172,3 +147,39 @@ def consume(iterator):
     Simplified from the function in https://docs.python.org/3/library/itertools.html
     """
     collections.deque(iterator, maxlen=0)
+
+
+def sql_bulk_insert_from_records(
+    *,
+    connection: Connection,
+    table: Table,
+    columns: T.Tuple[str],
+    record_iterable: T.Iterable,
+) -> None:
+    """Persist records from the given generator into the local database."""
+    consume(
+        sql_bulk_insert_from_records_incremental(
+            connection=connection,
+            table=table,
+            columns=columns,
+            record_iterable=record_iterable,
+        )
+    )
+
+
+def sql_bulk_insert_from_records_incremental(
+    *,
+    connection: Connection,
+    table: Table,
+    columns: T.Tuple[str],
+    record_iterable: T.Iterable,
+):
+    """Generator that persists batches of records from the given generator into the local database
+
+    Yields after every batch."""
+    dict_iterable = (dict(zip(columns, row)) for row in record_iterable)
+    for group in iterate_in_chunks(10000, dict_iterable):
+        with connection.begin():
+            connection.execute(table.insert(), group)
+        # self.session.flush()  -- Did this line do anything?
+        yield


Test patch:
diff --git a/cumulusci/tasks/bulkdata/tests/test_extract.py b/cumulusci/tasks/bulkdata/tests/test_extract.py
index 9e1cfbf5f..96f480c36 100644
--- a/cumulusci/tasks/bulkdata/tests/test_extract.py
+++ b/cumulusci/tasks/bulkdata/tests/test_extract.py
@@ -353,15 +353,18 @@ class TestExtractData:
         )
         step = mock.Mock()
         task.session = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
+        task.metadata = mock.MagicMock()
 
-        task._import_results(mapping, step)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.extract.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._import_results(mapping, step)
 
         task.session.connection.assert_called_once_with()
         step.get_results.assert_called_once_with()
-        task._sql_bulk_insert_from_records.assert_called_once_with(
+        sql_bulk_insert_from_records.assert_called_once_with(
             connection=task.session.connection.return_value,
-            table=mapping.table,
+            table=task.metadata.tables[mapping.table],
             columns=["sf_id", "Name", "AccountId"],
             record_iterable=log_mock.return_value,
         )
@@ -414,21 +417,24 @@ class TestExtractData:
             [["006000000000001", (date.today() + timedelta(days=9)).isoformat()]]
         )
         task.session = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
+        task.metadata = mock.MagicMock()
         task._init_task()
-        task._import_results(mapping, step)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.extract.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._import_results(mapping, step)
 
         task.session.connection.assert_called_once_with()
         step.get_results.assert_called_once_with()
-        task._sql_bulk_insert_from_records.assert_called_once_with(
+        sql_bulk_insert_from_records.assert_called_once_with(
             connection=task.session.connection.return_value,
-            table=mapping.table,
+            table=task.metadata.tables[mapping.table],
             columns=["sf_id", "CloseDate"],
             record_iterable=mock.ANY,
         )
 
         result = list(
-            task._sql_bulk_insert_from_records.call_args_list[0][1]["record_iterable"]
+            sql_bulk_insert_from_records.call_args_list[0][1]["record_iterable"]
         )
         assert result == [["006000000000001", "2020-07-10"]]
 
@@ -440,9 +446,7 @@ class TestExtractData:
             {"options": {"database_url": "sqlite://", "mapping": mapping_path}},
         )
         task._extract_record_types = mock.Mock()
-        task._sql_bulk_insert_from_records_incremental = mock.Mock(
-            return_value=[None, None]
-        )
+        task.metadata = mock.MagicMock()
         task.session = mock.Mock()
         task.org_config._is_person_accounts_enabled = False
 
@@ -455,10 +459,14 @@ class TestExtractData:
             lookups={},
             table="accounts",
         )
-        task._import_results(
-            mapping,
-            step,
-        )
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.extract.sql_bulk_insert_from_records_incremental",
+            return_value=[None, None],
+        ):
+            task._import_results(
+                mapping,
+                step,
+            )
         task._extract_record_types.assert_called_once_with(
             "Account",
             mapping.get_source_record_type_table(),
@@ -473,8 +481,8 @@ class TestExtractData:
             {"options": {"database_url": "sqlite://", "mapping": mapping_path}},
         )
         task._extract_record_types = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.session = mock.Mock()
+        task.metadata = mock.MagicMock()
         task.org_config._is_person_accounts_enabled = True
 
         step = mock.Mock()
@@ -483,22 +491,25 @@ class TestExtractData:
             ["000000000000002", "Business Account", "012000000000002", "false"],
         ]
 
-        task._import_results(
-            MappingStep(
-                sf_object="Account",
-                fields={
-                    "Id": "sf_id",
-                    "Name": "Name",
-                    "RecordTypeId": "RecordTypeId",
-                    "IsPersonAccount": "IsPersonAccount",
-                },
-                lookups={},
-            ),
-            step,
-        )
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.extract.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._import_results(
+                MappingStep(
+                    sf_object="Account",
+                    fields={
+                        "Id": "sf_id",
+                        "Name": "Name",
+                        "RecordTypeId": "RecordTypeId",
+                        "IsPersonAccount": "IsPersonAccount",
+                    },
+                    lookups={},
+                ),
+                step,
+            )
 
-        task._sql_bulk_insert_from_records.assert_called()
-        args, kwargs = task._sql_bulk_insert_from_records.call_args_list[0]
+        sql_bulk_insert_from_records.assert_called()
+        args, kwargs = sql_bulk_insert_from_records.call_args_list[0]
 
         records = [record for record in kwargs["record_iterable"]]
 
@@ -901,15 +912,18 @@ class TestExtractData:
         )
         step = mock.Mock()
         task.session = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
+        task.metadata = mock.MagicMock()
 
-        task._import_results(mapping, step)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.extract.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._import_results(mapping, step)
 
         task.session.connection.assert_called_once_with()
         step.get_results.assert_called_once_with()
-        task._sql_bulk_insert_from_records.assert_called_once_with(
+        sql_bulk_insert_from_records.assert_called_once_with(
             connection=task.session.connection.return_value,
-            table="Opportunity",
+            table=task.metadata.tables["Opportunity"],
             columns=["sf_id", "Name", "account_id"],
             record_iterable=log_mock.return_value,
         )
diff --git a/cumulusci/tasks/bulkdata/tests/test_load.py b/cumulusci/tasks/bulkdata/tests/test_load.py
index 0c6e7c77e..46f65966b 100644
--- a/cumulusci/tasks/bulkdata/tests/test_load.py
+++ b/cumulusci/tasks/bulkdata/tests/test_load.py
@@ -852,9 +852,9 @@ class TestLoadData:
             {"options": {"database_url": "sqlite://", "mapping": "mapping.yml"}},
         )
 
-        task.session = mock.Mock()
+        task.session = mock.MagicMock()
+        task.metadata = mock.MagicMock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
 
@@ -870,11 +870,14 @@ class TestLoadData:
         step.results = [DataOperationResult("001111111111111", True, None)]
 
         mapping = MappingStep(sf_object="Account")
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._process_job_results(mapping, step, local_ids)
 
         task.session.connection.assert_called_once()
         task._initialize_id_table.assert_called_once_with(mapping, True)
-        task._sql_bulk_insert_from_records.assert_called_once()
+        sql_bulk_insert_from_records.assert_called_once()
         task.session.commit.assert_called_once()
 
     def test_process_job_results__insert_rows_fail(self):
@@ -891,7 +894,6 @@ class TestLoadData:
 
         task.session = mock.Mock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
         task.logger = mock.Mock()
@@ -917,11 +919,14 @@ class TestLoadData:
         ]
 
         mapping = MappingStep(sf_object="Account", table="Account")
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._process_job_results(mapping, step, local_ids)
 
         task.session.connection.assert_called_once()
         task._initialize_id_table.assert_called_once_with(mapping, True)
-        task._sql_bulk_insert_from_records.assert_not_called()
+        sql_bulk_insert_from_records.assert_not_called()
         task.session.commit.assert_called_once()
         assert len(task.logger.mock_calls) == 4
 
@@ -933,7 +938,6 @@ class TestLoadData:
 
         task.session = mock.Mock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
 
@@ -949,11 +953,14 @@ class TestLoadData:
         step.results = [DataOperationResult("001111111111111", True, None)]
 
         mapping = MappingStep(sf_object="Account", action=DataOperationType.UPDATE)
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._process_job_results(mapping, step, local_ids)
 
         task.session.connection.assert_not_called()
         task._initialize_id_table.assert_not_called()
-        task._sql_bulk_insert_from_records.assert_not_called()
+        sql_bulk_insert_from_records.assert_not_called()
         task.session.commit.assert_not_called()
 
     def test_process_job_results__exception_failure(self):
@@ -964,7 +971,6 @@ class TestLoadData:
 
         task.session = mock.Mock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
 
@@ -982,7 +988,9 @@ class TestLoadData:
 
         mapping = MappingStep(sf_object="Account", action=DataOperationType.UPDATE)
 
-        with pytest.raises(BulkDataException) as e:
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"
+        ), pytest.raises(BulkDataException) as e:
             task._process_job_results(mapping, step, local_ids)
 
         assert "Error on record with id" in str(e.value)
@@ -1019,7 +1027,6 @@ class TestLoadData:
 
         task.session = mock.Mock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
         task._can_load_person_accounts = mock.Mock(
@@ -1046,7 +1053,8 @@ class TestLoadData:
         )
         if account_id_lookup:
             mapping.lookups["AccountId"] = account_id_lookup
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch("cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"):
+            task._process_job_results(mapping, step, local_ids)
 
         task._generate_contact_id_map_for_person_accounts.assert_not_called()
 
@@ -1080,8 +1088,8 @@ class TestLoadData:
         )
 
         task.session = mock.Mock()
+        task.metadata = mock.MagicMock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
         task._can_load_person_accounts = mock.Mock(
@@ -1108,7 +1116,8 @@ class TestLoadData:
         )
         if account_id_lookup:
             mapping.lookups["AccountId"] = account_id_lookup
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch("cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"):
+            task._process_job_results(mapping, step, local_ids)
 
         task._generate_contact_id_map_for_person_accounts.assert_not_called()
 
@@ -1141,9 +1150,9 @@ class TestLoadData:
             {"options": {"database_url": "sqlite://", "mapping": "mapping.yml"}},
         )
 
-        task.session = mock.Mock()
+        task.session = mock.MagicMock()
+        task.metadata = mock.MagicMock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
         task._can_load_person_accounts = mock.Mock(
@@ -1170,7 +1179,8 @@ class TestLoadData:
         )
         if account_id_lookup:
             mapping.lookups["AccountId"] = account_id_lookup
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch("cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"):
+            task._process_job_results(mapping, step, local_ids)
 
         task._generate_contact_id_map_for_person_accounts.assert_not_called()
 
@@ -1204,8 +1214,8 @@ class TestLoadData:
         )
 
         task.session = mock.Mock()
+        task.metadata = mock.MagicMock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
         task._can_load_person_accounts = mock.Mock(
@@ -1232,7 +1242,8 @@ class TestLoadData:
         )
         if account_id_lookup:
             mapping.lookups["AccountId"] = account_id_lookup
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch("cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"):
+            task._process_job_results(mapping, step, local_ids)
 
         task._generate_contact_id_map_for_person_accounts.assert_not_called()
 
@@ -1264,8 +1275,8 @@ class TestLoadData:
         )
 
         task.session = mock.Mock()
+        task.metadata = mock.MagicMock()
         task._initialize_id_table = mock.Mock()
-        task._sql_bulk_insert_from_records = mock.Mock()
         task.bulk = mock.Mock()
         task.sf = mock.Mock()
         task._can_load_person_accounts = mock.Mock(
@@ -1290,16 +1301,18 @@ class TestLoadData:
             action=action,
             lookups={"AccountId": account_id_lookup},
         )
-
-        task._process_job_results(mapping, step, local_ids)
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.load.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            task._process_job_results(mapping, step, local_ids)
 
         task._generate_contact_id_map_for_person_accounts.assert_called_once_with(
             mapping, mapping.lookups["AccountId"], task.session.connection.return_value
         )
 
-        task._sql_bulk_insert_from_records.assert_called_with(
+        sql_bulk_insert_from_records.assert_called_with(
             connection=task.session.connection.return_value,
-            table=task._initialize_id_table.return_value,
+            table=task.metadata.tables[task._initialize_id_table.return_value],
             columns=("id", "sf_id"),
             record_iterable=task._generate_contact_id_map_for_person_accounts.return_value,
         )
diff --git a/cumulusci/tasks/bulkdata/tests/test_utils.py b/cumulusci/tasks/bulkdata/tests/test_utils.py
index 455caef31..fa890f851 100644
--- a/cumulusci/tasks/bulkdata/tests/test_utils.py
+++ b/cumulusci/tasks/bulkdata/tests/test_utils.py
@@ -7,7 +7,11 @@ from sqlalchemy.orm import create_session, mapper
 
 from cumulusci.tasks import bulkdata
 from cumulusci.tasks.bulkdata.mapping_parser import parse_from_yaml
-from cumulusci.tasks.bulkdata.utils import create_table, generate_batches
+from cumulusci.tasks.bulkdata.utils import (
+    create_table,
+    generate_batches,
+    sql_bulk_insert_from_records,
+)
 from cumulusci.utils import temporary_dir
 
 
@@ -44,24 +48,27 @@ class TestSqlAlchemyMixin:
     @responses.activate
     def test_extract_record_types(self):
         util = bulkdata.utils.SqlAlchemyMixin()
-        util._sql_bulk_insert_from_records = mock.Mock()
         util.sf = mock.Mock()
         util.sf.query.return_value = {
             "totalSize": 1,
             "records": [{"Id": "012000000000000", "DeveloperName": "Organization"}],
         }
         util.logger = mock.Mock()
+        util.metadata = mock.MagicMock()
 
-        conn = mock.Mock()
-        util._extract_record_types("Account", "test_table", conn)
+        conn = mock.MagicMock()
+        with mock.patch(
+            "cumulusci.tasks.bulkdata.utils.sql_bulk_insert_from_records"
+        ) as sql_bulk_insert_from_records:
+            util._extract_record_types("Account", "test_table", conn)
 
         util.sf.query.assert_called_once_with(
             "SELECT Id, DeveloperName FROM RecordType WHERE SObjectType='Account'"
         )
-        util._sql_bulk_insert_from_records.assert_called_once()
-        call = util._sql_bulk_insert_from_records.call_args_list[0][1]
+        sql_bulk_insert_from_records.assert_called_once()
+        call = sql_bulk_insert_from_records.call_args_list[0][1]
         assert call["connection"] == conn
-        assert call["table"] == "test_table"
+        assert call["table"] == util.metadata.tables["test_table"]
         assert call["columns"] == ["record_type_id", "developer_name"]
         assert list(call["record_iterable"]) == [["012000000000000", "Organization"]]
 
@@ -76,15 +83,12 @@ class TestSqlAlchemyMixin:
         model = type("TestModel", (object,), {})
         mapper(model, id_t)
 
-        util = bulkdata.utils.SqlAlchemyMixin()
-        util.metadata = metadata
         session = create_session(bind=engine, autocommit=False)
-        util.session = session
         connection = session.connection()
 
-        util._sql_bulk_insert_from_records(
+        sql_bulk_insert_from_records(
             connection=connection,
-            table="TestTable",
+            table=id_t,
             columns=("id", "sf_id"),
             record_iterable=([f"{x}", f"00100000000000{x}"] for x in range(10)),
         )