DEVOPS-814 chore: sync CumulusCI from upstream (2026.02.12) (#21)

Merge pull request #21 from ClaritiSoftware/chore/DEVOPS-814

Author: dipakparmar

cumulusci/tasks/bulkdata/generate_from_yaml.py

@@ -95,6 +95,9 @@ def _init_options(self, kwargs):
         self.working_directory = self.options.get("working_directory")
         loading_rules = process_list_arg(self.options.get("loading_rules")) or []
         self.loading_rules = [Path(path) for path in loading_rules if path]
+        # These flags may be provided internally by callers (e.g., Snowfakery task)
+        self.strict_mode = bool(self.options.get("strict_mode", False))
+        self.validate_only = bool(self.options.get("validate_only", False))
 
     def _generate_data(self, db_url, mapping_file_path, num_records, current_batch_num):
         """Generate all of the data"""
@@ -171,6 +174,7 @@ def generate_data(self, dburl, num_records, current_batch_num):
                     "project_config": self.project_config,
                     **self.plugin_options,
                 },
+                strict_mode=self.strict_mode or self.validate_only,
             )
 
         if (

cumulusci/tasks/bulkdata/load.py

@@ -28,6 +28,7 @@
     AddPersonAccountsToQuery,
     AddRecordTypesToQuery,
     DynamicLookupQueryExtender,
+    register_sqlite_functions,
 )
 from cumulusci.tasks.bulkdata.step import (
     DEFAULT_BULK_BATCH_SIZE,
@@ -766,6 +767,9 @@ def _init_db(self):
         with self._database_url() as database_url:
             parent_engine = create_engine(database_url)
             with parent_engine.connect() as connection:
+                # Register custom SQLite functions for smart lookup resolution
+                register_sqlite_functions(connection)
+
                 # initialize the DB session
                 self.session = Session(connection)
 

cumulusci/tasks/bulkdata/mapping_parser.py

@@ -444,9 +444,9 @@ def replace_if_necessary(dct, name, replacement):
             except KeyError:
                 message = f"Field {self.sf_object}.{f} does not exist or is not visible to the current user."
                 if validation_result:
-                    validation_result.add_warning(message)
+                    validation_result.add_error(message)
                 else:
-                    logger.warning(message)
+                    logger.error(message)
             else:
                 del field_dict[f]
                 field_dict[new_name] = entry
@@ -463,9 +463,9 @@ def replace_if_necessary(dct, name, replacement):
             if f not in describe:
                 message = f"Field {self.sf_object}.{f} does not exist or is not visible to the current user."
                 if validation_result:
-                    validation_result.add_warning(message)
+                    validation_result.add_error(message)
                 else:
-                    logger.warning(message)
+                    logger.error(message)
                 error_in_f = True
             elif not self._check_field_permission(
                 describe,
@@ -480,9 +480,9 @@ def replace_if_necessary(dct, name, replacement):
                     + f"{relevant_permissions} for this operation."
                 )
                 if validation_result:
-                    validation_result.add_warning(message)
+                    validation_result.add_error(message)
                 else:
-                    logger.warning(message)
+                    logger.error(message)
                 error_in_f = True
 
             if error_in_f:
@@ -514,9 +514,9 @@ def _validate_sobject(
         except KeyError:
             message = f"sObject {self.sf_object} does not exist or is not visible to the current user."
             if validation_result:
-                validation_result.add_warning(message)
+                validation_result.add_error(message)
             else:
-                logger.warning(message)
+                logger.error(message)
             return False
 
         # Validate our access to this sObject.
@@ -525,9 +525,9 @@ def _validate_sobject(
         ):
             message = f"sObject {self.sf_object} does not have the correct permissions for {data_operation_type}."
             if validation_result:
-                validation_result.add_warning(message)
+                validation_result.add_error(message)
             else:
-                logger.warning(message)
+                logger.error(message)
             return False
 
         return True
@@ -545,7 +545,7 @@ def check_required(
             if fields_describe[field]["createable"] and not defaulted:
                 required_fields.add(field)
         missing_fields = required_fields.difference(
-            set(self.fields.keys()) | set(self.lookups)
+            set(self.fields.keys()) | set(self.lookups) | set(self.static.keys())
         )
         if len(missing_fields) > 0:
             message = f"One or more required fields are missing for loading on {self.sf_object} :{missing_fields}"

cumulusci/tasks/bulkdata/query_transformers.py

@@ -1,7 +1,8 @@
+import re
 import typing as T
 from functools import cached_property
 
-from sqlalchemy import String, and_, func, text
+from sqlalchemy import String, and_, case, func, text
 from sqlalchemy.orm import Query, aliased
 from sqlalchemy.sql import literal_column
 
@@ -10,6 +11,29 @@
 Criterion = T.Any
 ID_TABLE_NAME = "cumulusci_id_table"
 
+# Salesforce ID pattern: 15 or 18 alphanumeric characters
+# This matches the OID_REGEX pattern used in robotframework/Salesforce.py
+SF_ID_PATTERN = re.compile(r"^[a-zA-Z0-9]{15}$|^[a-zA-Z0-9]{18}$")
+
+
+def is_salesforce_id(value: T.Optional[str]) -> bool:
+    """Check if a value looks like a valid Salesforce ID."""
+    if value is None:
+        return False
+    return bool(SF_ID_PATTERN.match(str(value)))
+
+
+def _is_salesforce_id_sqlite(value: T.Optional[str]) -> int:
+    """SQLite UDF wrapper for is_salesforce_id."""
+    return 1 if is_salesforce_id(value) else 0
+
+
+def register_sqlite_functions(connection) -> None:
+    """Register custom SQLite functions on a database connection."""
+    # Get the underlying DBAPI connection
+    dbapi_connection = connection.connection.dbapi_connection
+    dbapi_connection.create_function("is_salesforce_id", 1, _is_salesforce_id_sqlite)
+
 
 class LoadQueryExtender:
     """Class that transforms a load.py query with columns, filters, joins"""
@@ -61,9 +85,26 @@ def __init__(self, mapping, metadata, model, _old_format) -> None:
 
     @cached_property
     def columns_to_add(self):
+        """Build column expressions for lookup fields with smart ID resolution."""
+        columns = []
         for lookup in self.lookups:
             lookup.aliased_table = aliased(self.metadata.tables[ID_TABLE_NAME])
-        return [lookup.aliased_table.columns.sf_id for lookup in self.lookups]
+            key_field = lookup.get_lookup_key_field(self.model)
+            value_column = getattr(self.model, key_field)
+
+            # The resolved SF ID from the ID table join (may be NULL)
+            sf_id_from_table = lookup.aliased_table.columns.sf_id
+
+            smart_lookup = case(
+                # If we found a match in the ID table, use that
+                (sf_id_from_table.isnot(None), sf_id_from_table),
+                # If the original value is already a SF ID, use it directly
+                (func.is_salesforce_id(value_column) == 1, value_column),
+                # Otherwise return NULL (lookup not found)
+                else_=None,
+            )
+            columns.append(smart_lookup)
+        return columns
 
     @cached_property
     def outerjoins_to_add(self):

cumulusci/tasks/bulkdata/snowfakery.py

@@ -141,9 +141,12 @@ class Snowfakery(BaseSalesforceApiTask):
             "Defaults to False."
         },
         "validate_only": {
-            "description": "Boolean: if True, only validate the generated mapping against the org schema without loading data. "
+            "description": "Boolean: if True, validates snowfakery recipe and generated mapping against the org schema without loading data. "
             "Defaults to False."
         },
+        "strict_mode": {
+            "description": "Boolean: If True, validates the Snowfakery recipe and generated mapping against the org schema (strict mode) and then proceeds with the run",
+        },
     }
 
     def _validate_options(self):
@@ -165,6 +168,7 @@ def _validate_options(self):
             self.options.get("drop_missing_schema", False)
         )
         self.validate_only = process_bool_arg(self.options.get("validate_only", False))
