Changes to support consent signals

src/fides/api/models/connectionconfig.py

@@ -166,7 +166,7 @@ class ConnectionConfig(Base):
     system = relationship(System, back_populates="connection_configs", uselist=False)
 
     consent_automation: RelationshipProperty[Optional[ConsentAutomation]] = (
-        relationship(ConsentAutomation, cascade="all, delete-orphan")
+        relationship(ConsentAutomation, uselist=False, cascade="all, delete-orphan")
     )
 
     # Identifies the privacy actions needed from this connection by the associated system.

src/fides/api/schemas/consentable_item.py

@@ -1,8 +1,9 @@
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 
-from pydantic import Field
+from pydantic import BaseModel, Field
 
 from fides.api.models.consent_automation import ConsentableItem as ConsentableItemModel
+from fides.api.models.privacy_notice import UserConsentPreference
 from fides.api.schemas.base_class import FidesSchema
 
 
@@ -75,3 +76,17 @@
         for item in consentable_items
         if item.parent_id is None
     ]
+
+
+class ConsentWebhookResult(BaseModel):
+    """
+    A wrapper class for the identity map and notice map values returned from a `PROCESS_CONSENT_WEBHOOK` function.
+    """
+
+    identity_map: Dict[str, Any] = {}
+    notice_map: Dict[str, UserConsentPreference] = {}
+
+    @property
+    def success(self) -> bool:
+        """Returns true if both the identity map and notice map are not empty."""
+        return bool(self.identity_map) and bool(self.notice_map)

src/fides/api/service/saas_request/saas_request_override_factory.py

@@ -8,7 +8,7 @@
     InvalidSaaSRequestOverrideException,
     NoSuchSaaSRequestOverrideException,
 )
-from fides.api.schemas.consentable_item import ConsentableItem
+from fides.api.schemas.consentable_item import ConsentableItem, ConsentWebhookResult
 from fides.api.util.collection_util import Row
 
 
@@ -31,7 +31,7 @@
 
 
 RequestOverrideFunction = Callable[
-    ..., Union[List[Row], List[ConsentableItem], int, bool, None]
+    ..., Union[ConsentWebhookResult, List[Row], List[ConsentableItem], int, bool, None]
 ]
 
 
@@ -244,7 +244,15 @@
 
 
 def validate_process_consent_webhook_function(f: Callable) -> None:
-    pass
+    sig: Signature = signature(f)
+    if sig.return_annotation is not ConsentWebhookResult:
+        raise InvalidSaaSRequestOverrideException(
+            "Provided SaaS process consent webhook function must return a ConsentWebhookResult"
+        )
+    if len(sig.parameters) < 4:
+        raise InvalidSaaSRequestOverrideException(
+            "Provided SaaS process consent webhook function must declare at least 4 parameters"
+        )
 
 
 # TODO: Avoid running this on import?

tests/ops/models/test_consent_automation.py

@@ -1,12 +1,15 @@
 from sqlalchemy import and_
 from sqlalchemy.orm import Session
 
+from fides.api.models.connectionconfig import ConnectionConfig
 from fides.api.models.consent_automation import ConsentableItem, ConsentAutomation
 
 
 class TestConsentAutomation:
 
-    def test_create_consent_automation(self, db: Session, connection_config):
+    def test_create_consent_automation(
+        self, db: Session, connection_config: ConnectionConfig
+    ):
         consentable_items = [
             {
                 "type": "Channel",
@@ -36,6 +39,10 @@ def test_create_consent_automation(self, db: Session, connection_config):
         assert consent_automation.connection_config_id == connection_config.id
         assert len(consent_automation.consentable_items) == 2
 
+        # test link from connection_config
+        db.refresh(connection_config)
+        assert len(connection_config.consent_automation.consentable_items) == 2
+
     def test_update_consent_automation_add_consentable_items(
         self, db: Session, connection_config, privacy_notice
     ):

tests/ops/service/saas_request/test_saas_request_override_factory.py

@@ -12,6 +12,7 @@
 from fides.api.models.policy import Policy
 from fides.api.models.privacy_notice import UserConsentPreference
 from fides.api.models.privacy_request import PrivacyRequest
+from fides.api.schemas.consentable_item import ConsentWebhookResult
 from fides.api.service.connectors.saas.authenticated_client import AuthenticatedClient
 from fides.api.service.saas_request.saas_request_override_factory import (
     SaaSRequestOverrideFactory,
@@ -101,6 +102,19 @@ def valid_consent_update_override(
     return True
 
 
+def valid_process_consent_webhook_override(
+    client: AuthenticatedClient,
+    secrets: Dict[str, Any],
+    payload: Any,
+    notice_id_to_preference_map: Dict[str, UserConsentPreference],
+    consentable_items: List[ConsentableItem],
+) -> ConsentWebhookResult:
+    """
+    A sample override function for process consent webhook requests with a valid function signature
+    """
+    return ConsentWebhookResult()
+
+
 @pytest.mark.unit_saas
 class TestSaasRequestOverrideFactory:
     """
@@ -213,6 +227,23 @@ def test_register_update_consent_override(self):
             SaaSRequestOverrideFactory.get_override(f_id, SaaSRequestType.READ)
         assert f"Custom SaaS override '{f_id}' does not exist." in str(exc.value)
 
+    def test_register_process_consent_webhook_override(self):
+        """
+        Test registering a valid `process_consent_webhook` override function
+        """
+
+        f_id = uuid()
+        register(f_id, SaaSRequestType.PROCESS_CONSENT_WEBHOOK)(
+            valid_process_consent_webhook_override
+        )
+        assert valid_consent_update_override == SaaSRequestOverrideFactory.get_override(
+            f_id, SaaSRequestType.PROCESS_CONSENT_WEBHOOK
+        )
+
+        with pytest.raises(NoSuchSaaSRequestOverrideException) as exc:
+            SaaSRequestOverrideFactory.get_override(f_id, SaaSRequestType.READ)
+        assert f"Custom SaaS override '{f_id}' does not exist." in str(exc.value)
+
     def test_reregister_override(self):
         """
         Test that registering a new override with the same ID and same request type