Repository: SFDO-Tooling/CumulusCI
Pull request #3695 (2023-11-03 07:07:01)

Make task `options` read only when using Pydantic option validation
[W-13798912](https://gus.lightning.force.com/lightning/r/ADM_Work__c/a07EE00001WfO3eYAF/view)

Made `self.options` to be Read-Only, if an `Options` class is defined in the Task. Added usage to documentation. Additionally, corrected `ListMetadataTypes` in `cumulusci/tasks/util.py` to use `parsed_options` instead of `options` (was missed in the pydantic options commit). 

Fixes: #3627


Patch:
diff --git a/cumulusci/core/tasks.py b/cumulusci/core/tasks.py
index e9f5b2f11..0d8269aea 100644
--- a/cumulusci/core/tasks.py
+++ b/cumulusci/core/tasks.py
@@ -28,7 +28,7 @@ from cumulusci.core.flowrunner import FlowCoordinator, StepSpec, StepVersion
 from cumulusci.utils import cd
 from cumulusci.utils.logging import redirect_output_to_logger
 from cumulusci.utils.metaprogramming import classproperty
-from cumulusci.utils.options import CCIOptions
+from cumulusci.utils.options import CCIOptions, ReadOnlyOptions
 
 CURRENT_TASK = threading.local()
 
@@ -163,6 +163,7 @@ class BaseTask:
                     opt: val for opt, val in self.options.items() if opt not in specials
                 }
                 self.parsed_options = self.Options(**options_without_specials)
+                self.options = ReadOnlyOptions(self.options)
             except ValidationError as e:
                 try:
                     errors = [
diff --git a/cumulusci/tasks/util.py b/cumulusci/tasks/util.py
index afb574dfc..50d43ce82 100644
--- a/cumulusci/tasks/util.py
+++ b/cumulusci/tasks/util.py
@@ -59,13 +59,13 @@ class ListMetadataTypes(BaseTask):
 
     def _init_options(self, kwargs):
         super(ListMetadataTypes, self)._init_options(kwargs)
-        if not self.options.get("package_xml"):
-            self.options["package_xml"] = os.path.join(
+        if not self.parsed_options.get("package_xml"):
+            self.parsed_options["package_xml"] = os.path.join(
                 self.project_config.repo_root, "src", "package.xml"
             )
 
     def _run_task(self):
-        dom = parse(self.options["package_xml"])
+        dom = parse(self.parsed_options["package_xml"])
         package = dom.getElementsByTagName("Package")[0]
         types = package.getElementsByTagName("types")
         type_list = []
@@ -75,7 +75,7 @@ class ListMetadataTypes(BaseTask):
             type_list.append(metadata_type)
         self.logger.info(
             "Metadata types found in %s:\r\n%s",
-            self.options["package_xml"],
+            self.parsed_options["package_xml"],
             "\r\n".join(type_list),
         )
 
diff --git a/cumulusci/utils/options.py b/cumulusci/utils/options.py
index 3f9fe11d9..5d5d2828f 100644
--- a/cumulusci/utils/options.py
+++ b/cumulusci/utils/options.py
@@ -4,8 +4,13 @@ from typing import Any, Dict, List
 
 from pydantic import DirectoryPath, Field, FilePath, create_model
 
+from cumulusci.core.exceptions import TaskOptionsError
 from cumulusci.utils.yaml.model_parser import CCIDictModel
 
+READONLYDICT_ERROR_MSG = (
+    "The 'options' dictionary is read-only. Please use 'parsed_options' instead."
+)
+
 
 def _describe_field(field):
     "Convert a Pydantic field into a CCI task_option dict"
@@ -18,6 +23,22 @@ def _describe_field(field):
     return rc
 
 
+class ReadOnlyOptions(dict):
+    """To enforce self.options to be read-only"""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    def __setitem__(self, key, value):
+        raise TaskOptionsError(READONLYDICT_ERROR_MSG)
+
+    def __delitem__(self, key):
+        raise TaskOptionsError(READONLYDICT_ERROR_MSG)
+
+    def pop(self, key, default=None):
+        raise TaskOptionsError(READONLYDICT_ERROR_MSG)
+
+
 class CCIOptions(CCIDictModel):
     "Base class for all options in tasks"
 
diff --git a/docs/config.md b/docs/config.md
index 12e3db1e8..99ca4fc41 100644
--- a/docs/config.md
+++ b/docs/config.md
@@ -176,6 +176,10 @@ and (2) A required file path.
 
 Once the options are defined, they can be accessed via the `parsed_options` property of the task.
 
+```{important}
+When the nested `Options` class is defined within your custom task (or is part of a class you inherit from), it restricts modifications to the `options` property of the task, making it read-only. To make any changes, you should instead modify the `parsed_options` property rather than the `options` property.
+```
+
 Some of the most commonly used types are:
 
 -   `pathlib.Path`: simply uses the type itself for validation by passing the value to Path(v);


Test patch:
diff --git a/cumulusci/utils/tests/test_option_parsing.py b/cumulusci/utils/tests/test_option_parsing.py
index 82fceb962..d36dd44df 100644
--- a/cumulusci/utils/tests/test_option_parsing.py
+++ b/cumulusci/utils/tests/test_option_parsing.py
@@ -12,10 +12,12 @@ from cumulusci.core.config import (
 from cumulusci.core.exceptions import TaskOptionsError
 from cumulusci.core.tasks import BaseTask
 from cumulusci.utils.options import (
+    READONLYDICT_ERROR_MSG,
     CCIOptions,
     Field,
     ListOfStringsOption,
     MappingOption,
+    ReadOnlyOptions,
 )
 
 ORG_ID = "00D000000000001"
@@ -45,6 +47,10 @@ class TaskToTestTypes(BaseTask):
                 print(key, repr(getattr(self.parsed_options, key)))
 
 
+class TaskWithoutOptions(BaseTask):
+    pass
+
+
 class TestTaskOptionsParsing:
     def setup_class(self):
         self.global_config = UniversalConfig()
@@ -154,3 +160,29 @@ class TestTaskOptionsParsing:
         assert "the_bool" in str(e.value)
         assert "req" in str(e.value)
         assert "Errors" in str(e.value)
+
+    def test_options_read_only(self):
+        # Has an Options class
+        task1 = TaskToTestTypes(self.project_config, self.task_config, self.org_config)
+        assert isinstance(task1.options, ReadOnlyOptions)
+        # Does not have an Options class
+        task2 = TaskWithoutOptions(
+            self.project_config, self.task_config, self.org_config
+        )
+        assert isinstance(task2.options, dict)
+
+    def test_init_options__options_read_only_error(self):
+        expected_error_msg = READONLYDICT_ERROR_MSG
+        task = TaskToTestTypes(self.project_config, self.task_config, self.org_config)
+        # Add new option
+        with pytest.raises(TaskOptionsError, match=expected_error_msg):
+            task.options["new_option"] = "something"
+        # Modify existing option
+        with pytest.raises(TaskOptionsError, match=expected_error_msg):
+            task.options["test_option"] = 456
+        # Delete existing option
+        with pytest.raises(TaskOptionsError, match=expected_error_msg):
+            del task.options["test_option"]
+        # Pop existing option
+        with pytest.raises(TaskOptionsError, match=expected_error_msg):
+            task.options.pop("test_option")