Repository: SFDO-Tooling/CumulusCI
Pull request #3109 (2022-03-09 00:31:07)

Snowfakery crash when loading into "directly connected" ("persistent") orgs
Snowfakery had a bug introduced when the get_service calling code changed in 40520bee4c1c .

This PR changes the method signature for SubprocessKeyChain.get_service()

It's hard to replicate in integration tests because it depends on a connected app-based org_config as opposed to the sfd-based org_config

W-10811651

# Issues Closed

Fixed a bug running the `snowfakery` task in parallel mode with orgs connected using a Connected App.



Patch:
diff --git a/cumulusci/core/keychain/subprocess_keychain.py b/cumulusci/core/keychain/subprocess_keychain.py
new file mode 100644
index 000000000..4dc20db7f
--- /dev/null
+++ b/cumulusci/core/keychain/subprocess_keychain.py
@@ -0,0 +1,24 @@
+import typing as T
+
+from cumulusci.core.exceptions import ServiceNotConfigured
+
+
+class SubprocessKeychain(T.NamedTuple):
+    """A pretend, in-memory keychain that knows about connected apps and nothing else.
+
+    This keychain is used by Snowfakery to make the connected app available to
+    tasks that are running in sub-processes. If Snowfakery ever needs a broader
+    range of services thaan just the connected app, this keychain will need to
+    become more complex.
+    """
+
+    connected_app: T.Any = None
+
+    def get_service(self, name, alias=None):
+        if name == "connected_app" and self.connected_app:
+            return self.connected_app
+
+        raise ServiceNotConfigured(name)
+
+    def set_org(self, *args):
+        pass
diff --git a/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py b/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py
index d25af2e8c..4bf2692bc 100644
--- a/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py
+++ b/cumulusci/tasks/bulkdata/snowfakery_utils/queue_manager.py
@@ -217,8 +217,9 @@ class Channel:
     def _configure_queues(self, recipe_options):
         """Configure two ParallelWorkerQueues for datagen and dataload"""
         try:
-            # in the future, the connected_app should come from the org
-            connected_app = self.project_config.keychain.get_service("connected_app")
+            connected_app = self.project_config.keychain.get_service(
+                "connected_app", self.org_config.connected_app
+            )
         except exc.ServiceNotConfigured:  # pragma: no cover
             # to discuss...when can this happen? What are the consequences?
             connected_app = None
diff --git a/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py b/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py
index d2258ad30..72c3b84ad 100644
--- a/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py
+++ b/cumulusci/utils/parallel/task_worker_queues/parallel_worker.py
@@ -17,7 +17,7 @@ from cumulusci.core.config import (
     TaskConfig,
     UniversalConfig,
 )
-from cumulusci.core.exceptions import ServiceNotConfigured
+from cumulusci.core.keychain.subprocess_keychain import SubprocessKeychain
 from cumulusci.core.utils import import_global
 
 
@@ -28,7 +28,7 @@ class SharedConfig(BaseModel):
     org_config: OrgConfig
     failures_dir: Path
     redirect_logging: bool
-    connected_app: T.Optional[BaseConfig]  # send a list of services
+    connected_app: T.Optional[BaseConfig]  # a connected app service
     outbox_dir: Path  # where do jobs go when they are done
 
     class Config:
@@ -114,7 +114,7 @@ class TaskWorker:
             self.task_options["working_directory"] = self.worker_config.working_dir
         task_config = TaskConfig({"options": self.task_options})
         connected_app = self.connected_app
-        keychain = SubprocessKeyChain(connected_app)
+        keychain = SubprocessKeychain(connected_app)
         self.project_config.set_keychain(keychain)
         self.org_config.keychain = keychain
         return task_class(
@@ -236,18 +236,3 @@ class ParallelWorker:
 
     def __repr__(self):
         return f"<Worker {self.worker_config.task_class.__name__} {self.worker_config.working_dir.name} Alive: {self.is_alive()}>"
-
-
-class SubprocessKeyChain(T.NamedTuple):
-    """A pretend, in-memory keychain that knows about connected apps and nothing else."""
-
-    connected_app: T.Any = None
-
-    def get_service(self, name):
-        if name == "connected_app" and self.connected_app:
-            return self.connected_app
-
-        raise ServiceNotConfigured(name)
-
-    def set_org(self, *args):
-        pass


Test patch:
diff --git a/cumulusci/tests/util.py b/cumulusci/tests/util.py
index 641790b7d..b4774a977 100644
--- a/cumulusci/tests/util.py
+++ b/cumulusci/tests/util.py
@@ -102,7 +102,7 @@ class DummyService(BaseConfig):
     password = "dummy_password"
     client_id = "ZOOMZOOM"
 
-    def __init__(self, name):
+    def __init__(self, name, alias):
         self.name = name
         super().__init__(name)
 
@@ -124,8 +124,8 @@ class DummyKeychain(BaseProjectKeychain):
     def cache_dir(self):
         return self._cache_dir or Path.home() / "project/.cci"
 
-    def get_service(self, name):
-        return DummyService(name)
+    def get_service(self, name, alias=None):
+        return DummyService(name, alias)
 
     def set_org(self, org: OrgConfig, global_org: bool):
         pass
diff --git a/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py b/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py
index c33943b12..6d49967ea 100644
--- a/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py
+++ b/cumulusci/utils/parallel/task_worker_queues/tests/test_parallel_worker.py
@@ -12,7 +12,7 @@ from cumulusci.core.config import BaseProjectConfig, OrgConfig, UniversalConfig
 from cumulusci.tasks.util import Sleep
 from cumulusci.utils.parallel.task_worker_queues.parallel_worker import (
     ParallelWorker,
-    SubprocessKeyChain,
+    SubprocessKeychain,
     TaskWorker,
     WorkerConfig,
 )
@@ -343,9 +343,9 @@ class TestTaskWorker:
 # Also we usually mock refresh_oauth_token which is what would
 # invoke this. In other words, using integration tests to cover this
 # function is far easier than using unit test.
-class TestSubprocessKeyChain:
+class TestSubprocessKeychain:
     def test_subprocess_keychain(self):
-        skc = SubprocessKeyChain("Blah")
+        skc = SubprocessKeychain("Blah")
 
         assert skc.get_service("connected_app") == "Blah"
         skc.set_org()