From e7a3133f46062637ec3e2ea83ac2b1de049810cc Mon Sep 17 00:00:00 2001 From: Dawn Pattison Date: Tue, 22 Aug 2023 14:00:03 -0700 Subject: [PATCH] WIP add legal bases to purposes --- src/fides/api/schemas/tcf.py | 9 ++++- src/fides/api/util/tcf_util.py | 47 +++++++++++++++------- tests/fixtures/application_fixtures.py | 2 + tests/ops/util/test_tcf_util.py | 54 +++++++++++++++++++++++++- 4 files changed, 96 insertions(+), 16 deletions(-) diff --git a/src/fides/api/schemas/tcf.py b/src/fides/api/schemas/tcf.py index 8123b9faef..4a67518867 100644 --- a/src/fides/api/schemas/tcf.py +++ b/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] = [] diff --git a/src/fides/api/util/tcf_util.py b/src/fides/api/util/tcf_util.py index c7e4d44024..3fa1c0efb6 100644 --- a/src/fides/api/util/tcf_util.py +++ b/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 diff --git a/tests/fixtures/application_fixtures.py b/tests/fixtures/application_fixtures.py index 69cdc315eb..0912ba5946 100644 --- a/tests/fixtures/application_fixtures.py +++ b/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, }, diff --git a/tests/ops/util/test_tcf_util.py b/tests/ops/util/test_tcf_util.py index bd557c310e..13b33650ff 100644 --- a/tests/ops/util/test_tcf_util.py +++ b/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