+        self.strict_mode = process_bool_arg(self.options.get("strict_mode", False))
 
         loading_rules = process_list_arg(self.options.get("loading_rules")) or []
         self.loading_rules = [Path(path) for path in loading_rules if path]
@@ -614,6 +618,7 @@ def _run_generate_and_load_subtask(
             "ignore_row_errors": self.ignore_row_errors,
             "drop_missing_schema": self.drop_missing_schema,
             "validate_only": validate_only,
+            "strict_mode": self.strict_mode,
         }
         subtask_config = TaskConfig({"options": options})
         subtask = GenerateAndLoadDataFromYaml(

cumulusci/tasks/bulkdata/tests/test_generate_from_snowfakery_task.py

@@ -206,6 +206,102 @@ def test_exception_handled_cleanly(self, generate_data):
             assert "Foo" in str(e.value)
         assert len(generate_data.mock_calls) == 1
 
+    @mock.patch("cumulusci.tasks.bulkdata.generate_from_yaml.generate_data")
+    def test_validate_only_forces_strict_mode(self, generate_data):
+        with temp_sqlite_database_url() as database_url:
+            task = _make_task(
+                GenerateDataFromYaml,
+                {
+                    "options": {
+                        "generator_yaml": simple_yaml,
+                        "database_url": database_url,
+                        "validate_only": True,
+                    }
+                },
+            )
+            task()
+
+        assert len(generate_data.mock_calls) == 1
+        _, kwargs = generate_data.call_args
+        assert kwargs["strict_mode"] is True
+
+    @mock.patch("cumulusci.tasks.bulkdata.generate_from_yaml.generate_data")
+    def test_strict_mode_only(self, generate_data):
+        with temp_sqlite_database_url() as database_url:
+            task = _make_task(
+                GenerateDataFromYaml,
+                {
+                    "options": {
+                        "generator_yaml": simple_yaml,
+                        "database_url": database_url,
+                        "strict_mode": True,
+                    }
+                },
+            )
+            task()
+
+        assert len(generate_data.mock_calls) == 1
+        _, kwargs = generate_data.call_args
+        assert kwargs["strict_mode"] is True
+
+    @mock.patch("cumulusci.tasks.bulkdata.generate_from_yaml.generate_data")
+    def test_defaults_no_strict_no_validate(self, generate_data):
+        with temp_sqlite_database_url() as database_url:
+            task = _make_task(
+                GenerateDataFromYaml,
+                {
+                    "options": {
+                        "generator_yaml": simple_yaml,
+                        "database_url": database_url,
+                    }
+                },
+            )
+            task()
+
+        assert len(generate_data.mock_calls) == 1
+        _, kwargs = generate_data.call_args
+        assert kwargs["strict_mode"] is False
+
+    def test_validate_only_runs_and_creates_mapping(self):
+        with temporary_file_path("mapping.yml") as mapping_path:
+            with temp_sqlite_database_url() as database_url:
+                task = _make_task(
+                    GenerateDataFromYaml,
+                    {
+                        "options": {
+                            "generator_yaml": simple_yaml,
+                            "database_url": database_url,
+                            "generate_mapping_file": mapping_path,
+                            "validate_only": True,
+                        }
+                    },
+                )
+                task()
+
+            assert mapping_path.exists()
+            mapping = yaml.safe_load(open(mapping_path))
+            assert mapping  # ensure something was written
+
+    def test_strict_mode_runs_and_creates_mapping(self):
+        with temporary_file_path("mapping.yml") as mapping_path:
+            with temp_sqlite_database_url() as database_url:
+                task = _make_task(
+                    GenerateDataFromYaml,
+                    {
+                        "options": {
+                            "generator_yaml": simple_yaml,
+                            "database_url": database_url,
+                            "generate_mapping_file": mapping_path,
+                            "strict_mode": True,
+                        }
+                    },
+                )
+                task()
+
+            assert mapping_path.exists()
+            mapping = yaml.safe_load(open(mapping_path))
+            assert mapping  # ensure something was written
+
     @mock.patch(
         "cumulusci.tasks.bulkdata.generate_and_load_data_from_yaml.GenerateAndLoadDataFromYaml._dataload"
     )