From 77a3d3daccea5d3d83ad7458fd6f55080da64797 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Thu, 1 Sep 2022 13:18:17 -0500
Subject: [PATCH 01/12] Send an email for each email-based dataset at the end
 of privacy request execution.

- Add a migration to create a new audit log type.  Create an audit log for the email send.
-  Throw an exception for email-based connectors and catch to override the default execution log.
- Add a draft of an email template
- Connect sending a "test email" with dummy data.  A fidesops admin could configure to check their email config was working.
---
 src/fidesops/ops/common_exceptions.py         |  4 ++
 .../ops/email_templates/get_email_template.py |  3 +
 .../ops/email_templates/template_names.py     |  1 +
 .../erasure_request_email_fulfillment.html    | 30 ++++++++++
 .../912d801f06c0_audit_log_email_send.py      | 32 ++++++++++
 src/fidesops/ops/schemas/email/email.py       |  1 +
 .../ops/service/connectors/email_connector.py | 59 +++++++++++++++++--
 .../service/email/email_dispatch_service.py   | 18 +++++-
 .../privacy_request/request_runner_service.py | 46 ++++++++++++++-
 src/fidesops/ops/task/graph_task.py           | 21 +++++--
 .../test_connection_config_endpoints.py       | 42 +++++++++++--
 tests/ops/fixtures/email_fixtures.py          |  1 +
 .../test_integration_email.py                 | 54 ++++++++++++++++-
 .../request_runner_service_test.py            | 43 ++++++++++++++
 14 files changed, 333 insertions(+), 22 deletions(-)
 create mode 100644 src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html
 create mode 100644 src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py

diff --git a/src/fidesops/ops/common_exceptions.py b/src/fidesops/ops/common_exceptions.py
index f5efcf33e..3567a85fb 100644
--- a/src/fidesops/ops/common_exceptions.py
+++ b/src/fidesops/ops/common_exceptions.py
@@ -101,6 +101,10 @@ class PrivacyRequestPaused(BaseException):
     """Halt Instruction Received on Privacy Request"""
 
 
+class PrivacyRequestErasureEmailSendRequired(BaseException):
+    """Erasure requests will need to be fulfilled by email send.  Exception is raised to change ExecutionLog details"""
+
+
 class SaaSConfigNotFoundException(FidesopsException):
     """Custom Exception - SaaS Config Not Found"""
 
diff --git a/src/fidesops/ops/email_templates/get_email_template.py b/src/fidesops/ops/email_templates/get_email_template.py
index 98c3f3c15..833dc36f8 100644
--- a/src/fidesops/ops/email_templates/get_email_template.py
+++ b/src/fidesops/ops/email_templates/get_email_template.py
@@ -5,6 +5,7 @@
 
 from fidesops.ops.common_exceptions import EmailTemplateUnhandledActionType
 from fidesops.ops.email_templates.template_names import (
+    EMAIL_ERASURE_REQUEST_FULFILLMENT,
     SUBJECT_IDENTITY_VERIFICATION_TEMPLATE,
 )
 from fidesops.ops.schemas.email.email import EmailActionType
@@ -22,6 +23,8 @@
 def get_email_template(action_type: EmailActionType) -> Template:
     if action_type == EmailActionType.SUBJECT_IDENTITY_VERIFICATION:
         return template_env.get_template(SUBJECT_IDENTITY_VERIFICATION_TEMPLATE)
