WIP add legal bases to purposes

src/fides/api/schemas/tcf.py

@@ -35,6 +35,7 @@ class EmbeddedVendor(FidesSchema):
 class TCFPurposeRecord(MappedPurpose, TCFSavedandServedDetails):
     """Schema for a TCF Purpose or a Special Purpose: returned in the TCF Overlay Experience"""
 
+    legal_bases: List[str] = []
     vendors: List[
         EmbeddedVendor
     ] = []  # Vendors that use this purpose or special purpose
@@ -52,6 +53,10 @@ class EmbeddedLineItem(FidesSchema):
     name: str
 
 
+class EmbeddedPurpose(EmbeddedLineItem):
+    legal_bases: List[str] = []
+
+
 class TCFDataCategoryRecord(FidesSchema):
     """Details for data categories on systems: Read-only"""
 
@@ -69,8 +74,8 @@ class TCFVendorRecord(TCFSavedandServedDetails):
     has_vendor_id: bool
     name: Optional[str]
     description: Optional[str]
-    purposes: List[EmbeddedLineItem] = []
-    special_purposes: List[EmbeddedLineItem] = []
+    purposes: List[EmbeddedPurpose] = []
+    special_purposes: List[EmbeddedPurpose] = []
     data_categories: List[TCFDataCategoryRecord] = []
     features: List[EmbeddedLineItem] = []
     special_features: List[EmbeddedLineItem] = []

src/fides/api/util/tcf_util.py

@@ -11,7 +11,7 @@
 )
 from fideslang.gvl.models import Feature, Purpose
 from sqlalchemy import case, func
-from sqlalchemy.dialects.postgresql import array_agg
+from sqlalchemy.dialects.postgresql import array, array_agg
 from sqlalchemy.engine import Row
 from sqlalchemy.orm import Query, Session
 
@@ -88,7 +88,16 @@ def get_tcf_component_and_vendors(
             System.name,
             System.description,
             System.vendor_id,
-            array_agg(PrivacyDeclaration.data_use).label("data_uses"),
+            array_agg(
+                array(
+                    [
+                        PrivacyDeclaration.data_use,
+                        PrivacyDeclaration.legal_basis_for_processing,
+                    ]
+                )
+            ).label(
+                "data_uses_and_legal_bases"
+            ),  # A list of lists with a data use and a legal basis for each privacy declaration
             array_agg(
                 case(
                     [
@@ -112,8 +121,9 @@ def get_tcf_component_and_vendors(
 
     tcf_record_type: Union[Type[TCFPurposeRecord], Type[TCFFeatureRecord]]
     get_tcf_construct: Callable
+    is_purpose_type: bool = tcf_component_name in ["purposes", "special_purposes"]
 
-    if tcf_component_name in ["purposes", "special_purposes"]:
+    if is_purpose_type:
         matching_systems = matching_systems.filter(
             PrivacyDeclaration.data_use.in_(relevant_uses_or_features)
         )
@@ -137,13 +147,16 @@ def get_tcf_component_and_vendors(
                 id=system_identifier,
                 name=record.name,
                 description=record.description,
-                has_vendor_id=bool(vendor_id),
+                has_vendor_id=bool(
+                    vendor_id
+                ),  # This will later let us separate TCF components
+                # into vendors and systems.
             )
 
-        # Pull the attributes we care about from the system record depending on the TCF component.
+        # Pull the attributes we care about from the system record depending on the type TCF component.
         relevant_system_attributes: Set[str] = (
-            record.data_uses
-            if tcf_component_name in ["purposes", "special_purposes"]
+            [data_use[0] for data_use in record.data_uses_and_legal_bases]
+            if is_purpose_type
             else get_system_features(
                 record, relevant_features=relevant_uses_or_features
             )
@@ -155,12 +168,20 @@ def get_tcf_component_and_vendors(
             if not fideslang_gvl_record:
                 continue
 
-            if fideslang_gvl_record.id not in matching_record_map:
-                # Collect relevant TCF component
-                matching_record_map[fideslang_gvl_record.id] = tcf_record_type(
-                    **fideslang_gvl_record.dict()
+            # Collect relevant TCF component
+            tcf_record = tcf_record_type(**fideslang_gvl_record.dict())
+            if is_purpose_type:
+                tcf_record.legal_bases = list(
+                    {
+                        use[1]
+                        for use in record.data_uses_and_legal_bases
+                        if use[1] is not None and use[0] in tcf_record.data_uses
+                    }
                 )
 
+            if fideslang_gvl_record.id not in matching_record_map:
+                matching_record_map[fideslang_gvl_record.id] = tcf_record
+
             # Embed the systems information beneath the TCF component data
             embedded_system_or_vendor_record = (
                 matching_record_map[fideslang_gvl_record.id].vendors
@@ -181,9 +202,9 @@ def get_tcf_component_and_vendors(
                 system_map[system_identifier], tcf_component_name
             )
             if fideslang_gvl_record.id not in [
-                tcf_record.id for tcf_record in system_tcf_subsection
+                tcf_sub_record.id for tcf_sub_record in system_tcf_subsection
             ]:
-                system_tcf_subsection.append(fideslang_gvl_record)
+                system_tcf_subsection.append(tcf_record)
 
     return matching_record_map, system_map
 

tests/fixtures/application_fixtures.py

@@ -2665,6 +2665,7 @@ def tcf_system(db: Session) -> System:
             "data_qualifier": "aggregated.anonymized.unlinked_pseudonymized.pseudonymized.identified",
             "data_subjects": ["customer"],
             "dataset_references": None,
+            "legal_basis_for_processing": "Consent",
             "egress": None,
             "ingress": None,
         },
@@ -2680,6 +2681,7 @@ def tcf_system(db: Session) -> System:
             "data_qualifier": "aggregated.anonymized.unlinked_pseudonymized.pseudonymized.identified",
             "data_subjects": ["customer"],
             "dataset_references": None,
+            "legal_basis_for_processing": "Legal obligations",
             "egress": None,
             "ingress": None,
         },

tests/ops/util/test_tcf_util.py

@@ -1,8 +1,10 @@
 import pytest
+from fideslang import GVL_PURPOSES, MAPPED_PURPOSES
+from fideslang.models import LegalBasisForProcessingEnum
 
+from fides.api.models.sql_models import PrivacyDeclaration
 from fides.api.schemas.tcf import EmbeddedVendor
 from fides.api.util.tcf_util import get_tcf_contents
-from tests.fixtures.saas.connection_template_fixtures import instantiate_connector
 
 
 class TestTCFContents:
@@ -128,6 +130,7 @@ def test_system_matches_subset_of_purpose_data_uses(self, db, tcf_system):
         for i, decl in enumerate(tcf_system.privacy_declarations):
             if i:
                 decl.data_use = "marketing.advertising.first_party.contextual"
+                decl.legal_basis_for_processing = "Consent"
                 tcf_system.privacy_declarations[0].save(db)
             else:
                 decl.delete(db)
@@ -141,6 +144,7 @@ def test_system_matches_subset_of_purpose_data_uses(self, db, tcf_system):
             "marketing.advertising.frequency_capping",
             "marketing.advertising.negative_targeting",
         ]
+        assert tcf_contents.tcf_purposes[0].legal_bases == ["Consent"]
         assert tcf_contents.tcf_purposes[0].vendors == [
             EmbeddedVendor(id="sendgrid", name="TCF System Test")
         ]
@@ -149,6 +153,7 @@ def test_system_matches_subset_of_purpose_data_uses(self, db, tcf_system):
         assert tcf_contents.tcf_vendors[0].id == "sendgrid"
         assert len(tcf_contents.tcf_vendors[0].purposes) == 1
         assert tcf_contents.tcf_vendors[0].purposes[0].id == 2
+        assert tcf_contents.tcf_vendors[0].purposes[0].legal_bases == ["Consent"]
         assert tcf_contents.tcf_features == []
         assert tcf_contents.tcf_special_features == []
 
@@ -163,6 +168,7 @@ def test_special_purposes(self, db):
             tcf_contents.tcf_special_purposes[0].name
             == "Ensure security, prevent and detect fraud, and fix errors\n"
         )
+        assert tcf_contents.tcf_special_purposes[0].legal_bases == ["Legal obligations"]
         assert tcf_contents.tcf_special_purposes[0].vendors == [
             EmbeddedVendor(id="sendgrid", name="TCF System Test")
         ]
@@ -173,8 +179,13 @@ def test_special_purposes(self, db):
         assert tcf_contents.tcf_vendors[0].id == "sendgrid"
         assert len(tcf_contents.tcf_vendors[0].purposes) == 1
         assert tcf_contents.tcf_vendors[0].purposes[0].id == 8
+        assert tcf_contents.tcf_vendors[0].purposes[0].legal_bases == ["Consent"]
+
         assert len(tcf_contents.tcf_vendors[0].special_purposes) == 1
         assert tcf_contents.tcf_vendors[0].special_purposes[0].id == 1
+        assert tcf_contents.tcf_vendors[0].special_purposes[0].legal_bases == [
+            "Legal obligations"
+        ]
         assert tcf_contents.tcf_features == []
         assert tcf_contents.tcf_special_features == []
 
@@ -234,3 +245,44 @@ def test_special_features(self, db, system):
         assert len(tcf_contents.tcf_vendors[0].special_features) == 1
         assert tcf_contents.tcf_vendors[0].special_features[0].id == 2
         assert len(tcf_contents.tcf_vendors[0].features) == 0
+
+    def test_system_with_multiple_privacy_declarations(self, db, system):
+        get_tcf_contents.cache_clear()
+
+        legal_bases = [item.value for item in LegalBasisForProcessingEnum]
+
+        for i, legal_basis in enumerate(legal_bases):
+            gvl_purpose = list(MAPPED_PURPOSES.values())[i]
+
+            PrivacyDeclaration.create(
+                db=db,
+                data={
+                    "name": gvl_purpose.name,
+                    "system_id": system.id,
+                    "data_categories": ["user.device.cookie_id"],
+                    "data_use": gvl_purpose.data_uses[0],
+                    "data_qualifier": "aggregated.anonymized.unlinked_pseudonymized.pseudonymized.identified",
+                    "data_subjects": ["customer"],
+                    "legal_basis_for_processing": legal_basis,
+                },
+            )
+
+        tcf_contents = get_tcf_contents(db)
+        assert len(tcf_contents.tcf_purposes) == 6
+        for i, purpose in enumerate(tcf_contents.tcf_purposes):
+            assert purpose.legal_bases == [legal_bases[i]]
+            assert purpose.id == list(MAPPED_PURPOSES.values())[i].id
+        assert (
+            len(tcf_contents.tcf_vendors) == 0
+        )  # No official vendor id on this system
+
+        assert len(tcf_contents.tcf_systems) == 0
+        system_info = tcf_contents.tcf_systems[0]
+
+        assert len(system_info.special_purposes) == 0
+        assert len(system_info.features) == 0
+        assert len(system_info.special_features) == 0
+        assert len(tcf_contents.tcf_systems[0].purposes) == 6
+        for i, embedded_purpose in enumerate(tcf_contents.tcf_systems[0].purposes):
+            assert embedded_purpose.legal_bases == [legal_bases[i]]
+            assert embedded_purpose.id == list(MAPPED_PURPOSES.values())[i].id