+    if action_type == EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT:
+        return template_env.get_template(EMAIL_ERASURE_REQUEST_FULFILLMENT)
 
     logger.error("No corresponding template linked to the %s", action_type)
     raise EmailTemplateUnhandledActionType(
diff --git a/src/fidesops/ops/email_templates/template_names.py b/src/fidesops/ops/email_templates/template_names.py
index e674b46cf..63e74eace 100644
--- a/src/fidesops/ops/email_templates/template_names.py
+++ b/src/fidesops/ops/email_templates/template_names.py
@@ -1 +1,2 @@
 SUBJECT_IDENTITY_VERIFICATION_TEMPLATE = "subject_identity_verification.html"
+EMAIL_ERASURE_REQUEST_FULFILLMENT = "erasure_request_email_fulfillment.html"
diff --git a/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html b/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html
new file mode 100644
index 000000000..667f88a0a
--- /dev/null
+++ b/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html
@@ -0,0 +1,30 @@
+<!DOCTYPE html>
+<html lang="en">
+   <head>
+      <meta charset="UTF-8">
+      <title>Erasure request</title>
+   </head>
+   <body>
+      <main>
+         <p>Please locate and erase the associated records from the following collections:</p>
+         {% for collection, action_required in dataset_collection_action_required.items() -%}
+         <p><b>{{ collection }}</b></p>
+         {% for action in action_required.action_needed -%}
+         <p>Locate the relevant records with:</p>
+         <ul>
+            {% for field, values in action.locators.items() -%}
+            <li> Field: <i>{{ field }}</i>, Values: {{ values|join(', ') }} </li>
+            {%- endfor %}
+         </ul>
+         {% if action.update -%}
+         <p>Erase the following fields:</p>
+         <ul>
+            {% for field_name, masking_strategy in action.update.items() -%}
+            <li><i>{{field_name}}</i></li>
+            {%- endfor %}
+         </ul>
+         {%- endif %}
+          {%- endfor %} {%- endfor %}
+      </main>
+   </body>
+</html>
\ No newline at end of file
diff --git a/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py b/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py
new file mode 100644
index 000000000..7d8745ce1
--- /dev/null
+++ b/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py
@@ -0,0 +1,32 @@
+"""audit log email send
+
+Revision ID: 912d801f06c0
+Revises: c2f7a29c4780
+Create Date: 2022-09-01 16:23:10.905356
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "912d801f06c0"
+down_revision = "c2f7a29c4780"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute("alter type auditlogaction add value 'email_sent'")
+
+
+def downgrade():
+    op.execute("delete from auditlog where action in ('email_sent')")
+    op.execute("alter type auditlogaction rename to auditlogaction_old")
+    op.execute("create type auditlogaction as enum('approved', 'denied', 'finished')")
+    op.execute(
+        (
+            "alter table auditlog alter column action type auditlogaction using "
+            "action::text::auditlogaction"
+        )
+    )
+    op.execute("drop type auditlogaction_old")
diff --git a/src/fidesops/ops/schemas/email/email.py b/src/fidesops/ops/schemas/email/email.py
index 827a7ccab..f77742acb 100644
--- a/src/fidesops/ops/schemas/email/email.py
+++ b/src/fidesops/ops/schemas/email/email.py
@@ -19,6 +19,7 @@ class EmailActionType(Enum):
 
     # verify email upon acct creation
     SUBJECT_IDENTITY_VERIFICATION = "subject_identity_verification"
+    EMAIL_ERASURE_REQUEST_FULFILLMENT = "email_erasure_fulfillment"
 
 
 class EmailTemplateBodyParams(Enum):
diff --git a/src/fidesops/ops/service/connectors/email_connector.py b/src/fidesops/ops/service/connectors/email_connector.py
index 80d2c34b4..49eaa4f83 100644
--- a/src/fidesops/ops/service/connectors/email_connector.py
+++ b/src/fidesops/ops/service/connectors/email_connector.py
@@ -1,14 +1,28 @@
 import logging
 from typing import Any, Dict, List, Optional
 
-from fidesops.ops.graph.config import FieldPath
+from sqlalchemy.orm import Session
+
+from fidesops.ops.common_exceptions import (
+    EmailDispatchException,
+    PrivacyRequestErasureEmailSendRequired,
+)
+from fidesops.ops.graph.config import CollectionAddress, FieldPath
 from fidesops.ops.graph.traversal import TraversalNode
 from fidesops.ops.models.connectionconfig import ConnectionTestStatus
 from fidesops.ops.models.policy import CurrentStep, Policy, Rule
-from fidesops.ops.models.privacy_request import ManualAction, PrivacyRequest
+from fidesops.ops.models.privacy_request import (
+    CollectionActionRequired,
+    ManualAction,
+    PrivacyRequest,
+)
+from fidesops.ops.schemas.connection_configuration import EmailSchema
+from fidesops.ops.schemas.email.email import EmailActionType
 from fidesops.ops.service.connectors.base_connector import BaseConnector
 from fidesops.ops.service.connectors.query_config import ManualQueryConfig
+from fidesops.ops.service.email.email_dispatch_service import dispatch_email
 from fidesops.ops.util.collection_util import Row, append
+from fidesops.ops.util.logger import Pii
 
 logger = logging.getLogger(__name__)
 
@@ -28,9 +42,38 @@ def close(self) -> None:
 
     def test_connection(self) -> Optional[ConnectionTestStatus]:
         """
-        Override to skip connection test for now
+        Sends an email to the "test_email" configured, just to establish that the email workflow is working.
         """
-        return ConnectionTestStatus.skipped
+        config = EmailSchema(**self.configuration.secrets or {})
+        logger.info("Starting test connection to %s", self.configuration.key)
+
+        db = Session.object_session(self.configuration)
+
+        try:
+            dispatch_email(
+                db=db,
+                action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
+                to_email=config.test_email,
+                email_body_params={
+                    "test_collection": CollectionActionRequired(
+                        step=CurrentStep.erasure,
+                        collection=CollectionAddress("test_dataset", "test_collection"),
+                        action_needed=[
+                            ManualAction(
+                                locators={"id": ["example_id"]},
+                                get=None,
+                                update={
+                                    "test_field": "null_rewrite",
+                                },
+                            )
+                        ],
+                    )
+                },
+            )
+        except EmailDispatchException as exc:
+            logger.info("Email connector test failed with exception %s", Pii(exc))
+            return ConnectionTestStatus.failed
+        return ConnectionTestStatus.succeeded
 
     def retrieve_data(  # type: ignore
         self,
@@ -40,6 +83,10 @@ def retrieve_data(  # type: ignore
         input_data: Dict[str, List[Any]],
     ) -> Optional[List[Row]]:
         """Access requests are not supported at this time."""
+        logger.info(
+            "Access request not supported for email connector `%s` at this time.",
+            node.address.value,
+        )
         return []
 
     def mask_data(  # type: ignore
@@ -49,7 +96,7 @@ def mask_data(  # type: ignore
         privacy_request: PrivacyRequest,
         rows: List[Row],
         input_data: Dict[str, List[Any]],
-    ) -> Optional[int]:
+    ) -> None:
         """Cache instructions for how to mask data in this collection.
         One email will be sent for all collections in this dataset at the end of the privacy request execution.
         """
@@ -65,7 +112,7 @@ def mask_data(  # type: ignore
             action_needed=[manual_action],
         )
 
-        return 0  # Fidesops itself does not mask this collection.
+        raise PrivacyRequestErasureEmailSendRequired("email prepared")
 
     def build_masking_instructions(
         self, node: TraversalNode, policy: Policy, input_data: Dict[str, List[Any]]
diff --git a/src/fidesops/ops/service/email/email_dispatch_service.py b/src/fidesops/ops/service/email/email_dispatch_service.py
index b3ee0550c..06c93fc1d 100644
--- a/src/fidesops/ops/service/email/email_dispatch_service.py
+++ b/src/fidesops/ops/service/email/email_dispatch_service.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Optional, Union
+from typing import Any, Dict, Optional, Union
 
 import requests
 from requests import Response
@@ -8,6 +8,7 @@
 from fidesops.ops.common_exceptions import EmailDispatchException
 from fidesops.ops.email_templates import get_email_template
 from fidesops.ops.models.email import EmailConfig
+from fidesops.ops.models.privacy_request import CollectionActionRequired
 from fidesops.ops.schemas.email.email import (
     EmailActionType,
     EmailForActionType,
@@ -25,7 +26,10 @@ def dispatch_email(
     db: Session,
     action_type: EmailActionType,
     to_email: Optional[str],
-    email_body_params: Union[SubjectIdentityVerificationBodyParams],
+    email_body_params: Union[
+        SubjectIdentityVerificationBodyParams,
+        Dict[str, Optional[CollectionActionRequired]],
+    ],
 ) -> None:
     if not to_email:
         raise EmailDispatchException("No email supplied.")
@@ -57,7 +61,7 @@ def dispatch_email(
 
 def _build_email(
     action_type: EmailActionType,
-    body_params: Union[SubjectIdentityVerificationBodyParams],
+    body_params: Any,
 ) -> EmailForActionType:
     if action_type == EmailActionType.SUBJECT_IDENTITY_VERIFICATION:
         template = get_email_template(action_type)
@@ -70,6 +74,14 @@ def _build_email(
                 }
             ),
         )
+    if action_type == EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT:
+        base_template = get_email_template(action_type)
+        return EmailForActionType(
+            subject="Data erasure request",
+            body=base_template.render(
+                {"dataset_collection_action_required": body_params}
+            ),
+        )
     logger.error("Email action type %s is not implemented", action_type)
     raise EmailDispatchException(f"Email action type {action_type} is not implemented")
 
diff --git a/src/fidesops/ops/service/privacy_request/request_runner_service.py b/src/fidesops/ops/service/privacy_request/request_runner_service.py
index 3b7f95b57..ad2e4b55d 100644
--- a/src/fidesops/ops/service/privacy_request/request_runner_service.py
+++ b/src/fidesops/ops/service/privacy_request/request_runner_service.py
@@ -22,7 +22,7 @@
     fideslog_graph_failure,
 )
 from fidesops.ops.graph.graph import DatasetGraph
-from fidesops.ops.models.connectionconfig import ConnectionConfig
+from fidesops.ops.models.connectionconfig import ConnectionConfig, ConnectionType
 from fidesops.ops.models.datasetconfig import DatasetConfig
 from fidesops.ops.models.policy import (
     ActionType,
@@ -32,7 +32,13 @@
     PolicyPreWebhook,
     WebhookTypes,
 )
-from fidesops.ops.models.privacy_request import PrivacyRequest, PrivacyRequestStatus
+from fidesops.ops.models.privacy_request import (
+    CollectionActionRequired,
+    PrivacyRequest,
+    PrivacyRequestStatus,
+)
+from fidesops.ops.schemas.email.email import EmailActionType
+from fidesops.ops.service.email.email_dispatch_service import dispatch_email
 from fidesops.ops.service.storage.storage_uploader_service import upload
 from fidesops.ops.task.filter_results import filter_data_categories
 from fidesops.ops.task.graph_task import (
@@ -298,6 +304,8 @@ async def run_privacy_request(
             _log_exception(exc, config.dev_mode)
             return
 
+        email_connector_send(db=session, privacy_request=privacy_request)
+
         # Run post-execution webhooks
         proceed = run_webhooks_and_report_status(
             db=session,
@@ -360,3 +368,37 @@ def generate_id_verification_code() -> str:
     Generate one-time identity verification code
     """
     return str(random.choice(range(100000, 999999)))
+
+
+def email_connector_send(db: Session, privacy_request: PrivacyRequest) -> None:
+    """
+    Send emails to configured third-parties with instructions on how to erase remaining data.
+    Combined all the collections on each email-based dataset into one email.
+    """
+    email_dataset_configs = db.query(DatasetConfig, ConnectionConfig).filter(
+        DatasetConfig.connection_config_id == ConnectionConfig.id,
+        ConnectionConfig.connection_type == ConnectionType.email,
+    )
+    for ds, cc in email_dataset_configs:
+        template_values: Dict[
+            str, Optional[CollectionActionRequired]
+        ] = privacy_request.get_email_connector_template_contents_by_dataset(
+            CurrentStep.erasure, ds.dataset.get("fides_key")
+        )
+
+        dispatch_email(
+            db,
+            action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
+            to_email=cc.secrets.get("to_email"),
+            email_body_params=template_values,
+        )
+
+        AuditLog.create(
+            db=db,
+            data={
+                "user_id": "system",
+                "privacy_request_id": privacy_request.id,
+                "action": AuditLogAction.email_sent,
+                "message": f"Erasure email instructions dispatched for '{ds.dataset.get('fides_key')}'",
+            },
+        )
diff --git a/src/fidesops/ops/task/graph_task.py b/src/fidesops/ops/task/graph_task.py
index 93f637db4..f9357eb40 100644
--- a/src/fidesops/ops/task/graph_task.py
+++ b/src/fidesops/ops/task/graph_task.py
@@ -10,7 +10,11 @@
 from dask.threaded import get
 from sqlalchemy.orm import Session
 
-from fidesops.ops.common_exceptions import CollectionDisabled, PrivacyRequestPaused
+from fidesops.ops.common_exceptions import (
+    CollectionDisabled,
+    PrivacyRequestErasureEmailSendRequired,
+    PrivacyRequestPaused,
+)
 from fidesops.ops.core.config import config
 from fidesops.ops.graph.analytics_events import (
     fideslog_graph_rerun,
@@ -63,7 +67,7 @@ def retry(
 
     def decorator(func: Callable) -> Callable:
         @wraps(func)
-        def result(*args: Any, **kwargs: Any) -> List[Optional[Row]]:
+        def result(*args: Any, **kwargs: Any) -> Any:
             func_delay = config.execution.task_retry_delay
             method_name = func.__name__
             self = args[0]
@@ -88,6 +92,12 @@ def result(*args: Any, **kwargs: Any) -> List[Optional[Row]]:
                     self.log_paused(action_type, ex)
                     # Re-raise to stop privacy request execution on pause.
                     raise
+                except PrivacyRequestErasureEmailSendRequired as exc:
+                    self.log_end(action_type, ex=None, success_override_msg=exc)
+                    self.resources.cache_erasure(
+                        f"{self.traversal_node.address.value}", 0
+                    )  # Cache that the erasure was performed in case we need to restart
+                    return 0
                 except CollectionDisabled as exc:
                     logger.warning(
                         "Skipping disabled collection %s for privacy_request: %s",
@@ -363,7 +373,10 @@ def log_skipped(self, action_type: ActionType, ex: str) -> None:
         self.update_status(str(ex), [], action_type, ExecutionLogStatus.skipped)
 
     def log_end(
-        self, action_type: ActionType, ex: Optional[BaseException] = None
+        self,
+        action_type: ActionType,
+        ex: Optional[BaseException] = None,
+        success_override_msg: Optional[BaseException] = None,
     ) -> None:
         """On completion activities"""
         if ex:
@@ -378,7 +391,7 @@ def log_end(
         else:
             logger.info("Ending %s, %s", self.resources.request.id, self.key)
             self.update_status(
-                "success",
+                str(success_override_msg) if success_override_msg else "success",
                 build_affected_field_logs(
                     self.traversal_node.node, self.resources.policy, action_type
                 ),
diff --git a/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py b/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
index 28c300575..882f940cf 100644
--- a/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
@@ -18,7 +18,11 @@
     STORAGE_DELETE,
 )
 from fidesops.ops.api.v1.urn_registry import CONNECTIONS, SAAS_CONFIG, V1_URL_PREFIX
+from fidesops.ops.graph.config import CollectionAddress
 from fidesops.ops.models.connectionconfig import ConnectionConfig
+from fidesops.ops.models.policy import CurrentStep
+from fidesops.ops.models.privacy_request import CollectionActionRequired, ManualAction
+from fidesops.ops.schemas.email.email import EmailActionType
 
 page_size = Params().size
 
@@ -1148,8 +1152,10 @@ def test_put_saas_example_connection_config_secrets_missing_saas_config(
             == f"A SaaS config to validate the secrets is unavailable for this connection config, please add one via {SAAS_CONFIG}"
         )
 
+    @mock.patch("fidesops.ops.service.connectors.email_connector.dispatch_email")
     def test_put_email_connection_config_secrets(
         self,
+        mock_dispatch_email,
         api_client: TestClient,
         db: Session,
         generate_auth_header,
@@ -1157,7 +1163,11 @@ def test_put_email_connection_config_secrets(
         url,
     ) -> None:
         auth_header = generate_auth_header(scopes=[CONNECTION_CREATE_OR_UPDATE])
-        payload = {"url": None, "to_email": "test@example.com"}
+        payload = {
+            "url": None,
+            "to_email": "test1@example.com",
+            "test_email": "test@example.com",
+        }
         url = f"{V1_URL_PREFIX}{CONNECTIONS}/{email_connection_config.key}/secret"
 
         resp = api_client.put(
@@ -1172,12 +1182,32 @@ def test_put_email_connection_config_secrets(
             body["msg"]
             == f"Secrets updated for ConnectionConfig with key: {email_connection_config.key}."
         )
-        assert body["test_status"] == "skipped" ""
+        assert body["test_status"] == "succeeded"
         db.refresh(email_connection_config)
         assert email_connection_config.secrets == {
-            "to_email": "test@example.com",
+            "to_email": "test1@example.com",
             "url": None,
-            "test_email": None,
+            "test_email": "test@example.com",
+        }
+        assert email_connection_config.last_test_timestamp is not None
+        assert email_connection_config.last_test_succeeded is not None
+
+        assert mock_dispatch_email.called
+        kwargs = mock_dispatch_email.call_args.kwargs
+        assert (
+            kwargs["action_type"] == EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT
+        )
+        assert kwargs["to_email"] == "test@example.com"
+        assert kwargs["email_body_params"] == {
+            "test_collection": CollectionActionRequired(
+                step=CurrentStep.erasure,
+                collection=CollectionAddress("test_dataset", "test_collection"),
+                action_needed=[
+                    ManualAction(
+                        locators={"id": ["example_id"]},
+                        get=None,
+                        update={"test_field": "null_rewrite"},
+                    )
+                ],
+            )
         }
-        assert email_connection_config.last_test_timestamp is None
-        assert email_connection_config.last_test_succeeded is None
diff --git a/tests/ops/fixtures/email_fixtures.py b/tests/ops/fixtures/email_fixtures.py
index 0117cd8ac..b9edb7737 100644
--- a/tests/ops/fixtures/email_fixtures.py
+++ b/tests/ops/fixtures/email_fixtures.py
@@ -22,6 +22,7 @@ def email_connection_config(db: Session) -> Generator:
             "key": "my_email_connection_config",
             "connection_type": ConnectionType.email,
             "access": AccessLevel.write,
+            "secrets": {"to_email": "test@example.com"},
         },
     )
     yield connection_config
diff --git a/tests/ops/integration_tests/test_integration_email.py b/tests/ops/integration_tests/test_integration_email.py
index aab6a244d..d61d77ba0 100644
--- a/tests/ops/integration_tests/test_integration_email.py
+++ b/tests/ops/integration_tests/test_integration_email.py
@@ -1,26 +1,47 @@
+from unittest import mock
+
 import pytest as pytest
+from fideslib.models.audit_log import AuditLog, AuditLogAction
 
 from fidesops.ops.graph.config import CollectionAddress
 from fidesops.ops.graph.graph import DatasetGraph
 from fidesops.ops.models.datasetconfig import convert_dataset_to_graph
 from fidesops.ops.models.policy import CurrentStep
-from fidesops.ops.models.privacy_request import CollectionActionRequired, ManualAction
+from fidesops.ops.models.privacy_request import (
+    CollectionActionRequired,
+    ExecutionLog,
+    ExecutionLogStatus,
+    ManualAction,
+)
 from fidesops.ops.schemas.dataset import FidesopsDataset
+from fidesops.ops.schemas.email.email import EmailActionType
+from fidesops.ops.service.privacy_request.request_runner_service import (
+    email_connector_send,
+)
 from fidesops.ops.task import graph_task
 
 
 @pytest.mark.integration_postgres
 @pytest.mark.integration
+@mock.patch(
+    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
+)
+@pytest.mark.asyncio
 async def test_collections_with_manual_erasure_confirmation(
+    mock_email_dispatch,
     db,
     erasure_policy,
     integration_postgres_config,
     email_connection_config,
     privacy_request,
     example_datasets,
+    email_dataset_config,
+    email_config,
 ) -> None:
     """Run an erasure privacy request with a postgres dataset and an email dataset.
     The email dataset has three separate collections.
+
+    Call the email send and verify what would have been emailed
     """
     privacy_request.policy = erasure_policy
     rule = erasure_policy.rules[0]
@@ -139,3 +160,34 @@ async def test_collections_with_manual_erasure_confirmation(
             ],
         ),
     }, "Only two collections need masking, but all are included in case they include relevant data locators."
+
+    children_logs = db.query(ExecutionLog).filter(
+        ExecutionLog.privacy_request_id == privacy_request.id,
+        ExecutionLog.dataset_name == email_dataset_config.fides_key,
+        ExecutionLog.collection_name == "children",
+    )
+    assert {"starting", "email prepared"} == {
+        log.message for log in children_logs
+    }, "Execution Log given unique message"
+    assert {ExecutionLogStatus.in_processing, ExecutionLogStatus.complete} == {
+        log.status for log in children_logs
+    }
+
+    email_connector_send(db, privacy_request)
+    assert mock_email_dispatch.called
+    call_args = mock_email_dispatch.call_args[1]
+    assert call_args["action_type"] == EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT
+    assert call_args["to_email"] == "test@example.com"
+    assert call_args["email_body_params"] == raw_email_template_values
+
+    created_email_audit_log = (
+        db.query(AuditLog)
+        .filter(AuditLog.privacy_request_id == privacy_request.id)
+        .all()[0]
+    )
+    assert (
+        created_email_audit_log.message
+        == "Erasure email instructions dispatched for 'email_dataset'"
+    )
+    assert created_email_audit_log.user_id == "system"
+    assert created_email_audit_log.action == AuditLogAction.email_sent
diff --git a/tests/ops/service/privacy_request/request_runner_service_test.py b/tests/ops/service/privacy_request/request_runner_service_test.py
index 782c6d749..792e2efb5 100644
--- a/tests/ops/service/privacy_request/request_runner_service_test.py
+++ b/tests/ops/service/privacy_request/request_runner_service_test.py
@@ -16,6 +16,7 @@
     PrivacyRequestPaused,
 )
 from fidesops.ops.core.config import config
+from fidesops.ops.models.email import EmailConfig
 from fidesops.ops.models.policy import CurrentStep, PolicyPostWebhook
 from fidesops.ops.models.privacy_request import (
     ActionType,
@@ -24,6 +25,7 @@
     PrivacyRequest,
     PrivacyRequestStatus,
 )
+from fidesops.ops.schemas.email.email import EmailForActionType
 from fidesops.ops.schemas.external_https import SecondPartyResponseFormat
 from fidesops.ops.schemas.masking.masking_configuration import (
     HmacMaskingConfiguration,
@@ -1550,3 +1552,44 @@ def test_privacy_request_log_failure(
         assert sent_event.status_code == 500
         assert sent_event.error == "KeyError"
         assert sent_event.extra_data == {"privacy_request": pr.id}
+
+
+@mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+@pytest.mark.integration
+def test_create_and_process_erasure_request_email_connector(
+    mailgun_send,
+    email_connection_config,
+    erasure_policy,
+    integration_postgres_config,
+    run_privacy_request_task,
+    email_dataset_config,
+    postgres_example_test_dataset_config_read_access,
+    email_config,
+    db,
+):
+    """
+    Asserts that mailgun was called and verifies email template renders without error
+    """
+    rule = erasure_policy.rules[0]
+    target = rule.targets[0]
+    target.data_category = "user.childrens"
+    target.save(db=db)
+
+    email = "customer-1@example.com"
+    data = {
+        "requested_at": "2021-08-30T16:09:37.359Z",
+        "policy_key": erasure_policy.key,
+        "identity": {"email": email},
+    }
+
+    pr = get_privacy_request_results(
+        db,
+        erasure_policy,
+        run_privacy_request_task,
+        data,
+    )
+    pr.delete(db=db)
+    assert mailgun_send.called
+    kwargs = mailgun_send.call_args.kwargs
+    assert type(kwargs["email_config"]) == EmailConfig
+    assert type(kwargs["email"]) == EmailForActionType

From e981ea4e6caf849c14fa6a5161d253175f217676 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Thu, 1 Sep 2022 17:17:33 -0500
Subject: [PATCH 02/12] Add more "checkpoints" to privacy request execution -
 these are locations from which we can resume privacy request execution
 without having to run from the beginning.

- Add more options to CurrentStep Enum
- Cache the checkpoint if an email send fails, so we can retry from the same step.
---
 .../v1/endpoints/privacy_request_endpoints.py | 45 ++++++++------
 src/fidesops/ops/models/policy.py             |  3 +
 src/fidesops/ops/models/privacy_request.py    | 61 +++++++++++++------
 src/fidesops/ops/schemas/privacy_request.py   |  6 +-
 .../ops/service/connectors/email_connector.py |  4 +-
 .../service/email/email_dispatch_service.py   |  4 +-
 .../privacy_request/request_runner_service.py | 53 ++++++++++++----
 src/fidesops/ops/task/graph_task.py           |  2 +-
 .../test_connection_config_endpoints.py       |  4 +-
 .../test_privacy_request_endpoints.py         | 51 ++++++++++++----
 tests/ops/integration_tests/test_execution.py |  4 +-
 .../test_integration_email.py                 | 10 +--
 tests/ops/models/test_privacy_request.py      | 50 +++++++++++++--
 .../request_runner_service_test.py            | 55 +++++++++++++++++
 14 files changed, 268 insertions(+), 84 deletions(-)

diff --git a/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py b/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py
index bf8839243..d6e9e3597 100644
--- a/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py
+++ b/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py
@@ -79,7 +79,7 @@
 from fidesops.ops.schemas.privacy_request import (
     BulkPostPrivacyRequests,
     BulkReviewResponse,
-    CollectionActionRequired,
+    CheckpointActionRequired,
     DenyPrivacyRequests,
     ExecutionLogDetailResponse,
     PrivacyRequestCreate,
@@ -492,16 +492,16 @@ def attach_resume_instructions(privacy_request: PrivacyRequest) -> None:
     about how to resume manually if applicable.
     """
     resume_endpoint: Optional[str] = None
-    stopped_collection_details: Optional[CollectionActionRequired] = None
+    action_required_details: Optional[CheckpointActionRequired] = None
 
     if privacy_request.status == PrivacyRequestStatus.paused:
-        stopped_collection_details = privacy_request.get_paused_collection_details()
+        action_required_details = privacy_request.get_paused_collection_details()
 
-        if stopped_collection_details:
+        if action_required_details:
             # Graph is paused on a specific collection
             resume_endpoint = (
                 PRIVACY_REQUEST_MANUAL_ERASURE
-                if stopped_collection_details.step == CurrentStep.erasure
+                if action_required_details.step == CurrentStep.erasure
                 else PRIVACY_REQUEST_MANUAL_INPUT
             )
         else:
@@ -509,16 +509,16 @@ def attach_resume_instructions(privacy_request: PrivacyRequest) -> None:
             resume_endpoint = PRIVACY_REQUEST_RESUME
 
     elif privacy_request.status == PrivacyRequestStatus.error:
-        stopped_collection_details = privacy_request.get_failed_collection_details()
+        action_required_details = privacy_request.get_failed_checkpoint_details()
         resume_endpoint = PRIVACY_REQUEST_RETRY
 
-    if stopped_collection_details:
-        stopped_collection_details.step = stopped_collection_details.step.value  # type: ignore
-        stopped_collection_details.collection = (
-            stopped_collection_details.collection.value  # type: ignore
+    if action_required_details:
+        action_required_details.step = action_required_details.step.value  # type: ignore
+        action_required_details.collection = (
+            action_required_details.collection.value  # type: ignore
         )
 
-    privacy_request.stopped_collection_details = stopped_collection_details
+    privacy_request.action_required_details = action_required_details
     # replaces the placeholder in the url with the privacy request id
     privacy_request.resume_endpoint = (
         resume_endpoint.format(privacy_request_id=privacy_request.id)
@@ -784,7 +784,10 @@ async def resume_privacy_request_with_manual_input(
     manual_rows: List[Row] = [],
     manual_count: Optional[int] = None,
 ) -> PrivacyRequest:
-    """Resume privacy request after validating and caching manual data for an access or an erasure request."""
+    """Resume privacy request after validating and caching manual data for an access or an erasure request.
+
+    This assumes the privacy request is being resumed from a specific collection in the graph.
+    """
     privacy_request: PrivacyRequest = get_privacy_request_or_error(
         db, privacy_request_id
     )
@@ -796,7 +799,7 @@ async def resume_privacy_request_with_manual_input(
         )
 
     paused_details: Optional[
-        CollectionActionRequired
+        CheckpointActionRequired
     ] = privacy_request.get_paused_collection_details()
     if not paused_details:
         raise HTTPException(
@@ -805,7 +808,7 @@ async def resume_privacy_request_with_manual_input(
         )
 
     paused_step: CurrentStep = paused_details.step
-    paused_collection: CollectionAddress = paused_details.collection
+    paused_collection: Optional[CollectionAddress] = paused_details.collection
 
     if paused_step != expected_paused_step:
         raise HTTPException(
@@ -818,6 +821,12 @@ async def resume_privacy_request_with_manual_input(
     dataset_graphs = [dataset_config.get_graph() for dataset_config in datasets]
     dataset_graph = DatasetGraph(*dataset_graphs)
 
+    if not paused_collection:
+        raise HTTPException(
+            status_code=HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Cannot save manual data on paused collection. No paused collection saved'.",
+        )
+
     node: Optional[Node] = dataset_graph.nodes.get(paused_collection)
     if not node:
         raise HTTPException(
@@ -939,8 +948,8 @@ async def restart_privacy_request_from_failure(
         )
 
     failed_details: Optional[
-        CollectionActionRequired
-    ] = privacy_request.get_failed_collection_details()
+        CheckpointActionRequired
+    ] = privacy_request.get_failed_checkpoint_details()
     if not failed_details:
         raise HTTPException(
             status_code=HTTP_400_BAD_REQUEST,
@@ -948,7 +957,7 @@ async def restart_privacy_request_from_failure(
         )
 
     failed_step: CurrentStep = failed_details.step
-    failed_collection: CollectionAddress = failed_details.collection
+    failed_collection: Optional[CollectionAddress] = failed_details.collection
 
     logger.info(
         "Restarting failed privacy request '%s' from '%s step, 'collection '%s'",
@@ -964,7 +973,7 @@ async def restart_privacy_request_from_failure(
         from_step=failed_step.value,
     )
 
-    privacy_request.cache_failed_collection_details()  # Reset failed step and collection to None
+    privacy_request.cache_failed_checkpoint_details()  # Reset failed step and collection to None
 
     return privacy_request
 
diff --git a/src/fidesops/ops/models/policy.py b/src/fidesops/ops/models/policy.py
index 7f7322e93..5bb40ee80 100644
--- a/src/fidesops/ops/models/policy.py
+++ b/src/fidesops/ops/models/policy.py
@@ -27,8 +27,11 @@
 
 
 class CurrentStep(EnumType):
+    pre_webhooks = "pre_webhooks"
     access = "access"
     erasure = "erasure"
+    erasure_email_post_send = "erasure_email_post_send"
+    post_webhooks = "post_webhooks"
 
 
 class ActionType(str, EnumType):
diff --git a/src/fidesops/ops/models/privacy_request.py b/src/fidesops/ops/models/privacy_request.py
index 29e2bbb5b..2555717b3 100644
--- a/src/fidesops/ops/models/privacy_request.py
+++ b/src/fidesops/ops/models/privacy_request.py
@@ -69,6 +69,15 @@
 logger = logging.getLogger(__name__)
 
 
+EXECUTION_CHECKPOINTS = [
+    CurrentStep.pre_webhooks,
+    CurrentStep.access,
+    CurrentStep.erasure,
+    CurrentStep.erasure_email_post_send,
+    CurrentStep.post_webhooks,
+]
+
+
 class ManualAction(BaseSchema):
     """Surface how to retrieve or mask data in a database-agnostic way
 
@@ -82,8 +91,8 @@ class ManualAction(BaseSchema):
     update: Optional[Dict[str, Any]]
 
 
-class CollectionActionRequired(BaseSchema):
-    """Describes actions needed on a given collection.
+class CheckpointActionRequired(BaseSchema):
+    """Describes actions needed on a particular checkpoint.
 
     Examples are a paused collection that needs manual input, a failed collection that
     needs to be restarted, or a collection where instructions need to be emailed to a third
@@ -91,7 +100,7 @@ class CollectionActionRequired(BaseSchema):
     """
 
     step: CurrentStep
-    collection: CollectionAddress
+    collection: Optional[CollectionAddress]
     action_needed: Optional[List[ManualAction]] = None
 
     class Config:
@@ -380,14 +389,14 @@ def cache_email_connector_template_contents(
 
     def get_email_connector_template_contents_by_dataset(
         self, step: CurrentStep, dataset: str
-    ) -> Dict[str, Optional[CollectionActionRequired]]:
+    ) -> Dict[str, Optional[CheckpointActionRequired]]:
         """Retrieve the raw details to populate an email template for collections on a given dataset."""
         cache: FidesopsRedis = get_cache()
         email_contents: Dict[str, Optional[Any]] = cache.get_encoded_objects_by_prefix(
             f"EMAIL_INFORMATION__{self.id}__{step.value}__{dataset}"
         )
         return {
-            k.split("__")[-1]: CollectionActionRequired.parse_obj(v) if v else None
+            k.split("__")[-1]: CheckpointActionRequired.parse_obj(v) if v else None
             for k, v in email_contents.items()
         }
 
@@ -409,7 +418,7 @@ def cache_paused_collection_details(
 
     def get_paused_collection_details(
         self,
-    ) -> Optional[CollectionActionRequired]:
+    ) -> Optional[CheckpointActionRequired]:
         """Return details about the paused step, paused collection, and any action needed to resume the paused privacy request.
 
         The paused step lets us know if we should resume privacy request execution from the "access" or the "erasure"
@@ -418,14 +427,16 @@ def get_paused_collection_details(
         """
         return get_action_required_details(cached_key=f"EN_PAUSED_LOCATION__{self.id}")
 
-    def cache_failed_collection_details(
+    def cache_failed_checkpoint_details(
         self,
         step: Optional[CurrentStep] = None,
         collection: Optional[CollectionAddress] = None,
     ) -> None:
         """
-        Cache details about the failed step and failed collection details. No specific input data is required to resume
-        a failed request, so action_needed is None.
+        Cache a checkpoint where the privacy request failed so we can later resume from this failure point.
+
+        Cache details about the failed step and failed collection details (where applicable).
+        No specific input data is required to resume a failed request, so action_needed is None.
         """
         cache_action_required(
             cache_key=f"FAILED_LOCATION__{self.id}",
@@ -434,9 +445,9 @@ def cache_failed_collection_details(
             action_needed=None,
         )
 
-    def get_failed_collection_details(
+    def get_failed_checkpoint_details(
         self,
-    ) -> Optional[CollectionActionRequired]:
+    ) -> Optional[CheckpointActionRequired]:
         """Get details about the failed step (access or erasure) and collection that triggered failure.
 
         The failed step lets us know if we should resume privacy request execution from the "access" or the "erasure"
@@ -704,21 +715,21 @@ def cache_action_required(
     The "step" describes whether action is needed in the access or the erasure portion of the request.
     """
     cache: FidesopsRedis = get_cache()
-    current_collection: Optional[CollectionActionRequired] = None
-    if collection and step:
-        current_collection = CollectionActionRequired(
+    action_required: Optional[CheckpointActionRequired] = None
+    if step:
+        action_required = CheckpointActionRequired(
             step=step, collection=collection, action_needed=action_needed
         )
 
     cache.set_encoded_object(
         cache_key,
-        current_collection.dict() if current_collection else None,
+        action_required.dict() if action_required else None,
     )
 
 
 def get_action_required_details(
     cached_key: str,
-) -> Optional[CollectionActionRequired]:
+) -> Optional[CheckpointActionRequired]:
     """Get details about the action required for a given collection.
 
     The "step" lets us know if action is needed in the "access" or the "erasure" portion of the privacy request flow.
@@ -726,11 +737,11 @@ def get_action_required_details(
     performed to complete the request.
     """
     cache: FidesopsRedis = get_cache()
-    cached_stopped: Optional[CollectionActionRequired] = cache.get_encoded_by_key(
+    cached_stopped: Optional[CheckpointActionRequired] = cache.get_encoded_by_key(
         cached_key
     )
     return (
-        CollectionActionRequired.parse_obj(cached_stopped) if cached_stopped else None
+        CheckpointActionRequired.parse_obj(cached_stopped) if cached_stopped else None
     )
 
 
@@ -778,3 +789,17 @@ class ExecutionLog(Base):
         nullable=False,
         index=True,
     )
+
+
+def can_run_checkpoint(
+    request_checkpoint: CurrentStep, from_checkpoint: Optional[CurrentStep] = None
+) -> bool:
+    """Determine whether we should run a specific checkpoint in privacy request execution
+
+    If there's no from_checkpoint specified we should always run the current checkpoint.
+    """
+    if not from_checkpoint:
+        return True
+    return EXECUTION_CHECKPOINTS.index(
+        request_checkpoint
+    ) >= EXECUTION_CHECKPOINTS.index(from_checkpoint)
diff --git a/src/fidesops/ops/schemas/privacy_request.py b/src/fidesops/ops/schemas/privacy_request.py
index eb9b4c76b..65955e059 100644
--- a/src/fidesops/ops/schemas/privacy_request.py
+++ b/src/fidesops/ops/schemas/privacy_request.py
@@ -9,7 +9,7 @@
 from fidesops.ops.core.config import config
 from fidesops.ops.models.policy import ActionType
 from fidesops.ops.models.privacy_request import (
-    CollectionActionRequired,
+    CheckpointActionRequired,
     ExecutionLogStatus,
     PrivacyRequestStatus,
 )
@@ -133,7 +133,7 @@ class RowCountRequest(BaseSchema):
     row_count: int
 
 
-class CollectionActionRequiredDetails(CollectionActionRequired):
+class CheckpointActionRequiredDetails(CheckpointActionRequired):
 
     collection: Optional[str] = None  # type: ignore
 
@@ -164,7 +164,7 @@ class PrivacyRequestResponse(BaseSchema):
     # creation.
     identity: Optional[Dict[str, Optional[str]]]
     policy: PolicySchema
-    stopped_collection_details: Optional[CollectionActionRequiredDetails] = None
+    action_required_details: Optional[CheckpointActionRequiredDetails] = None
     resume_endpoint: Optional[str]
 
     class Config:
diff --git a/src/fidesops/ops/service/connectors/email_connector.py b/src/fidesops/ops/service/connectors/email_connector.py
index 49eaa4f83..beb1a78c2 100644
--- a/src/fidesops/ops/service/connectors/email_connector.py
+++ b/src/fidesops/ops/service/connectors/email_connector.py
@@ -12,7 +12,7 @@
 from fidesops.ops.models.connectionconfig import ConnectionTestStatus
 from fidesops.ops.models.policy import CurrentStep, Policy, Rule
 from fidesops.ops.models.privacy_request import (
-    CollectionActionRequired,
+    CheckpointActionRequired,
     ManualAction,
     PrivacyRequest,
 )
@@ -55,7 +55,7 @@ def test_connection(self) -> Optional[ConnectionTestStatus]:
                 action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
                 to_email=config.test_email,
                 email_body_params={
-                    "test_collection": CollectionActionRequired(
+                    "test_collection": CheckpointActionRequired(
                         step=CurrentStep.erasure,
                         collection=CollectionAddress("test_dataset", "test_collection"),
                         action_needed=[
diff --git a/src/fidesops/ops/service/email/email_dispatch_service.py b/src/fidesops/ops/service/email/email_dispatch_service.py
index 06c93fc1d..b098aa0dc 100644
--- a/src/fidesops/ops/service/email/email_dispatch_service.py
+++ b/src/fidesops/ops/service/email/email_dispatch_service.py
@@ -8,7 +8,7 @@
 from fidesops.ops.common_exceptions import EmailDispatchException
 from fidesops.ops.email_templates import get_email_template
 from fidesops.ops.models.email import EmailConfig
-from fidesops.ops.models.privacy_request import CollectionActionRequired
+from fidesops.ops.models.privacy_request import CheckpointActionRequired
 from fidesops.ops.schemas.email.email import (
     EmailActionType,
     EmailForActionType,
@@ -28,7 +28,7 @@ def dispatch_email(
     to_email: Optional[str],
     email_body_params: Union[
         SubjectIdentityVerificationBodyParams,
-        Dict[str, Optional[CollectionActionRequired]],
+        Dict[str, Optional[CheckpointActionRequired]],
     ],
 ) -> None:
     if not to_email:
diff --git a/src/fidesops/ops/service/privacy_request/request_runner_service.py b/src/fidesops/ops/service/privacy_request/request_runner_service.py
index ad2e4b55d..891115551 100644
--- a/src/fidesops/ops/service/privacy_request/request_runner_service.py
+++ b/src/fidesops/ops/service/privacy_request/request_runner_service.py
@@ -14,6 +14,7 @@
 from fidesops.ops import common_exceptions
 from fidesops.ops.common_exceptions import (
     ClientUnsuccessfulException,
+    EmailDispatchException,
     PrivacyRequestPaused,
 )
 from fidesops.ops.core.config import config
@@ -33,9 +34,10 @@
     WebhookTypes,
 )
 from fidesops.ops.models.privacy_request import (
-    CollectionActionRequired,
+    CheckpointActionRequired,
     PrivacyRequest,
     PrivacyRequestStatus,
+    can_run_checkpoint,
 )
 from fidesops.ops.schemas.email.email import EmailActionType
 from fidesops.ops.service.email.email_dispatch_service import dispatch_email
@@ -80,6 +82,8 @@ def run_webhooks_and_report_status(
             webhook_cls.order > pre_webhook.order,
         )
 
+    current_step = CurrentStep[f"{webhook_cls.prefix}_webhooks"]
+
     for webhook in webhooks.order_by(webhook_cls.order):
         try:
             privacy_request.trigger_policy_webhook(webhook)
@@ -100,6 +104,7 @@ def run_webhooks_and_report_status(
                 Pii(str(exc.args[0])),
             )
             privacy_request.error_processing(db)
+            privacy_request.cache_failed_checkpoint_details(current_step)
             return False
         except ValidationError:
             logging.error(
@@ -108,6 +113,7 @@ def run_webhooks_and_report_status(
                 webhook.key,
             )
             privacy_request.error_processing(db)
+            privacy_request.cache_failed_checkpoint_details(current_step)
             return False
 
     return True
@@ -214,10 +220,7 @@ async def run_privacy_request(
     Celery does not like for the function to be async so the @sync decorator runs the
     coroutine for it.
     """
-    if from_step is not None:
-        # Re-cast `from_step` into an Enum to enforce the validation since unserializable objects
-        # can't be passed into and between tasks
-        from_step = CurrentStep(from_step)  # type: ignore
+    resume_step: Optional[CurrentStep] = CurrentStep(from_step) if from_step else None  # type: ignore
 
     with self.session as session:
 
@@ -230,7 +233,9 @@ async def run_privacy_request(
         logging.info("Dispatching privacy request %s", privacy_request.id)
         privacy_request.start_processing(session)
 
-        if not from_step:  # Skip if we're resuming from the access or erasure step.
+        if can_run_checkpoint(
+            request_checkpoint=CurrentStep.pre_webhooks, from_checkpoint=resume_step
+        ):
             # Run pre-execution webhooks
             proceed = run_webhooks_and_report_status(
                 session,
@@ -256,9 +261,9 @@ async def run_privacy_request(
             identity_data = privacy_request.get_cached_identity_data()
             connection_configs = ConnectionConfig.all(db=session)
 
-            if (
-                from_step != CurrentStep.erasure
-            ):  # Skip if we're resuming from erasure step
+            if can_run_checkpoint(
+                request_checkpoint=CurrentStep.access, from_checkpoint=resume_step
+            ):
                 access_result: Dict[str, List[Row]] = await run_access_request(
                     privacy_request=privacy_request,
                     policy=policy,
@@ -276,7 +281,11 @@ async def run_privacy_request(
                     privacy_request,
                 )
 
-            if policy.get_rules_for_action(action_type=ActionType.erasure):
+            if policy.get_rules_for_action(
+                action_type=ActionType.erasure
+            ) and can_run_checkpoint(
+                request_checkpoint=CurrentStep.erasure, from_checkpoint=resume_step
+            ):
                 # We only need to run the erasure once until masking strategies are handled
                 await run_erasure(
                     privacy_request=privacy_request,
@@ -297,14 +306,32 @@ async def run_privacy_request(
 
         except BaseException as exc:  # pylint: disable=broad-except
             privacy_request.error_processing(db=session)
-            # If dev mode, log traceback
+            # Send analytics to Fideslog
             await fideslog_graph_failure(
                 failed_graph_analytics_event(privacy_request, exc)
             )
+            # If dev mode, log traceback
             _log_exception(exc, config.dev_mode)
             return
 
-        email_connector_send(db=session, privacy_request=privacy_request)
+        # Send erasure requests via email to third parties where applicable
+        if can_run_checkpoint(
+            request_checkpoint=CurrentStep.erasure_email_post_send,
+            from_checkpoint=resume_step,
+        ):
+            try:
+                email_connector_send(db=session, privacy_request=privacy_request)
+            except EmailDispatchException as exc:
+                privacy_request.cache_failed_checkpoint_details(
+                    step=CurrentStep.erasure_email_post_send, collection=None
+                )
+                privacy_request.error_processing(db=session)
+                await fideslog_graph_failure(
+                    failed_graph_analytics_event(privacy_request, exc)
+                )
+                # If dev mode, log traceback
+                _log_exception(exc, config.dev_mode)
+                return
 
         # Run post-execution webhooks
         proceed = run_webhooks_and_report_status(
@@ -381,7 +408,7 @@ def email_connector_send(db: Session, privacy_request: PrivacyRequest) -> None:
     )
     for ds, cc in email_dataset_configs:
         template_values: Dict[
-            str, Optional[CollectionActionRequired]
+            str, Optional[CheckpointActionRequired]
         ] = privacy_request.get_email_connector_template_contents_by_dataset(
             CurrentStep.erasure, ds.dataset.get("fides_key")
         )
diff --git a/src/fidesops/ops/task/graph_task.py b/src/fidesops/ops/task/graph_task.py
index f9357eb40..8ec8054b5 100644
--- a/src/fidesops/ops/task/graph_task.py
+++ b/src/fidesops/ops/task/graph_task.py
@@ -118,7 +118,7 @@ def result(*args: Any, **kwargs: Any) -> Any:
                     raised_ex = ex
 
             self.log_end(action_type, raised_ex)
-            self.resources.request.cache_failed_collection_details(
+            self.resources.request.cache_failed_checkpoint_details(
                 step=action_type, collection=self.traversal_node.address
             )
             # Re-raise to stop privacy request execution on failure.
diff --git a/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py b/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
index 882f940cf..e7e17ff40 100644
--- a/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
@@ -21,7 +21,7 @@
 from fidesops.ops.graph.config import CollectionAddress
 from fidesops.ops.models.connectionconfig import ConnectionConfig
 from fidesops.ops.models.policy import CurrentStep
-from fidesops.ops.models.privacy_request import CollectionActionRequired, ManualAction
+from fidesops.ops.models.privacy_request import CheckpointActionRequired, ManualAction
 from fidesops.ops.schemas.email.email import EmailActionType
 
 page_size = Params().size
@@ -1199,7 +1199,7 @@ def test_put_email_connection_config_secrets(
         )
         assert kwargs["to_email"] == "test@example.com"
         assert kwargs["email_body_params"] == {
-            "test_collection": CollectionActionRequired(
+            "test_collection": CheckpointActionRequired(
                 step=CurrentStep.erasure,
                 collection=CollectionAddress("test_dataset", "test_collection"),
                 action_needed=[
diff --git a/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py b/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
index 1cdb59ba4..302d77c08 100644
--- a/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
@@ -594,7 +594,7 @@ def test_get_privacy_requests_by_id(
                             ).rules
                         ],
                     },
-                    "stopped_collection_details": None,
+                    "action_required_details": None,
                     "resume_endpoint": None,
                 }
             ],
@@ -649,7 +649,7 @@ def test_get_privacy_requests_by_partial_id(
                             ).rules
                         ],
                     },
-                    "stopped_collection_details": None,
+                    "action_required_details": None,
                     "resume_endpoint": None,
                 }
             ],
@@ -1004,7 +1004,7 @@ def test_verbose_privacy_requests(
                             ).rules
                         ],
                     },
-                    "stopped_collection_details": None,
+                    "action_required_details": None,
                     "resume_endpoint": None,
                     "results": {
                         "Request approved": [
@@ -1198,7 +1198,7 @@ def test_get_paused_access_privacy_request_resume_info(
 
         data = response.json()["items"][0]
         assert data["status"] == "paused"
-        assert data["stopped_collection_details"] == {
+        assert data["action_required_details"] == {
             "step": "access",
             "collection": "manual_dataset:manual_collection",
             "action_needed": [
@@ -1239,7 +1239,7 @@ def test_get_paused_erasure_privacy_request_resume_info(
 
         data = response.json()["items"][0]
         assert data["status"] == "paused"
-        assert data["stopped_collection_details"] == {
+        assert data["action_required_details"] == {
             "step": "erasure",
             "collection": "manual_dataset:another_collection",
             "action_needed": [
@@ -1266,7 +1266,7 @@ def test_get_paused_webhook_resume_info(
 
         data = response.json()["items"][0]
         assert data["status"] == "paused"
-        assert data["stopped_collection_details"] is None
+        assert data["action_required_details"] is None
         assert data["resume_endpoint"] == "/privacy-request/{}/resume".format(
             privacy_request.id
         )
@@ -1277,7 +1277,7 @@ def test_get_failed_request_resume_info(
         # Mock the privacy request being in an errored state waiting for retry
         privacy_request.status = PrivacyRequestStatus.error
         privacy_request.save(db)
-        privacy_request.cache_failed_collection_details(
+        privacy_request.cache_failed_checkpoint_details(
             step=CurrentStep.erasure,
             collection=CollectionAddress("manual_example", "another_collection"),
         )
@@ -1288,7 +1288,7 @@ def test_get_failed_request_resume_info(
 
         data = response.json()["items"][0]
         assert data["status"] == "error"
-        assert data["stopped_collection_details"] == {
+        assert data["action_required_details"] == {
             "step": "erasure",
             "collection": "manual_example:another_collection",
             "action_needed": None,
@@ -2025,7 +2025,7 @@ def test_resume_privacy_request(
                     for rule in PolicyResponse.from_orm(privacy_request.policy).rules
                 ],
             },
-            "stopped_collection_details": None,
+            "action_required_details": None,
             "resume_endpoint": None,
         }
 
@@ -2373,7 +2373,7 @@ def test_restart_from_failure_not_errored(
             == f"Cannot restart privacy request from failure: privacy request '{privacy_request.id}' status = in_processing."
         )
 
-    def test_restart_from_failure_no_stopped_collection(
+    def test_restart_from_failure_no_stopped_step(
         self, api_client, url, generate_auth_header, db, privacy_request
     ):
         auth_header = generate_auth_header(scopes=[PRIVACY_REQUEST_CALLBACK_RESUME])
@@ -2392,14 +2392,14 @@ def test_restart_from_failure_no_stopped_collection(
     @mock.patch(
         "fidesops.ops.service.privacy_request.request_runner_service.run_privacy_request.delay"
     )
-    def test_restart_from_failure(
+    def test_restart_from_failure_from_specific_collection(
         self, submit_mock, api_client, url, generate_auth_header, db, privacy_request
     ):
         auth_header = generate_auth_header(scopes=[PRIVACY_REQUEST_CALLBACK_RESUME])
         privacy_request.status = PrivacyRequestStatus.error
         privacy_request.save(db)
 
-        privacy_request.cache_failed_collection_details(
+        privacy_request.cache_failed_checkpoint_details(
             step=CurrentStep.access,
             collection=CollectionAddress("test_dataset", "test_collection"),
         )
@@ -2416,6 +2416,33 @@ def test_restart_from_failure(
             from_webhook_id=None,
         )
 
+    @mock.patch(
+        "fidesops.ops.service.privacy_request.request_runner_service.run_privacy_request.delay"
+    )
+    def test_restart_from_failure_outside_graph(
+        self, submit_mock, api_client, url, generate_auth_header, db, privacy_request
+    ):
+        auth_header = generate_auth_header(scopes=[PRIVACY_REQUEST_CALLBACK_RESUME])
+        privacy_request.status = PrivacyRequestStatus.error
+        privacy_request.save(db)
+
+        privacy_request.cache_failed_checkpoint_details(
+            step=CurrentStep.erasure_email_post_send,
+            collection=None,
+        )
+
+        response = api_client.post(url, headers=auth_header)
+        assert response.status_code == 200
+
+        db.refresh(privacy_request)
+        assert privacy_request.status == PrivacyRequestStatus.in_processing
+
+        submit_mock.assert_called_with(
+            privacy_request_id=privacy_request.id,
+            from_step=CurrentStep.erasure_email_post_send.value,
+            from_webhook_id=None,
+        )
+
 
 class TestVerifyIdentity:
     code = "123456"
diff --git a/tests/ops/integration_tests/test_execution.py b/tests/ops/integration_tests/test_execution.py
index db13cdf82..e8d76980b 100644
--- a/tests/ops/integration_tests/test_execution.py
+++ b/tests/ops/integration_tests/test_execution.py
@@ -17,7 +17,7 @@
 from fidesops.ops.models.datasetconfig import convert_dataset_to_graph
 from fidesops.ops.models.policy import CurrentStep
 from fidesops.ops.models.privacy_request import (
-    CollectionActionRequired,
+    CheckpointActionRequired,
     ExecutionLog,
     PrivacyRequest,
 )
@@ -632,7 +632,7 @@ async def test_restart_graph_from_failure(
         ("mongo_test:customer_details", "in_processing"),
         ("mongo_test:customer_details", "error"),
     ]
-    assert privacy_request.get_failed_collection_details() == CollectionActionRequired(
+    assert privacy_request.get_failed_checkpoint_details() == CheckpointActionRequired(
         step=CurrentStep.access,
         collection=CollectionAddress("mongo_test", "customer_details"),
     )
diff --git a/tests/ops/integration_tests/test_integration_email.py b/tests/ops/integration_tests/test_integration_email.py
index d61d77ba0..6351fe75e 100644
--- a/tests/ops/integration_tests/test_integration_email.py
+++ b/tests/ops/integration_tests/test_integration_email.py
@@ -8,7 +8,7 @@
 from fidesops.ops.models.datasetconfig import convert_dataset_to_graph
 from fidesops.ops.models.policy import CurrentStep
 from fidesops.ops.models.privacy_request import (
-    CollectionActionRequired,
+    CheckpointActionRequired,
     ExecutionLog,
     ExecutionLogStatus,
     ManualAction,
@@ -27,7 +27,7 @@
     "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
 )
 @pytest.mark.asyncio
-async def test_collections_with_manual_erasure_confirmation(
+async def test_email_connector_cache_and_delayed_send(
     mock_email_dispatch,
     db,
     erasure_policy,
@@ -114,7 +114,7 @@ async def test_collections_with_manual_erasure_confirmation(
     )
 
     assert raw_email_template_values == {
-        "children": CollectionActionRequired(
+        "children": CheckpointActionRequired(
             step=CurrentStep.erasure,
             collection=CollectionAddress("email_dataset", "children"),
             action_needed=[
@@ -135,7 +135,7 @@ async def test_collections_with_manual_erasure_confirmation(
                 )
             ],
         ),
-        "daycare_customer": CollectionActionRequired(
+        "daycare_customer": CheckpointActionRequired(
             step=CurrentStep.erasure,
             collection=CollectionAddress("email_dataset", "daycare_customer"),
             action_needed=[
@@ -148,7 +148,7 @@ async def test_collections_with_manual_erasure_confirmation(
                 )
             ],
         ),
-        "payment": CollectionActionRequired(
+        "payment": CheckpointActionRequired(
             step=CurrentStep.erasure,
             collection=CollectionAddress("email_dataset", "payment"),
             action_needed=[
diff --git a/tests/ops/models/test_privacy_request.py b/tests/ops/models/test_privacy_request.py
index bb5175a60..338cf486e 100644
--- a/tests/ops/models/test_privacy_request.py
+++ b/tests/ops/models/test_privacy_request.py
@@ -14,9 +14,10 @@
 from fidesops.ops.graph.config import CollectionAddress
 from fidesops.ops.models.policy import CurrentStep, Policy
 from fidesops.ops.models.privacy_request import (
-    CollectionActionRequired,
+    CheckpointActionRequired,
     PrivacyRequest,
     PrivacyRequestStatus,
+    can_run_checkpoint,
 )
 from fidesops.ops.schemas.redis_cache import PrivacyRequestIdentity
 from fidesops.ops.service.connectors.manual_connector import ManualAction
@@ -456,19 +457,19 @@ def test_zero_cached(self, privacy_request):
 class TestPrivacyRequestCacheFailedStep:
     def test_cache_failed_step_and_collection(self, privacy_request):
 
-        privacy_request.cache_failed_collection_details(
+        privacy_request.cache_failed_checkpoint_details(
             step=CurrentStep.erasure, collection=paused_location
         )
 
-        cached_data = privacy_request.get_failed_collection_details()
+        cached_data = privacy_request.get_failed_checkpoint_details()
         assert cached_data.step == CurrentStep.erasure
         assert cached_data.collection == paused_location
         assert cached_data.action_needed is None
 
     def test_cache_null_step_and_location(self, privacy_request):
-        privacy_request.cache_failed_collection_details()
+        privacy_request.cache_failed_checkpoint_details()
 
-        cached_data = privacy_request.get_failed_collection_details()
+        cached_data = privacy_request.get_failed_checkpoint_details()
         assert cached_data is None
 
 
@@ -505,7 +506,7 @@ def test_cache_template_contents(self, privacy_request):
         assert privacy_request.get_email_connector_template_contents_by_dataset(
             CurrentStep.erasure, "email_dataset"
         ) == {
-            "test_collection": CollectionActionRequired(
+            "test_collection": CheckpointActionRequired(
                 step=CurrentStep.erasure,
                 collection=CollectionAddress("email_dataset", "test_collection"),
                 action_needed=[
@@ -517,3 +518,40 @@ def test_cache_template_contents(self, privacy_request):
                 ],
             )
         }
+
+
+class TestCanRunFromCheckpoint:
+    def test_can_run_from_checkpoint(self):
+        assert (
+            can_run_checkpoint(
+                request_checkpoint=CurrentStep.erasure_email_post_send,
+                from_checkpoint=CurrentStep.erasure,
+            )
+            is True
+        )
+
+    def test_can_run_from_equivalent_checkpoint(self):
+        assert (
+            can_run_checkpoint(
+                request_checkpoint=CurrentStep.erasure,
+                from_checkpoint=CurrentStep.erasure,
+            )
+            is True
+        )
+
+    def test_cannot_run_from_completed_checkpoint(self):
+        assert (
+            can_run_checkpoint(
+                request_checkpoint=CurrentStep.access,
+                from_checkpoint=CurrentStep.erasure,
+            )
+            is False
+        )
+
+    def test_can_run_if_no_saved_checkpoint(self):
+        assert (
+            can_run_checkpoint(
+                request_checkpoint=CurrentStep.access,
+            )
+            is True
+        )
diff --git a/tests/ops/service/privacy_request/request_runner_service_test.py b/tests/ops/service/privacy_request/request_runner_service_test.py
index 792e2efb5..c2da89460 100644
--- a/tests/ops/service/privacy_request/request_runner_service_test.py
+++ b/tests/ops/service/privacy_request/request_runner_service_test.py
@@ -20,6 +20,7 @@
 from fidesops.ops.models.policy import CurrentStep, PolicyPostWebhook
 from fidesops.ops.models.privacy_request import (
     ActionType,
+    CheckpointActionRequired,
     ExecutionLog,
     PolicyPreWebhook,
     PrivacyRequest,
@@ -1448,6 +1449,10 @@ def test_run_webhooks_client_error(
         assert not proceed
         assert privacy_request.status == PrivacyRequestStatus.error
         assert privacy_request.finished_processing_at is not None
+        assert (
+            privacy_request.get_failed_checkpoint_details()
+            == CheckpointActionRequired(step=CurrentStep.pre_webhooks)
+        )
         assert privacy_request.paused_at is None
 
     @mock.patch(
@@ -1593,3 +1598,53 @@ def test_create_and_process_erasure_request_email_connector(
     kwargs = mailgun_send.call_args.kwargs
     assert type(kwargs["email_config"]) == EmailConfig
     assert type(kwargs["email"]) == EmailForActionType
+
+
+@mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+@pytest.mark.integration
+def test_create_and_process_erasure_request_email_connector_email_send_error(
+    mailgun_send,
+    email_connection_config,
+    erasure_policy,
+    integration_postgres_config,
+    run_privacy_request_task,
+    email_dataset_config,
+    postgres_example_test_dataset_config_read_access,
+    db,
+):
+    """
+    Force error by having no email config setup
+    """
+    rule = erasure_policy.rules[0]
+    target = rule.targets[0]
+    target.data_category = "user.childrens"
+    target.save(db=db)
+
+    email = "customer-1@example.com"
+    data = {
+        "requested_at": "2021-08-30T16:09:37.359Z",
+        "policy_key": erasure_policy.key,
+        "identity": {"email": email},
+    }
+
+    pr = get_privacy_request_results(
+        db,
+        erasure_policy,
+        run_privacy_request_task,
+        data,
+    )
+    db.refresh(pr)
+    assert pr.status == PrivacyRequestStatus.error
+    assert pr.get_failed_checkpoint_details() == CheckpointActionRequired(
+        step=CurrentStep.erasure_email_post_send, collection=None, action_needed=None
+    )
+    cached_email_contents = pr.get_email_connector_template_contents_by_dataset(
+        CurrentStep.erasure, "email_dataset"
+    )
+    assert list(cached_email_contents.keys()) == [
+        "payment",
+        "children",
+        "daycare_customer",
+    ]
+    pr.delete(db=db)
+    assert mailgun_send.called is False

From 1601293119ddde9b4978a78bdf5d7e465f142dce Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Fri, 2 Sep 2022 09:30:12 -0500
Subject: [PATCH 03/12] Don't send an email if the connection config is read
 only or there are no updates to be applied to any of the collections on the
 dataset.

---
 .../privacy_request/request_runner_service.py |  26 +-
 .../test_integration_email.py                 |   4 +-
 .../request_runner_service_test.py            | 271 ++++++++++++------
 3 files changed, 217 insertions(+), 84 deletions(-)

diff --git a/src/fidesops/ops/service/privacy_request/request_runner_service.py b/src/fidesops/ops/service/privacy_request/request_runner_service.py
index 891115551..cf8c231ed 100644
--- a/src/fidesops/ops/service/privacy_request/request_runner_service.py
+++ b/src/fidesops/ops/service/privacy_request/request_runner_service.py
@@ -320,7 +320,9 @@ async def run_privacy_request(
             from_checkpoint=resume_step,
         ):
             try:
-                email_connector_send(db=session, privacy_request=privacy_request)
+                email_connector_erasure_send(
+                    db=session, privacy_request=privacy_request
+                )
             except EmailDispatchException as exc:
                 privacy_request.cache_failed_checkpoint_details(
                     step=CurrentStep.erasure_email_post_send, collection=None
@@ -397,7 +399,7 @@ def generate_id_verification_code() -> str:
     return str(random.choice(range(100000, 999999)))
 
 
-def email_connector_send(db: Session, privacy_request: PrivacyRequest) -> None:
+def email_connector_erasure_send(db: Session, privacy_request: PrivacyRequest) -> None:
     """
     Send emails to configured third-parties with instructions on how to erase remaining data.
     Combined all the collections on each email-based dataset into one email.
@@ -413,6 +415,26 @@ def email_connector_send(db: Session, privacy_request: PrivacyRequest) -> None:
             CurrentStep.erasure, ds.dataset.get("fides_key")
         )
 
+        if not template_values:
+            logger.info(
+                "No email sent: no template values saved for '%s'",
+                ds.dataset.get("fides_key"),
+            )
+            return
+
+        if not any(
+            (
+                action_required.action_needed[0].update
+                if action_required and action_required.action_needed
+                else False
+                for action_required in template_values.values()
+            )
+        ):
+            logger.info(
+                "No email sent: no masking needed on '%s'", ds.dataset.get("fides_key")
+            )
+            return
+
         dispatch_email(
             db,
             action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
diff --git a/tests/ops/integration_tests/test_integration_email.py b/tests/ops/integration_tests/test_integration_email.py
index 6351fe75e..b5190dc31 100644
--- a/tests/ops/integration_tests/test_integration_email.py
+++ b/tests/ops/integration_tests/test_integration_email.py
@@ -16,7 +16,7 @@
 from fidesops.ops.schemas.dataset import FidesopsDataset
 from fidesops.ops.schemas.email.email import EmailActionType
 from fidesops.ops.service.privacy_request.request_runner_service import (
-    email_connector_send,
+    email_connector_erasure_send,
 )
 from fidesops.ops.task import graph_task
 
@@ -173,7 +173,7 @@ async def test_email_connector_cache_and_delayed_send(
         log.status for log in children_logs
     }
 
-    email_connector_send(db, privacy_request)
+    email_connector_erasure_send(db, privacy_request)
     assert mock_email_dispatch.called
     call_args = mock_email_dispatch.call_args[1]
     assert call_args["action_type"] == EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT
diff --git a/tests/ops/service/privacy_request/request_runner_service_test.py b/tests/ops/service/privacy_request/request_runner_service_test.py
index c2da89460..952c27325 100644
--- a/tests/ops/service/privacy_request/request_runner_service_test.py
+++ b/tests/ops/service/privacy_request/request_runner_service_test.py
@@ -16,6 +16,7 @@
     PrivacyRequestPaused,
 )
 from fidesops.ops.core.config import config
+from fidesops.ops.models.connectionconfig import AccessLevel
 from fidesops.ops.models.email import EmailConfig
 from fidesops.ops.models.policy import CurrentStep, PolicyPostWebhook
 from fidesops.ops.models.privacy_request import (
@@ -1559,92 +1560,202 @@ def test_privacy_request_log_failure(
         assert sent_event.extra_data == {"privacy_request": pr.id}
 
 
-@mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
-@pytest.mark.integration
-def test_create_and_process_erasure_request_email_connector(
-    mailgun_send,
-    email_connection_config,
-    erasure_policy,
-    integration_postgres_config,
-    run_privacy_request_task,
-    email_dataset_config,
-    postgres_example_test_dataset_config_read_access,
-    email_config,
-    db,
-):
-    """
-    Asserts that mailgun was called and verifies email template renders without error
-    """
-    rule = erasure_policy.rules[0]
-    target = rule.targets[0]
-    target.data_category = "user.childrens"
-    target.save(db=db)
-
-    email = "customer-1@example.com"
-    data = {
-        "requested_at": "2021-08-30T16:09:37.359Z",
-        "policy_key": erasure_policy.key,
-        "identity": {"email": email},
-    }
-
-    pr = get_privacy_request_results(
+class TestPrivacyRequestsEmailConnector:
+    @mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+    @pytest.mark.integration
+    def test_create_and_process_erasure_request_email_connector(
+        self,
+        mailgun_send,
+        email_connection_config,
+        erasure_policy,
+        integration_postgres_config,
+        run_privacy_request_task,
+        email_dataset_config,
+        postgres_example_test_dataset_config_read_access,
+        email_config,
         db,
+    ):
+        """
+        Asserts that mailgun was called and verifies email template renders without error
+        """
+        rule = erasure_policy.rules[0]
+        target = rule.targets[0]
+        target.data_category = "user.childrens"
+        target.save(db=db)
+
+        email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": erasure_policy.key,
+            "identity": {"email": email},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            erasure_policy,
+            run_privacy_request_task,
+            data,
+        )
+        pr.delete(db=db)
+        assert mailgun_send.called
+        kwargs = mailgun_send.call_args.kwargs
+        assert type(kwargs["email_config"]) == EmailConfig
+        assert type(kwargs["email"]) == EmailForActionType
+
+    @mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+    @pytest.mark.integration
+    def test_create_and_process_erasure_request_email_connector_email_send_error(
+        self,
+        mailgun_send,
+        email_connection_config,
         erasure_policy,
+        integration_postgres_config,
         run_privacy_request_task,
-        data,
-    )
-    pr.delete(db=db)
-    assert mailgun_send.called
-    kwargs = mailgun_send.call_args.kwargs
-    assert type(kwargs["email_config"]) == EmailConfig
-    assert type(kwargs["email"]) == EmailForActionType
+        email_dataset_config,
+        postgres_example_test_dataset_config_read_access,
+        db,
+    ):
+        """
+        Force error by having no email config setup
+        """
+        rule = erasure_policy.rules[0]
+        target = rule.targets[0]
+        target.data_category = "user.childrens"
+        target.save(db=db)
+
+        email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": erasure_policy.key,
+            "identity": {"email": email},
+        }
 
+        pr = get_privacy_request_results(
+            db,
+            erasure_policy,
+            run_privacy_request_task,
+            data,
+        )
+        db.refresh(pr)
+        assert pr.status == PrivacyRequestStatus.error
+        assert pr.get_failed_checkpoint_details() == CheckpointActionRequired(
+            step=CurrentStep.erasure_email_post_send,
+            collection=None,
+            action_needed=None,
+        )
+        cached_email_contents = pr.get_email_connector_template_contents_by_dataset(
+            CurrentStep.erasure, "email_dataset"
+        )
+        assert set(cached_email_contents.keys()) == {
+            "payment",
+            "children",
+            "daycare_customer",
+        }
+        pr.delete(db=db)
+        assert mailgun_send.called is False
 
-@mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
-@pytest.mark.integration
-def test_create_and_process_erasure_request_email_connector_email_send_error(
-    mailgun_send,
-    email_connection_config,
-    erasure_policy,
-    integration_postgres_config,
-    run_privacy_request_task,
-    email_dataset_config,
-    postgres_example_test_dataset_config_read_access,
-    db,
-):
-    """
-    Force error by having no email config setup
-    """
-    rule = erasure_policy.rules[0]
-    target = rule.targets[0]
-    target.data_category = "user.childrens"
-    target.save(db=db)
+    @mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+    @pytest.mark.integration
+    def test_email_connector_read_only_permissions(
+        self,
+        mailgun_send,
+        email_connection_config,
+        erasure_policy,
+        integration_postgres_config,
+        run_privacy_request_task,
+        email_dataset_config,
+        email_config,
+        postgres_example_test_dataset_config_read_access,
+        db,
+    ):
+        """
+        Set email config to read only - don't send email in this case.
+        """
+        rule = erasure_policy.rules[0]
+        target = rule.targets[0]
+        target.data_category = "user.childrens"
+        target.save(db=db)
+
+        email_connection_config.access = AccessLevel.read
+        email_connection_config.save(db)
+
+        email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": erasure_policy.key,
+            "identity": {"email": email},
+        }
 
-    email = "customer-1@example.com"
-    data = {
-        "requested_at": "2021-08-30T16:09:37.359Z",
-        "policy_key": erasure_policy.key,
-        "identity": {"email": email},
-    }
+        pr = get_privacy_request_results(
+            db,
+            erasure_policy,
+            run_privacy_request_task,
+            data,
+        )
+        db.refresh(pr)
+        assert pr.status == PrivacyRequestStatus.complete
+        cached_email_contents = pr.get_email_connector_template_contents_by_dataset(
+            CurrentStep.erasure, "email_dataset"
+        )
+        assert (
+            set(cached_email_contents.keys()) == set()
+        ), "No data cached to erase, because this connector is read-only"
 
-    pr = get_privacy_request_results(
-        db,
+        pr.delete(db=db)
+        assert (
+            mailgun_send.called is False
+        ), "Email not sent because the connection was read only"
+
+    @mock.patch("fidesops.ops.service.email.email_dispatch_service._mailgun_dispatcher")
+    @pytest.mark.integration
+    def test_email_connector_no_updates_needed(
+        self,
+        mailgun_send,
+        email_connection_config,
         erasure_policy,
+        integration_postgres_config,
         run_privacy_request_task,
-        data,
-    )
-    db.refresh(pr)
-    assert pr.status == PrivacyRequestStatus.error
-    assert pr.get_failed_checkpoint_details() == CheckpointActionRequired(
-        step=CurrentStep.erasure_email_post_send, collection=None, action_needed=None
-    )
-    cached_email_contents = pr.get_email_connector_template_contents_by_dataset(
-        CurrentStep.erasure, "email_dataset"
-    )
-    assert list(cached_email_contents.keys()) == [
-        "payment",
-        "children",
-        "daycare_customer",
-    ]
-    pr.delete(db=db)
-    assert mailgun_send.called is False
+        email_dataset_config,
+        email_config,
+        postgres_example_test_dataset_config_read_access,
+        db,
+    ):
+        """
+        Don't send an email when there are no erasures needed
+        """
+        rule = erasure_policy.rules[0]
+        target = rule.targets[0]
+        target.data_category = "user.job_title"  # Add a data category that does not apply to the email dataset
+        target.save(db=db)
+
+        email = "customer-1@example.com"
+        data = {
+            "requested_at": "2021-08-30T16:09:37.359Z",
+            "policy_key": erasure_policy.key,
+            "identity": {"email": email},
+        }
+
+        pr = get_privacy_request_results(
+            db,
+            erasure_policy,
+            run_privacy_request_task,
+            data,
+        )
+        db.refresh(pr)
+        assert pr.status == PrivacyRequestStatus.complete
+        cached_email_contents = pr.get_email_connector_template_contents_by_dataset(
+            CurrentStep.erasure, "email_dataset"
+        )
+        assert set(cached_email_contents.keys()) == {
+            "payment",
+            "children",
+            "daycare_customer",
+        }
+        assert cached_email_contents["payment"].action_needed[0].update is None
+        assert cached_email_contents["children"].action_needed[0].update is None
+        assert cached_email_contents["daycare_customer"].action_needed[0].update is None
+
+        pr.delete(db=db)
+        assert (
+            mailgun_send.called is False
+        ), "Email not sent because no updates are needed. Data category doesn't apply to any of the collections."

From f539a012f5b077025099902d19649d8803d0bfd6 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Fri, 2 Sep 2022 10:44:21 -0500
Subject: [PATCH 04/12] Don't assume there's a collection when building
 "resume" details. A failed privacy request can be resumed outside of the
 traversal.

---
 CHANGELOG.md                                  |  1 +
 .../v1/endpoints/privacy_request_endpoints.py |  2 +-
 .../privacy_request/request_runner_service.py |  9 ++++++-
 .../test_privacy_request_endpoints.py         | 26 ++++++++++++++++++-
 4 files changed, 35 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 597494a18..28646ca7c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -29,6 +29,7 @@ The types of changes are:
 * Adds users and owners configuration for Hubspot connector [#1091](https://github.com/ethyca/fidesops/pull/1091)
 * Foundations for a new email connector type [#1142](https://github.com/ethyca/fidesops/pull/1142)
 * Have the new email connector cache action needed for each collection [#1168](https://github.com/ethyca/fidesops/pull/1168)
+* Wrap up the email connector - it sends an email with erasure instructions as part of request execution [#1246](https://github.com/ethyca/fidesops/pull/1246)
 
 ### Docs
 
diff --git a/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py b/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py
index d6e9e3597..1f31ce2f1 100644
--- a/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py
+++ b/src/fidesops/ops/api/v1/endpoints/privacy_request_endpoints.py
@@ -515,7 +515,7 @@ def attach_resume_instructions(privacy_request: PrivacyRequest) -> None:
     if action_required_details:
         action_required_details.step = action_required_details.step.value  # type: ignore
         action_required_details.collection = (
-            action_required_details.collection.value  # type: ignore
+            action_required_details.collection.value if action_required_details.collection else None  # type: ignore
         )
 
     privacy_request.action_required_details = action_required_details
diff --git a/src/fidesops/ops/service/privacy_request/request_runner_service.py b/src/fidesops/ops/service/privacy_request/request_runner_service.py
index cf8c231ed..a3e6c9502 100644
--- a/src/fidesops/ops/service/privacy_request/request_runner_service.py
+++ b/src/fidesops/ops/service/privacy_request/request_runner_service.py
@@ -209,7 +209,7 @@ async def run_privacy_request(
     from_webhook_id: Optional[str] = None,
     from_step: Optional[str] = None,
 ) -> None:
-    # pylint: disable=too-many-locals
+    # pylint: disable=too-many-locals, too-many-statements
     """
     Dispatch a privacy_request into the execution layer by:
         1. Generate a graph from all the currently configured datasets
@@ -221,6 +221,8 @@ async def run_privacy_request(
     coroutine for it.
     """
     resume_step: Optional[CurrentStep] = CurrentStep(from_step) if from_step else None  # type: ignore
+    if from_step:
+        logger.info("Resuming privacy request from checkpoint: '%s'", from_step)
 
     with self.session as session:
 
@@ -442,6 +444,11 @@ def email_connector_erasure_send(db: Session, privacy_request: PrivacyRequest) -
             email_body_params=template_values,
         )
 
+        logger.info(
+            "Email send succeeded for request '%s' for dataset: '%s'",
+            privacy_request.id,
+            ds.dataset.get("fides_key"),
+        )
         AuditLog.create(
             db=db,
             data={
diff --git a/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py b/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
index 302d77c08..c914296b5 100644
--- a/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
@@ -1271,7 +1271,7 @@ def test_get_paused_webhook_resume_info(
             privacy_request.id
         )
 
-    def test_get_failed_request_resume_info(
+    def test_get_failed_request_resume_info_from_collection(
         self, db, privacy_request, generate_auth_header, api_client, url
     ):
         # Mock the privacy request being in an errored state waiting for retry
@@ -1295,6 +1295,30 @@ def test_get_failed_request_resume_info(
         }
         assert data["resume_endpoint"] == f"/privacy-request/{privacy_request.id}/retry"
 
+    def test_get_failed_request_resume_info_from_email_send(
+        self, db, privacy_request, generate_auth_header, api_client, url
+    ):
+        # Mock the privacy request being in an errored state waiting for retry
+        privacy_request.status = PrivacyRequestStatus.error
+        privacy_request.save(db)
+        privacy_request.cache_failed_checkpoint_details(
+            step=CurrentStep.erasure_email_post_send,
+            collection=None,
+        )
+
+        auth_header = generate_auth_header(scopes=[PRIVACY_REQUEST_READ])
+        response = api_client.get(url, headers=auth_header)
+        assert 200 == response.status_code
+
+        data = response.json()["items"][0]
+        assert data["status"] == "error"
+        assert data["action_required_details"] == {
+            "step": "erasure_email_post_send",
+            "collection": None,
+            "action_needed": None,
+        }
+        assert data["resume_endpoint"] == f"/privacy-request/{privacy_request.id}/retry"
+
 
 class TestGetExecutionLogs:
     @pytest.fixture(scope="function")

From e1dc1fbe7bd1ca6267847f2986a4a934a691d4f9 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Fri, 2 Sep 2022 13:27:51 -0500
Subject: [PATCH 05/12] Add a first draft of docs for setting up an email
 connector.

---
 .../docs/guides/email_communications.md       |   2 +-
 docs/fidesops/docs/guides/email_connectors.md | 106 ++++++++++++++++++
 docs/fidesops/mkdocs.yml                      |   1 +
 3 files changed, 108 insertions(+), 1 deletion(-)
 create mode 100644 docs/fidesops/docs/guides/email_connectors.md

diff --git a/docs/fidesops/docs/guides/email_communications.md b/docs/fidesops/docs/guides/email_communications.md
index bc36f1f70..08d3a35f1 100644
--- a/docs/fidesops/docs/guides/email_communications.md
+++ b/docs/fidesops/docs/guides/email_communications.md
@@ -6,7 +6,7 @@ Fidesops supports email server configurations for sending processing notices to
 Supported modes of use:
 
 - Subject Identity Verification - for more information on identity verification in subject requests, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
-
+- Erasure Request Email Fulfillment - sends an email to configured third parties to process erasures for a given data subject.  See [Email Connectors](email_connectors.md) for more information.
 
 ## Prerequisites
 
diff --git a/docs/fidesops/docs/guides/email_connectors.md b/docs/fidesops/docs/guides/email_connectors.md
new file mode 100644
index 000000000..aaed647b0
--- /dev/null
+++ b/docs/fidesops/docs/guides/email_connectors.md
@@ -0,0 +1,106 @@
+# Setting up an Email Connector
+
+
+## What is the purpose of the Email Connector?
+
+The Email Connector is a ConnectionConfig type that emails a third party to ask them to complete
+a Privacy Request when their service cannot be accessed automatically. 
+
+Fidesops will gather details about each collection described on the third party service as part of request execution and 
+wait to send a single email to the service after all collections have been visited.  Fidesops does not 
+collect confirmation that the erasure was completed by the third party; the EmailConnector is only responsible 
+for notifying them.
+
+Importantly, only *erasure* requests are supported at this time for EmailConnectors.  
+
+
+## What pieces should be configured for an Email Connector?
+
+In short, you will need to create an `email` ConnectionConfig to store minor details about the third party service, and a DatasetConfig
+to describe the data contained in their datasource.  You will also need to have configured a separate [EmailConfig](email_communications.md) which
+is used system-wide to carry out the actual email send.
+
+
+## Setting up the ConnectionConfig
+
+Create a ConnectionConfig object with
+
+```json title="<code>PATCH api/v1/connection</code>"
+[
+  { 
+    "name": "Email Connection Config",
+    "key": "third_party_email_connector",
+    "connection_type": "email",
+    "access": "write"
+  }
+]
+```
+EmailConnectors must be given "write" access in order to send an email.
+
+
+## Configuring who to notify
+
+Save a `to_email` on the ConnectionConfig secrets.  This is the user that will be notified via email to complete
+an erasure. Only one `to_email` is supported at this time.
+
+Optionally, configure a `test_email` to which you have access, to verify that your setup is working.  Provided your 
+EmailConfig is set up properly, you should receive an email similar to the one sent to third-party services, containing
+dummy data.
+
+```json title="<code>PUT api/v1/connection/<email_connection_config_key>/secret</code>" 
+{
+    "test_email": "my_email@example.com",
+    "to_email": "third_party@example.com
+}
+```
+
+## Configuring the dataset
+
+Describe the collections and fields on a third party source with a [DatasetConfig](datasets.md), the same way you'd describe attributes 
+on a database.  If you do not know the exact data structure of a third party, you might configure a single collection
+with the fields you'd like masked.
+
+As with all collections that support erasures, a primary key must be specified on each collection.
+
+
+```json title="<code>PUT api/v1/connection/<email_connection_config_key>/dataset" 
+[
+    {
+      "fides_key": "email_dataset",
+      "name": "Dataset not accessible automatically",
+      "description": "Third party data - will email to request erasure",
+      "collections": [
+        {
+          "name": "daycare_customer",
+          "fields": [
+            {
+              "name": "id",
+              "data_categories": [
+                "system.operations"
+              ],
+              "fidesops_meta": {
+                "primary_key": true
+              }
+            },
+            {
+              "name": "child_health_concerns",
+              "data_categories": [
+                "user.biometric_health"
+              ]
+            },
+            {
+              "name": "user_email",
+              "data_categories": [
+                "user.contact.email"
+              ],
+              "fidesops_meta": {
+                "identity": "email"
+              }
+            }
+          ]
+        }
+      ]
+    }
+]
+```
+
diff --git a/docs/fidesops/mkdocs.yml b/docs/fidesops/mkdocs.yml
index c76f1f03a..60f9784fe 100644
--- a/docs/fidesops/mkdocs.yml
+++ b/docs/fidesops/mkdocs.yml
@@ -28,6 +28,7 @@ nav:
       - Configure OneTrust Integration: guides/onetrust.md
       - Preview Query Execution: guides/query_execution.md
       - Data Rights Protocol: guides/data_rights_protocol.md
+      - Configure Email Connectors: guides/email_connectors.md
   - SaaS Connectors:
       - Connect to SaaS Applications: saas_connectors/saas_connectors.md
       - SaaS Configuration: saas_connectors/saas_config.md

From c09b4e735f1259bc54fdf1db443c0cbfbb89b192 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Fri, 2 Sep 2022 16:04:30 -0500
Subject: [PATCH 06/12] Moves the email connector send method to the email
 connector file.

---
 src/fidesops/ops/models/privacy_request.py    |  2 +-
 .../ops/service/connectors/email_connector.py | 71 ++++++++++++++++++-
 .../privacy_request/request_runner_service.py | 65 +----------------
 .../test_integration_email.py                 |  4 +-
 4 files changed, 73 insertions(+), 69 deletions(-)

diff --git a/src/fidesops/ops/models/privacy_request.py b/src/fidesops/ops/models/privacy_request.py
index 2555717b3..0f3b1652e 100644
--- a/src/fidesops/ops/models/privacy_request.py
+++ b/src/fidesops/ops/models/privacy_request.py
@@ -68,7 +68,7 @@
 
 logger = logging.getLogger(__name__)
 
-
+# Locations from which privacy request execution can be resumed, in order.
 EXECUTION_CHECKPOINTS = [
     CurrentStep.pre_webhooks,
     CurrentStep.access,
diff --git a/src/fidesops/ops/service/connectors/email_connector.py b/src/fidesops/ops/service/connectors/email_connector.py
index beb1a78c2..5731e8f0b 100644
--- a/src/fidesops/ops/service/connectors/email_connector.py
+++ b/src/fidesops/ops/service/connectors/email_connector.py
@@ -1,6 +1,7 @@
 import logging
 from typing import Any, Dict, List, Optional
 
+from fideslib.models.audit_log import AuditLog, AuditLogAction
 from sqlalchemy.orm import Session
 
 from fidesops.ops.common_exceptions import (
@@ -9,7 +10,12 @@
 )
 from fidesops.ops.graph.config import CollectionAddress, FieldPath
 from fidesops.ops.graph.traversal import TraversalNode
-from fidesops.ops.models.connectionconfig import ConnectionTestStatus
+from fidesops.ops.models.connectionconfig import (
+    ConnectionConfig,
+    ConnectionTestStatus,
+    ConnectionType,
+)
+from fidesops.ops.models.datasetconfig import DatasetConfig
 from fidesops.ops.models.policy import CurrentStep, Policy, Rule
 from fidesops.ops.models.privacy_request import (
     CheckpointActionRequired,
@@ -84,7 +90,7 @@ def retrieve_data(  # type: ignore
     ) -> Optional[List[Row]]:
         """Access requests are not supported at this time."""
         logger.info(
-            "Access request not supported for email connector `%s` at this time.",
+            "Access requests not supported for email connector '%s' at this time.",
             node.address.value,
         )
         return []
@@ -112,6 +118,8 @@ def mask_data(  # type: ignore
             action_needed=[manual_action],
         )
 
+        # Raises a special exception just to update the ExecutionLog message.  The email send itself
+        # is postponed until all collections have been visited.
         raise PrivacyRequestErasureEmailSendRequired("email prepared")
 
     def build_masking_instructions(
@@ -148,3 +156,62 @@ def build_masking_instructions(
         # Returns a ManualAction even if there are no fields to mask on this collection,
         # because the locators still may be needed to find data to mask on dependent collections
         return ManualAction(locators=locators, update=mask_map if mask_map else None)
+
+
+def email_connector_erasure_send(db: Session, privacy_request: PrivacyRequest) -> None:
+    """
+    Send emails to configured third-parties with instructions on how to erase remaining data.
+    Combined all the collections on each email-based dataset into one email.
+    """
+    email_dataset_configs = db.query(DatasetConfig, ConnectionConfig).filter(
+        DatasetConfig.connection_config_id == ConnectionConfig.id,
+        ConnectionConfig.connection_type == ConnectionType.email,
+    )
+    for ds, cc in email_dataset_configs:
+        template_values: Dict[
+            str, Optional[CheckpointActionRequired]
+        ] = privacy_request.get_email_connector_template_contents_by_dataset(
+            CurrentStep.erasure, ds.dataset.get("fides_key")
+        )
+
+        if not template_values:
+            logger.info(
+                "No email sent: no template values saved for '%s'",
+                ds.dataset.get("fides_key"),
+            )
+            return
+
+        if not any(
+            (
+                action_required.action_needed[0].update
+                if action_required and action_required.action_needed
+                else False
+                for action_required in template_values.values()
+            )
+        ):
+            logger.info(
+                "No email sent: no masking needed on '%s'", ds.dataset.get("fides_key")
+            )
+            return
+
+        dispatch_email(
+            db,
+            action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
+            to_email=cc.secrets.get("to_email"),
+            email_body_params=template_values,
+        )
+
+        logger.info(
+            "Email send succeeded for request '%s' for dataset: '%s'",
+            privacy_request.id,
+            ds.dataset.get("fides_key"),
+        )
+        AuditLog.create(
+            db=db,
+            data={
+                "user_id": "system",
+                "privacy_request_id": privacy_request.id,
+                "action": AuditLogAction.email_sent,
+                "message": f"Erasure email instructions dispatched for '{ds.dataset.get('fides_key')}'",
+            },
+        )
diff --git a/src/fidesops/ops/service/privacy_request/request_runner_service.py b/src/fidesops/ops/service/privacy_request/request_runner_service.py
index a3e6c9502..b7cd9ef2d 100644
--- a/src/fidesops/ops/service/privacy_request/request_runner_service.py
+++ b/src/fidesops/ops/service/privacy_request/request_runner_service.py
@@ -23,7 +23,7 @@
     fideslog_graph_failure,
 )
 from fidesops.ops.graph.graph import DatasetGraph
-from fidesops.ops.models.connectionconfig import ConnectionConfig, ConnectionType
+from fidesops.ops.models.connectionconfig import ConnectionConfig
 from fidesops.ops.models.datasetconfig import DatasetConfig
 from fidesops.ops.models.policy import (
     ActionType,
@@ -34,13 +34,11 @@
     WebhookTypes,
 )
 from fidesops.ops.models.privacy_request import (
-    CheckpointActionRequired,
     PrivacyRequest,
     PrivacyRequestStatus,
     can_run_checkpoint,
 )
-from fidesops.ops.schemas.email.email import EmailActionType
-from fidesops.ops.service.email.email_dispatch_service import dispatch_email
+from fidesops.ops.service.connectors.email_connector import email_connector_erasure_send
 from fidesops.ops.service.storage.storage_uploader_service import upload
 from fidesops.ops.task.filter_results import filter_data_categories
 from fidesops.ops.task.graph_task import (
@@ -399,62 +397,3 @@ def generate_id_verification_code() -> str:
     Generate one-time identity verification code
     """
     return str(random.choice(range(100000, 999999)))
-
-
-def email_connector_erasure_send(db: Session, privacy_request: PrivacyRequest) -> None:
-    """
-    Send emails to configured third-parties with instructions on how to erase remaining data.
-    Combined all the collections on each email-based dataset into one email.
-    """
-    email_dataset_configs = db.query(DatasetConfig, ConnectionConfig).filter(
-        DatasetConfig.connection_config_id == ConnectionConfig.id,
-        ConnectionConfig.connection_type == ConnectionType.email,
-    )
-    for ds, cc in email_dataset_configs:
-        template_values: Dict[
-            str, Optional[CheckpointActionRequired]
-        ] = privacy_request.get_email_connector_template_contents_by_dataset(
-            CurrentStep.erasure, ds.dataset.get("fides_key")
-        )
-
-        if not template_values:
-            logger.info(
-                "No email sent: no template values saved for '%s'",
-                ds.dataset.get("fides_key"),
-            )
-            return
-
-        if not any(
-            (
-                action_required.action_needed[0].update
-                if action_required and action_required.action_needed
-                else False
-                for action_required in template_values.values()
-            )
-        ):
-            logger.info(
-                "No email sent: no masking needed on '%s'", ds.dataset.get("fides_key")
-            )
-            return
-
-        dispatch_email(
-            db,
-            action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
-            to_email=cc.secrets.get("to_email"),
-            email_body_params=template_values,
-        )
-
-        logger.info(
-            "Email send succeeded for request '%s' for dataset: '%s'",
-            privacy_request.id,
-            ds.dataset.get("fides_key"),
-        )
-        AuditLog.create(
-            db=db,
-            data={
-                "user_id": "system",
-                "privacy_request_id": privacy_request.id,
-                "action": AuditLogAction.email_sent,
-                "message": f"Erasure email instructions dispatched for '{ds.dataset.get('fides_key')}'",
-            },
-        )
diff --git a/tests/ops/integration_tests/test_integration_email.py b/tests/ops/integration_tests/test_integration_email.py
index b5190dc31..5493efbef 100644
--- a/tests/ops/integration_tests/test_integration_email.py
+++ b/tests/ops/integration_tests/test_integration_email.py
@@ -15,9 +15,7 @@
 )
 from fidesops.ops.schemas.dataset import FidesopsDataset
 from fidesops.ops.schemas.email.email import EmailActionType
-from fidesops.ops.service.privacy_request.request_runner_service import (
-    email_connector_erasure_send,
-)
+from fidesops.ops.service.connectors.email_connector import email_connector_erasure_send
 from fidesops.ops.task import graph_task
 
 

From c8afe55437269051828a4617e6b0c448f7127c41 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Fri, 2 Sep 2022 16:34:54 -0500
Subject: [PATCH 07/12] Update mock location.

---
 tests/ops/integration_tests/test_integration_email.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/tests/ops/integration_tests/test_integration_email.py b/tests/ops/integration_tests/test_integration_email.py
index 5493efbef..909f57b66 100644
--- a/tests/ops/integration_tests/test_integration_email.py
+++ b/tests/ops/integration_tests/test_integration_email.py
@@ -21,9 +21,7 @@
 
 @pytest.mark.integration_postgres
 @pytest.mark.integration
-@mock.patch(
-    "fidesops.ops.service.privacy_request.request_runner_service.dispatch_email"
-)
+@mock.patch("fidesops.ops.service.connectors.email_connector.dispatch_email")
 @pytest.mark.asyncio
 async def test_email_connector_cache_and_delayed_send(
     mock_email_dispatch,

From d9e169a0253fe774e8202cfa22cb0bbbce43663e Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Fri, 2 Sep 2022 16:57:58 -0500
Subject: [PATCH 08/12] Bump downrev.

---
 .../migrations/versions/912d801f06c0_audit_log_email_send.py  | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py b/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py
index 7d8745ce1..58bf2e095 100644
--- a/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py
+++ b/src/fidesops/ops/migrations/versions/912d801f06c0_audit_log_email_send.py
@@ -1,7 +1,7 @@
 """audit log email send
 
 Revision ID: 912d801f06c0
-Revises: c2f7a29c4780
+Revises: bde646a6f51e
 Create Date: 2022-09-01 16:23:10.905356
 
 """
@@ -10,7 +10,7 @@
 
 # revision identifiers, used by Alembic.
 revision = "912d801f06c0"
-down_revision = "c2f7a29c4780"
+down_revision = "bde646a6f51e"
 branch_labels = None
 depends_on = None
 

From 10ba182be4e00e1d04305288627239a674d07619 Mon Sep 17 00:00:00 2001
From: Cole <cole@ethyca.com>
Date: Tue, 6 Sep 2022 13:59:58 -0400
Subject: [PATCH 09/12] update email connector guides

---
 .../docs/guides/email_communications.md       | 117 ++++++++++++++++--
 docs/fidesops/docs/guides/email_connectors.md | 106 ----------------
 docs/fidesops/docs/guides/privacy_requests.md |   6 +-
 docs/fidesops/mkdocs.yml                      |   3 +-
 4 files changed, 112 insertions(+), 120 deletions(-)
 delete mode 100644 docs/fidesops/docs/guides/email_connectors.md

diff --git a/docs/fidesops/docs/guides/email_communications.md b/docs/fidesops/docs/guides/email_communications.md
index 08d3a35f1..8442a4d7d 100644
--- a/docs/fidesops/docs/guides/email_communications.md
+++ b/docs/fidesops/docs/guides/email_communications.md
@@ -1,12 +1,12 @@
-# Configure Email Communications
-## What is email used for?
+# Configure Automatic Emails
+## What is a fidesops Email Connection?
 
-Fidesops supports email server configurations for sending processing notices to privacy request subjects. Future updates will support outbound email communications with data processors.
+Fidesops supports configuring third party email servers to handle outbound communications.
 
 Supported modes of use:
 
-- Subject Identity Verification - for more information on identity verification in subject requests, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
-- Erasure Request Email Fulfillment - sends an email to configured third parties to process erasures for a given data subject.  See [Email Connectors](email_connectors.md) for more information.
+- Subject Identity Verification - sends a verification code to the user's email address prior to  for more information on identity verification in subject requests, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
+- Erasure Request Email Fulfillment - sends an email to configured third parties to process erasures for a given data subject.  See [creating Email Connectors](#create-an-email-connector) for more information.
 
 ## Prerequisites
 
@@ -16,12 +16,12 @@ Fidesops currently supports Mailgun for email integrations. Ensure you register
 
     Follow the [Mailgun documentation](https://documentation.mailgun.com/en/latest/api-intro.html#authentication-1) to create a new Domain Sending Key for fidesops. 
 
-    !!! Note 
-        Mailgun automatically generates a **primary account API key** when you sign up for an account. This key allows you to perform all CRUD operations via Mailgun's API endpoints, and for any of your sending domains. For security purposes, using a new **domain sending key** is recommended over your primary API key.
+!!! Note 
+    Mailgun automatically generates a **primary account API key** when you sign up for an account. This key allows you to perform all CRUD operations via Mailgun's API endpoints, and for any of your sending domains. For security purposes, using a new **domain sending key** is recommended over your primary API key.
 
 ## Configuration
 
-### Create the email configuration
+### Create the email config
 
 ```json title="<code>POST api/v1/email/config"
 {
@@ -47,7 +47,7 @@ Fidesops currently supports Mailgun for email integrations. Ensure you register
 
 ### Add the email configuration secrets 
 
-```json title="<code>POST api/v1/email/config/{{email_config_key}}/secret"
+```json title="<code>POST api/v1/email/config/{email_config_key}/secret"
 {
     "mailgun_api_key": "nc123849ycnpq98fnu"
 }
@@ -58,3 +58,102 @@ Fidesops currently supports Mailgun for email integrations. Ensure you register
 |---|----|
 | `mailgun_api_key` | Your Mailgun Domain Sending Key. |
 
+## Email third-party services
+
+Once your email server is configured, you can create an email connector to send automatic erasure requests to third-party services. Fidesops will gather details about each collection described in the connector, and send a single email to the service after all collections have been visited. 
+
+!!! Note
+    Fidesops does not collect confirmation that the erasure was completed by the third party.
+
+
+### Create the connector
+
+Ensure you have created your [email configuration](#configuration) prior to creating a new email connector.
+
+```json title="<code>PATCH api/v1/connection</code>"
+[
+  { 
+    "name": "Email Connection Config",
+    "key": "third_party_email_connector",
+    "connection_type": "email",
+    "access": "write"
+  }
+]
+```
+
+| Field | Description |
+|----|----|
+| `key` | A unique key used to manage your email connector. This is auto-generated from `name` if left blank. Accepted values are alphanumeric, `_`, and `.`. |
+| `name` | A unique user-friendly name for your email connector. |
+| `connection_type` | Must be `email` to create a new email connector. |
+| `access` | Email connectors must be given `write` access in order to send an email. |
+
+
+### Configure notifications
+
+Once your email connector is created, configure any outbound email addresses:
+
+```json title="<code>PUT api/v1/connection/{email_connection_config_key}/secret</code>" 
+{
+    "test_email": "my_email@example.com",
+    "to_email": "third_party@example.com
+}
+```
+
+| Field | Description |
+|----|----|
+| `{email_connection_config_key}` | The unique key that represents the email connection to use. |
+| `to_email` | The user that will be notified via email to complete an erasure request. *Only one `to_email` is supported at this time.* |
+| `test_email` | *Optional.* An email to which you have access for verifying your setup. If your email configuration is working, you will receive an email with mock data similar to the one sent to third-party services. |
+
+### Configure the dataset
+
+Lastly, configure the collections and fields you would like to request be erased or masked. Fidesops will use these fields to compose an email to the third-party service. 
+
+```json title="<code>PUT api/v1/connection/{email_connection_config_key}/dataset" 
+[
+    {
+      "fides_key": "email_dataset",
+      "name": "Dataset not accessible automatically",
+      "description": "Third party data - will email to request erasure",
+      "collections": [
+        {
+          "name": "daycare_customer",
+          "fields": [
+            {
+              "name": "id",
+              "data_categories": [
+                "system.operations"
+              ],
+              "fidesops_meta": {
+                "primary_key": true
+              }
+            },
+            {
+              "name": "child_health_concerns",
+              "data_categories": [
+                "user.biometric_health"
+              ]
+            },
+            {
+              "name": "user_email",
+              "data_categories": [
+                "user.contact.email"
+              ],
+              "fidesops_meta": {
+                "identity": "email"
+              }
+            }
+          ]
+        }
+      ]
+    }
+]
+```
+
+| Field | Description |
+|----|----|
+| `fides_key` | A unique key used to manage your email dataset. This is auto-generated from `name` if left blank. Accepted values are alphanumeric, `_`, and `.`. |
+| `name` | A unique user-friendly name for your email dataset. |
+| `description` | Any additional information used to describe this email dataset. |
+| `collections` | Any collections and associated fields belonging to the third party service, similar to a configured fidesops [Dataset](datasets.md). If you do not know the exact data structure of a third party's database, you can configure a single collection with the fields you would like masked. **Note:** A primary key must be specified on each collection. |
\ No newline at end of file
diff --git a/docs/fidesops/docs/guides/email_connectors.md b/docs/fidesops/docs/guides/email_connectors.md
deleted file mode 100644
index aaed647b0..000000000
--- a/docs/fidesops/docs/guides/email_connectors.md
+++ /dev/null
@@ -1,106 +0,0 @@
-# Setting up an Email Connector
-
-
-## What is the purpose of the Email Connector?
-
-The Email Connector is a ConnectionConfig type that emails a third party to ask them to complete
-a Privacy Request when their service cannot be accessed automatically. 
-
-Fidesops will gather details about each collection described on the third party service as part of request execution and 
-wait to send a single email to the service after all collections have been visited.  Fidesops does not 
-collect confirmation that the erasure was completed by the third party; the EmailConnector is only responsible 
-for notifying them.
-
-Importantly, only *erasure* requests are supported at this time for EmailConnectors.  
-
-
-## What pieces should be configured for an Email Connector?
-
-In short, you will need to create an `email` ConnectionConfig to store minor details about the third party service, and a DatasetConfig
-to describe the data contained in their datasource.  You will also need to have configured a separate [EmailConfig](email_communications.md) which
-is used system-wide to carry out the actual email send.
-
-
-## Setting up the ConnectionConfig
-
-Create a ConnectionConfig object with
-
-```json title="<code>PATCH api/v1/connection</code>"
-[
-  { 
-    "name": "Email Connection Config",
-    "key": "third_party_email_connector",
-    "connection_type": "email",
-    "access": "write"
-  }
-]
-```
-EmailConnectors must be given "write" access in order to send an email.
-
-
-## Configuring who to notify
-
-Save a `to_email` on the ConnectionConfig secrets.  This is the user that will be notified via email to complete
-an erasure. Only one `to_email` is supported at this time.
-
-Optionally, configure a `test_email` to which you have access, to verify that your setup is working.  Provided your 
-EmailConfig is set up properly, you should receive an email similar to the one sent to third-party services, containing
-dummy data.
-
-```json title="<code>PUT api/v1/connection/<email_connection_config_key>/secret</code>" 
-{
-    "test_email": "my_email@example.com",
-    "to_email": "third_party@example.com
-}
-```
-
-## Configuring the dataset
-
-Describe the collections and fields on a third party source with a [DatasetConfig](datasets.md), the same way you'd describe attributes 
-on a database.  If you do not know the exact data structure of a third party, you might configure a single collection
-with the fields you'd like masked.
-
-As with all collections that support erasures, a primary key must be specified on each collection.
-
-
-```json title="<code>PUT api/v1/connection/<email_connection_config_key>/dataset" 
-[
-    {
-      "fides_key": "email_dataset",
-      "name": "Dataset not accessible automatically",
-      "description": "Third party data - will email to request erasure",
-      "collections": [
-        {
-          "name": "daycare_customer",
-          "fields": [
-            {
-              "name": "id",
-              "data_categories": [
-                "system.operations"
-              ],
-              "fidesops_meta": {
-                "primary_key": true
-              }
-            },
-            {
-              "name": "child_health_concerns",
-              "data_categories": [
-                "user.biometric_health"
-              ]
-            },
-            {
-              "name": "user_email",
-              "data_categories": [
-                "user.contact.email"
-              ],
-              "fidesops_meta": {
-                "identity": "email"
-              }
-            }
-          ]
-        }
-      ]
-    }
-]
-```
-
diff --git a/docs/fidesops/docs/guides/privacy_requests.md b/docs/fidesops/docs/guides/privacy_requests.md
index 5b3f1fea3..ead5b2fc1 100644
--- a/docs/fidesops/docs/guides/privacy_requests.md
+++ b/docs/fidesops/docs/guides/privacy_requests.md
@@ -43,13 +43,13 @@ A full list of attributes available to set on the Privacy Request can be found i
 ## Subject Identity Verification 
 
 To have users verify their identity before their Privacy Request is executed, set the `subject_identity_verification_required` 
-variable in your `fidesops.toml` to `TRUE`. You must also set up an EmailConfig that lets Fidesops send automated emails 
+variable in your `fidesops.toml` to `TRUE`. You must also set up an [EmailConfig](./email_communications.md) that lets fidesops send automated emails 
 to your users.
 
-When a user submits a PrivacyRequest, they will be emailed a six-digit code.  They must supply that verification code to Fidesops
+When a user submits a PrivacyRequest, they will be emailed a six-digit code.  They must supply that verification code to fidesops
 to continue privacy request execution.  Until the Privacy Request identity is verified, it will have a status of: `identity_unverified`.
 
-```json title="<code>POST api/v1/privacy-request/<privacy_request_id>/verify</code>"
+```json title="<code>POST api/v1/privacy-request/{privacy_request_id}/verify</code>"
 {"code": "<verification code here>"}
 ```
 
diff --git a/docs/fidesops/mkdocs.yml b/docs/fidesops/mkdocs.yml
index 60f9784fe..08208849d 100644
--- a/docs/fidesops/mkdocs.yml
+++ b/docs/fidesops/mkdocs.yml
@@ -19,7 +19,6 @@ nav:
           - User Management: ui/user_management.md
   - How-To Guides:
       - View Available Connection Types: guides/connection_types.md
-      - Configure Email Communications: guides/email_communications.md
       - Annotate Complex Fields: guides/complex_fields.md
       - Configure Data Masking: guides/masking_strategies.md
       - Configure Storage Destinations: guides/storage.md
@@ -28,7 +27,7 @@ nav:
       - Configure OneTrust Integration: guides/onetrust.md
       - Preview Query Execution: guides/query_execution.md
       - Data Rights Protocol: guides/data_rights_protocol.md
-      - Configure Email Connectors: guides/email_connectors.md
+      - Configure Automatic Emails: guides/email_communications.md
   - SaaS Connectors:
       - Connect to SaaS Applications: saas_connectors/saas_connectors.md
       - SaaS Configuration: saas_connectors/saas_config.md

From 1e855d01e06a04b6a0f9da79639b02502f2b1e54 Mon Sep 17 00:00:00 2001
From: Cole <cole@ethyca.com>
Date: Tue, 6 Sep 2022 15:34:54 -0400
Subject: [PATCH 10/12] correct link, broken sentence

---
 docs/fidesops/docs/guides/email_communications.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/fidesops/docs/guides/email_communications.md b/docs/fidesops/docs/guides/email_communications.md
index 8442a4d7d..05600852a 100644
--- a/docs/fidesops/docs/guides/email_communications.md
+++ b/docs/fidesops/docs/guides/email_communications.md
@@ -5,8 +5,8 @@ Fidesops supports configuring third party email servers to handle outbound commu
 
 Supported modes of use:
 
-- Subject Identity Verification - sends a verification code to the user's email address prior to  for more information on identity verification in subject requests, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
-- Erasure Request Email Fulfillment - sends an email to configured third parties to process erasures for a given data subject.  See [creating Email Connectors](#create-an-email-connector) for more information.
+- Subject Identity Verification - sends a verification code to the user's email address prior to processing a subject request. for more information on identity verification, see the [Privacy Requests](privacy_requests.md#subject-identity-verification) guide.
+- Erasure Request Email Fulfillment - sends an email to configured third parties to process erasures for a given data subject.  See [creating email Connectors](#email-third-party-services) for more information.
 
 ## Prerequisites
 

From d70d71a1eeebbab4e8ed6f487a7ad41eb5c68575 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Tue, 6 Sep 2022 16:17:58 -0500
Subject: [PATCH 11/12] Create a new EmailRequestFulfillmentBodyParams type to
 be used once the cached email details are extracted by dataset.

---
 .../docs/guides/email_communications.md       |  2 +-
 docs/fidesops/docs/guides/reporting.md        | 10 +++---
 .../erasure_request_email_fulfillment.html    |  4 +--
 src/fidesops/ops/models/privacy_request.py    | 13 +++++--
 .../ops/service/connectors/email_connector.py | 13 ++++---
 .../service/email/email_dispatch_service.py   |  6 ++--
 .../test_integration_email.py                 |  8 +++--
 tests/ops/models/test_privacy_request.py      |  4 ++-
 .../request_runner_service_test.py            | 36 ++++++++++++++-----
 9 files changed, 65 insertions(+), 31 deletions(-)

diff --git a/docs/fidesops/docs/guides/email_communications.md b/docs/fidesops/docs/guides/email_communications.md
index 05600852a..4b82444e6 100644
--- a/docs/fidesops/docs/guides/email_communications.md
+++ b/docs/fidesops/docs/guides/email_communications.md
@@ -96,7 +96,7 @@ Once your email connector is created, configure any outbound email addresses:
 ```json title="<code>PUT api/v1/connection/{email_connection_config_key}/secret</code>" 
 {
     "test_email": "my_email@example.com",
-    "to_email": "third_party@example.com
+    "to_email": "third_party@example.com"
 }
 ```
 
diff --git a/docs/fidesops/docs/guides/reporting.md b/docs/fidesops/docs/guides/reporting.md
index f0081be29..e6399412e 100644
--- a/docs/fidesops/docs/guides/reporting.md
+++ b/docs/fidesops/docs/guides/reporting.md
@@ -222,7 +222,7 @@ To retrieve information to resume or retry a privacy request, the following endp
 
 ### Paused access request example
 
-The request below is in a `paused` state because we're waiting on manual input from the user to proceed. If we look at the `stopped_collection_details` key, we can see that the request
+The request below is in a `paused` state because we're waiting on manual input from the user to proceed. If we look at the `action_required_details` key, we can see that the request
 paused execution during the `access` step of the `manual_key:filing_cabinet` collection.  The `action_needed.locators` field shows the user they should
 fetch the record in the filing cabinet with a `customer_id` of `72909`, and pull the `authorized_user`, `customer_id`, `id`, and `payment_card_id` fields
 from that record.  These values should be manually uploaded to the `resume_endpoint`.  See the [Manual Data](https://ethyca.github.io/fidesops/guides/manual_data/#resuming-a-paused-access-privacy-request) 
@@ -237,7 +237,7 @@ guides for more information on resuming a paused access request.
             "created_at": "2022-06-06T20:12:28.809815+00:00",
             "started_processing_at": "2022-06-06T20:12:28.986462+00:00",
             ...,
-            "stopped_collection_details": {
+            "action_required_details": {
                 "step": "access",
                 "collection": "manual_key:filing_cabinet",
                 "action_needed": [
@@ -268,7 +268,7 @@ guides for more information on resuming a paused access request.
 
 ### Paused erasure request example
 
-The request below is in a `paused` state because we're waiting on the user to confirm they've masked the appropriate data before proceeding.  The `stopped_collection_details` shows us that the request
+The request below is in a `paused` state because we're waiting on the user to confirm they've masked the appropriate data before proceeding.  The `action_required_details` shows us that the request
 paused execution during the `erasure` step of the `manual_key:filing_cabinet` collection.  Looking at `action_needed.locators` field, we can
 see that the user should find the record in the filing cabinet with an `id` of 2, and replace its `authorized_user` with `None`. 
 A confirmation of the masked records count should be uploaded to the `resume_endpoint`  See the [Manual Data](https://ethyca.github.io/fidesops/guides/manual_data/#resuming-a-paused-erasure-privacy-request) 
@@ -284,7 +284,7 @@ guides for more information on resuming a paused erasure request.
             "finished_processing_at": null,
             "status": "paused",
             ...,
-            "stopped_collection_details": {
+            "action_required_details": {
                 "step": "erasure",
                 "collection": "manual_key:filing_cabinet",
                 "action_needed": [
@@ -325,7 +325,7 @@ After troubleshooting the issues with your postgres connection, you would resume
             "finished_processing_at": null,
             "status": "error",
             ...,
-            "stopped_collection_details": {
+            "action_required_details": {
               "step": "erasure",
               "collection": "postgres_dataset:payment_card",
               "action_needed": null
diff --git a/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html b/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html
index 667f88a0a..4a6d6febf 100644
--- a/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html
+++ b/src/fidesops/ops/email_templates/templates/erasure_request_email_fulfillment.html
@@ -7,8 +7,8 @@
    <body>
       <main>
          <p>Please locate and erase the associated records from the following collections:</p>
-         {% for collection, action_required in dataset_collection_action_required.items() -%}
-         <p><b>{{ collection }}</b></p>
+         {% for collection_address, action_required in dataset_collection_action_required.items() -%}
+         <p><b>{{ collection_address.collection }}</b></p>
          {% for action in action_required.action_needed -%}
          <p>Locate the relevant records with:</p>
          <ul>
diff --git a/src/fidesops/ops/models/privacy_request.py b/src/fidesops/ops/models/privacy_request.py
index 0f3b1652e..72e2401b6 100644
--- a/src/fidesops/ops/models/privacy_request.py
+++ b/src/fidesops/ops/models/privacy_request.py
@@ -107,6 +107,11 @@ class Config:
         arbitrary_types_allowed = True
 
 
+EmailRequestFulfillmentBodyParams = Dict[
+    CollectionAddress, Optional[CheckpointActionRequired]
+]
+
+
 class PrivacyRequestStatus(str, EnumType):
     """Enum for privacy request statuses, reflecting where they are in the Privacy Request Lifecycle"""
 
@@ -389,14 +394,18 @@ def cache_email_connector_template_contents(
 
     def get_email_connector_template_contents_by_dataset(
         self, step: CurrentStep, dataset: str
-    ) -> Dict[str, Optional[CheckpointActionRequired]]:
+    ) -> EmailRequestFulfillmentBodyParams:
         """Retrieve the raw details to populate an email template for collections on a given dataset."""
         cache: FidesopsRedis = get_cache()
         email_contents: Dict[str, Optional[Any]] = cache.get_encoded_objects_by_prefix(
             f"EMAIL_INFORMATION__{self.id}__{step.value}__{dataset}"
         )
         return {
-            k.split("__")[-1]: CheckpointActionRequired.parse_obj(v) if v else None
+            CollectionAddress(
+                k.split("__")[-2], k.split("__")[-1]
+            ): CheckpointActionRequired.parse_obj(v)
+            if v
+            else None
             for k, v in email_contents.items()
         }
 
diff --git a/src/fidesops/ops/service/connectors/email_connector.py b/src/fidesops/ops/service/connectors/email_connector.py
index 5731e8f0b..009eb5ec3 100644
--- a/src/fidesops/ops/service/connectors/email_connector.py
+++ b/src/fidesops/ops/service/connectors/email_connector.py
@@ -19,6 +19,7 @@
 from fidesops.ops.models.policy import CurrentStep, Policy, Rule
 from fidesops.ops.models.privacy_request import (
     CheckpointActionRequired,
+    EmailRequestFulfillmentBodyParams,
     ManualAction,
     PrivacyRequest,
 )
@@ -61,7 +62,9 @@ def test_connection(self) -> Optional[ConnectionTestStatus]:
                 action_type=EmailActionType.EMAIL_ERASURE_REQUEST_FULFILLMENT,
                 to_email=config.test_email,
                 email_body_params={
-                    "test_collection": CheckpointActionRequired(
+                    CollectionAddress(
+                        "test_dataset", "test_collection"
+                    ): CheckpointActionRequired(
                         step=CurrentStep.erasure,
                         collection=CollectionAddress("test_dataset", "test_collection"),
                         action_needed=[
@@ -168,10 +171,10 @@ def email_connector_erasure_send(db: Session, privacy_request: PrivacyRequest) -
         ConnectionConfig.connection_type == ConnectionType.email,
     )
     for ds, cc in email_dataset_configs:
-        template_values: Dict[
-            str, Optional[CheckpointActionRequired]
-        ] = privacy_request.get_email_connector_template_contents_by_dataset(
-            CurrentStep.erasure, ds.dataset.get("fides_key")
+        template_values: EmailRequestFulfillmentBodyParams = (
+            privacy_request.get_email_connector_template_contents_by_dataset(
+                CurrentStep.erasure, ds.dataset.get("fides_key")
+            )
         )
 
         if not template_values:
diff --git a/src/fidesops/ops/service/email/email_dispatch_service.py b/src/fidesops/ops/service/email/email_dispatch_service.py
index b098aa0dc..d2b44235a 100644
--- a/src/fidesops/ops/service/email/email_dispatch_service.py
+++ b/src/fidesops/ops/service/email/email_dispatch_service.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, Optional, Union
+from typing import Any, Optional, Union
 
 import requests
 from requests import Response
@@ -8,7 +8,7 @@
 from fidesops.ops.common_exceptions import EmailDispatchException
 from fidesops.ops.email_templates import get_email_template
 from fidesops.ops.models.email import EmailConfig
-from fidesops.ops.models.privacy_request import CheckpointActionRequired
+from fidesops.ops.models.privacy_request import EmailRequestFulfillmentBodyParams
 from fidesops.ops.schemas.email.email import (
     EmailActionType,
     EmailForActionType,
@@ -28,7 +28,7 @@ def dispatch_email(
     to_email: Optional[str],
     email_body_params: Union[
         SubjectIdentityVerificationBodyParams,
-        Dict[str, Optional[CheckpointActionRequired]],
+        EmailRequestFulfillmentBodyParams,
     ],
 ) -> None:
     if not to_email:
diff --git a/tests/ops/integration_tests/test_integration_email.py b/tests/ops/integration_tests/test_integration_email.py
index 909f57b66..41c3a18da 100644
--- a/tests/ops/integration_tests/test_integration_email.py
+++ b/tests/ops/integration_tests/test_integration_email.py
@@ -110,7 +110,7 @@ async def test_email_connector_cache_and_delayed_send(
     )
 
     assert raw_email_template_values == {
-        "children": CheckpointActionRequired(
+        CollectionAddress("email_dataset", "children"): CheckpointActionRequired(
             step=CurrentStep.erasure,
             collection=CollectionAddress("email_dataset", "children"),
             action_needed=[
@@ -131,7 +131,9 @@ async def test_email_connector_cache_and_delayed_send(
                 )
             ],
         ),
-        "daycare_customer": CheckpointActionRequired(
+        CollectionAddress(
+            "email_dataset", "daycare_customer"
+        ): CheckpointActionRequired(
             step=CurrentStep.erasure,
             collection=CollectionAddress("email_dataset", "daycare_customer"),
             action_needed=[
@@ -144,7 +146,7 @@ async def test_email_connector_cache_and_delayed_send(
                 )
             ],
         ),
-        "payment": CheckpointActionRequired(
+        CollectionAddress("email_dataset", "payment"): CheckpointActionRequired(
             step=CurrentStep.erasure,
             collection=CollectionAddress("email_dataset", "payment"),
             action_needed=[
diff --git a/tests/ops/models/test_privacy_request.py b/tests/ops/models/test_privacy_request.py
index 338cf486e..9a4950392 100644
--- a/tests/ops/models/test_privacy_request.py
+++ b/tests/ops/models/test_privacy_request.py
@@ -506,7 +506,9 @@ def test_cache_template_contents(self, privacy_request):
         assert privacy_request.get_email_connector_template_contents_by_dataset(
             CurrentStep.erasure, "email_dataset"
         ) == {
-            "test_collection": CheckpointActionRequired(
+            CollectionAddress(
+                "email_dataset", "test_collection"
+            ): CheckpointActionRequired(
                 step=CurrentStep.erasure,
                 collection=CollectionAddress("email_dataset", "test_collection"),
                 action_needed=[
diff --git a/tests/ops/service/privacy_request/request_runner_service_test.py b/tests/ops/service/privacy_request/request_runner_service_test.py
index 952c27325..c8337b27b 100644
--- a/tests/ops/service/privacy_request/request_runner_service_test.py
+++ b/tests/ops/service/privacy_request/request_runner_service_test.py
@@ -16,6 +16,7 @@
     PrivacyRequestPaused,
 )
 from fidesops.ops.core.config import config
+from fidesops.ops.graph.config import CollectionAddress
 from fidesops.ops.models.connectionconfig import AccessLevel
 from fidesops.ops.models.email import EmailConfig
 from fidesops.ops.models.policy import CurrentStep, PolicyPostWebhook
@@ -1647,9 +1648,9 @@ def test_create_and_process_erasure_request_email_connector_email_send_error(
             CurrentStep.erasure, "email_dataset"
         )
         assert set(cached_email_contents.keys()) == {
-            "payment",
-            "children",
-            "daycare_customer",
+            CollectionAddress("email_dataset", "payment"),
+            CollectionAddress("email_dataset", "children"),
+            CollectionAddress("email_dataset", "daycare_customer"),
         }
         pr.delete(db=db)
         assert mailgun_send.called is False
@@ -1747,13 +1748,30 @@ def test_email_connector_no_updates_needed(
             CurrentStep.erasure, "email_dataset"
         )
         assert set(cached_email_contents.keys()) == {
-            "payment",
-            "children",
-            "daycare_customer",
+            CollectionAddress("email_dataset", "payment"),
+            CollectionAddress("email_dataset", "children"),
+            CollectionAddress("email_dataset", "daycare_customer"),
         }
-        assert cached_email_contents["payment"].action_needed[0].update is None
-        assert cached_email_contents["children"].action_needed[0].update is None
-        assert cached_email_contents["daycare_customer"].action_needed[0].update is None
+        assert (
+            cached_email_contents[CollectionAddress("email_dataset", "payment")]
+            .action_needed[0]
+            .update
+            is None
+        )
+        assert (
+            cached_email_contents[CollectionAddress("email_dataset", "children")]
+            .action_needed[0]
+            .update
+            is None
+        )
+        assert (
+            cached_email_contents[
+                CollectionAddress("email_dataset", "daycare_customer")
+            ]
+            .action_needed[0]
+            .update
+            is None
+        )
 
         pr.delete(db=db)
         assert (

From 9a20d02412b96967f1934ceb25c7e32e3650beb1 Mon Sep 17 00:00:00 2001
From: Dawn Pattison <pattisdr@users.noreply.github.com>
Date: Tue, 6 Sep 2022 16:45:43 -0500
Subject: [PATCH 12/12] Fix missed test.

---
 .../ops/api/v1/endpoints/test_connection_config_endpoints.py  | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py b/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
index e7e17ff40..816eabefa 100644
--- a/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_connection_config_endpoints.py
@@ -1199,7 +1199,9 @@ def test_put_email_connection_config_secrets(
         )
         assert kwargs["to_email"] == "test@example.com"
         assert kwargs["email_body_params"] == {
-            "test_collection": CheckpointActionRequired(
+            CollectionAddress(
+                "test_dataset", "test_collection"
+            ): CheckpointActionRequired(
                 step=CurrentStep.erasure,
                 collection=CollectionAddress("test_dataset", "test_collection"),
                 action_needed=[