From 125b18ed996c3ec4c38c8d17c98ba6734fc0943f Mon Sep 17 00:00:00 2001 From: Adam Sachs Date: Fri, 28 Apr 2023 10:09:10 -0700 Subject: [PATCH 1/5] initial cut at refactored privacy declaration upsert to maintain constant ids --- src/fides/api/ctl/routes/system.py | 40 ++++++++++++++++++++++++------ 1 file changed, 33 insertions(+), 7 deletions(-) diff --git a/src/fides/api/ctl/routes/system.py b/src/fides/api/ctl/routes/system.py index 75676a39a6a..e99f8b9bb96 100644 --- a/src/fides/api/ctl/routes/system.py +++ b/src/fides/api/ctl/routes/system.py @@ -224,16 +224,42 @@ async def upsert_privacy_declarations( ) -> None: """Helper to handle the specific upsert logic for privacy declarations""" - async with db.begin(): - # clear out any existing privacy declarations on the System - for existing_declaration in system.privacy_declarations: - await db.delete(existing_declaration) + def privacy_declaration_logical_id( + privacy_declaration: PrivacyDeclaration, + ) -> str: + """ + Internal helper to standardize a logical 'id' for privacy declarations. + As of now, this is based on the `data_use` and the `name` of the declaration, if provided. + """ + return f"{privacy_declaration.data_use}:{privacy_declaration.name or ''}" - # iterate through declarations specified on the request and create them + async with db.begin(): + # map existing declarations by their logical identifier + existing_declarations: Dict[str, PrivacyDeclaration] = { + privacy_declaration_logical_id(existing_declaration): existing_declaration + for existing_declaration in system.privacy_declarations + } + + # iterate through declarations specified on the request and upsert + # looking for "matching" existing declarations based on data_use and name for privacy_declaration in resource.privacy_declarations: + # prepare our 'payload' for either create or update data = privacy_declaration.dict() - data["system_id"] = system.id # add FK back to system - PrivacyDeclaration.create(db, data=data) + data["system_id"] = system.id # include FK back to system + + # if we find matching declaration, remove it from our map + if existing_declaration := existing_declarations.pop( + privacy_declaration_logical_id(privacy_declaration), None + ): + # and update existing declaration *in place* + existing_declaration.update(db, data=data) + else: + # otherwise, create a new declaration record + PrivacyDeclaration.create(db, data=data) + + # delete any existing privacy declarations that have not been "matched" in the request + for existing_declarations in existing_declarations.values(): + await db.delete(existing_declarations) async def update_system(resource: SystemSchema, db: AsyncSession) -> Dict: From 64a6b906214bb83f520cf7b0872503ad94a839cc Mon Sep 17 00:00:00 2001 From: Adam Sachs Date: Fri, 28 Apr 2023 12:29:22 -0700 Subject: [PATCH 2/5] add validation to prevent duplicate privacy declarations in system update request --- src/fides/api/ctl/routes/system.py | 43 +++++---- tests/ctl/core/test_api.py | 137 +++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+), 17 deletions(-) diff --git a/src/fides/api/ctl/routes/system.py b/src/fides/api/ctl/routes/system.py index e99f8b9bb96..5d94729cafd 100644 --- a/src/fides/api/ctl/routes/system.py +++ b/src/fides/api/ctl/routes/system.py @@ -67,15 +67,15 @@ def get_system(db: Session, fides_key: str) -> System: return system -async def validate_privacy_declarations_data_uses( - db: AsyncSession, system: SystemSchema -) -> None: +async def validate_privacy_declarations(db: AsyncSession, system: SystemSchema) -> None: """ - Ensure that the `PrivacyDeclaration`s on the provided `System` resource reference - valid `DataUse` records. + Ensure that the `PrivacyDeclaration`s on the provided `System` resource are valid: + - that they reference valid `DataUse` records + - that there are not "duplicate" `PrivacyDeclaration`s as defined by their "logical ID" If not, a `400` is raised """ + logical_ids = set() for privacy_declaration in system.privacy_declarations: try: await get_resource( @@ -88,6 +88,14 @@ async def validate_privacy_declarations_data_uses( status_code=HTTP_400_BAD_REQUEST, detail=f"Invalid privacy declaration referencing unknown DataUse {privacy_declaration.data_use}", ) + logical_id = privacy_declaration_logical_id(privacy_declaration) + if logical_id in logical_ids: + raise HTTPException( + status_code=HTTP_400_BAD_REQUEST, + detail=f"Duplicate privacy declarations specified with data use {privacy_declaration.data_use}", + ) + + logical_ids.add(logical_id) @system_connections_router.get( @@ -162,7 +170,7 @@ async def update( Update a System by the fides_key extracted from the request body. Defined outside of the crud routes to add additional "system manager" permission checks. """ - await validate_privacy_declarations_data_uses(db, resource) + await validate_privacy_declarations(db, resource) return await update_system(resource, db) @@ -174,7 +182,7 @@ async def upsert_system( updated = 0 # first pass to validate privacy declarations before proceeding for resource in resources: - await validate_privacy_declarations_data_uses(db, resource) + await validate_privacy_declarations(db, resource) for resource in resources: try: @@ -219,20 +227,21 @@ async def upsert( } +def privacy_declaration_logical_id( + privacy_declaration: PrivacyDeclaration, +) -> str: + """ + Helper to standardize a logical 'id' for privacy declarations. + As of now, this is based on the `data_use` and the `name` of the declaration, if provided. + """ + return f"{privacy_declaration.data_use}:{privacy_declaration.name or ''}" + + async def upsert_privacy_declarations( db: AsyncSession, resource: SystemSchema, system: System ) -> None: """Helper to handle the specific upsert logic for privacy declarations""" - def privacy_declaration_logical_id( - privacy_declaration: PrivacyDeclaration, - ) -> str: - """ - Internal helper to standardize a logical 'id' for privacy declarations. - As of now, this is based on the `data_use` and the `name` of the declaration, if provided. - """ - return f"{privacy_declaration.data_use}:{privacy_declaration.name or ''}" - async with db.begin(): # map existing declarations by their logical identifier existing_declarations: Dict[str, PrivacyDeclaration] = { @@ -348,7 +357,7 @@ async def create( Override `System` create/POST to handle `.privacy_declarations` defined inline, for backward compatibility and ease of use for API users. """ - await validate_privacy_declarations_data_uses(db, resource) + await validate_privacy_declarations(db, resource) # copy out the declarations to be stored separately # as they will be processed AFTER the system is added privacy_declarations = resource.privacy_declarations diff --git a/tests/ctl/core/test_api.py b/tests/ctl/core/test_api.py index 568fc9cbb1c..e155e4907ba 100644 --- a/tests/ctl/core/test_api.py +++ b/tests/ctl/core/test_api.py @@ -842,6 +842,7 @@ def test_system_update_privacy_declaration_invalid_data_use( test_config, system_update_request_body, generate_role_header, + db, ): auth_header = generate_role_header(roles=[OWNER]) system_update_request_body.privacy_declarations[0].data_use = "invalid_data_use" @@ -853,7 +854,143 @@ def test_system_update_privacy_declaration_invalid_data_use( ) assert result.status_code == HTTP_400_BAD_REQUEST # assert the system's privacy declaration has not been updated + db.refresh(system) + assert system.privacy_declarations[0].data_use == "advertising" + + def test_system_update_privacy_declaration_invalid_duplicate( + self, + system, + test_config, + system_update_request_body, + generate_role_header, + db, + ): + auth_header = generate_role_header(roles=[OWNER]) + + # test that 'exact' duplicate fails (data_use and name match) + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="declaration-name", # same as initial PrivacyDeclaration + data_categories=["user.payment"], # other fields can differ + data_use="provide", # same as initial PrivacyDeclaration + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + result = _api.update( + url=test_config.cli.server_url, + headers=auth_header, + resource_type="system", + json_resource=system_update_request_body.json(exclude_none=True), + ) + assert result.status_code == HTTP_400_BAD_REQUEST + # assert the system's privacy declaration has not been updated + db.refresh(system) + assert system.privacy_declarations[0].data_use == "advertising" + + # test that duplicate with no name on either declaration fails + system_update_request_body.privacy_declarations = [] + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="", # no name specified + data_categories=["user.payment"], # other fields can differ + data_use="provide", # identical data use + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="", # no name specified + data_categories=["user.payment"], # other fields can differ + data_use="provide", # identicial data use + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + result = _api.update( + url=test_config.cli.server_url, + headers=auth_header, + resource_type="system", + json_resource=system_update_request_body.json(exclude_none=True), + ) + assert result.status_code == HTTP_400_BAD_REQUEST + # assert the system's privacy declaration has not been updated + db.refresh(system) + assert system.privacy_declarations[0].data_use == "advertising" + + # test that duplicate data_use with no name on one declaration succeeds + system_update_request_body.privacy_declarations = [] + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="", # no name specified + data_categories=["user.payment"], # other fields can differ + data_use="provide", # identical data use + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="new declaration", # this name distinguishes the declaration from the above + data_categories=["user.payment"], # other fields can differ + data_use="provide", # identicial data use + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + result = _api.update( + url=test_config.cli.server_url, + headers=auth_header, + resource_type="system", + json_resource=system_update_request_body.json(exclude_none=True), + ) + assert result.status_code == HTTP_200_OK + # assert the system's privacy declarations have been updated + db.refresh(system) + # both declarations should have 'provide' data_use since the update was allowed + assert system.privacy_declarations[0].data_use == "provide" + assert system.privacy_declarations[1].data_use == "provide" + + # test that duplicate data_use with differeing names on declarations succeeds + system_update_request_body.privacy_declarations = [] + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="new declaration 1", # specify a unique name here + data_categories=["user.payment"], # other fields can differ + data_use="advertising", # identical data use + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="new declaration 2", # this name distinguishes the declaration from the above + data_categories=["user.payment"], # other fields can differ + data_use="advertising", # identicial data use + data_subjects=["anonymous_user"], # other fields can differ + data_qualifier="aggregated", # other fields can differ + dataset_references=[], + ) + ) + result = _api.update( + url=test_config.cli.server_url, + headers=auth_header, + resource_type="system", + json_resource=system_update_request_body.json(exclude_none=True), + ) + assert result.status_code == HTTP_200_OK + # assert the system's privacy declarations have been updated + db.refresh(system) + # both declarations should have 'advertising' data_use since the update was allowed assert system.privacy_declarations[0].data_use == "advertising" + assert system.privacy_declarations[1].data_use == "advertising" @pytest.mark.parametrize( "update_declarations", From 781c548d49cc519f88ad733d64c91d8fa2e005fd Mon Sep 17 00:00:00 2001 From: Adam Sachs Date: Fri, 28 Apr 2023 13:41:39 -0700 Subject: [PATCH 3/5] add test coverage specifically for privacy declaration record management --- tests/conftest.py | 25 +++++ tests/ctl/core/test_api.py | 200 ++++++++++++++++++++++++++++++++++++- 2 files changed, 223 insertions(+), 2 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 41164687995..9500369a700 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -968,6 +968,31 @@ def system(db: Session) -> System: return system +@pytest.fixture(scope="function") +def system_multiple_decs(db: Session, system: System) -> System: + """ + Add an additional PrivacyDeclaration onto the base System to test scenarios with + multiple PrivacyDeclarations on a given system + """ + PrivacyDeclaration.create( + db=db, + data={ + "name": "Collect data for third party sharing", + "system_id": system.id, + "data_categories": ["user.device.cookie_id"], + "data_use": "third_party_sharing", + "data_qualifier": "aggregated.anonymized.unlinked_pseudonymized.pseudonymized.identified", + "data_subjects": ["customer"], + "dataset_references": None, + "egress": None, + "ingress": None, + }, + ) + + db.refresh(system) + return system + + @pytest.fixture(scope="function") def system_third_party_sharing(db: Session) -> System: system_third_party_sharing = System.create( diff --git a/tests/ctl/core/test_api.py b/tests/ctl/core/test_api.py index e155e4907ba..5d96c45cf08 100644 --- a/tests/ctl/core/test_api.py +++ b/tests/ctl/core/test_api.py @@ -24,6 +24,7 @@ from fides.api.ctl.database.crud import get_resource from fides.api.ctl.routes import health from fides.api.ctl.routes.util import API_PREFIX, CLI_SCOPE_PREFIX_MAPPING +from fides.api.ctl.schemas.system import PrivacyDeclarationResponse from fides.api.ctl.sql_models import Dataset, PrivacyDeclaration, System from fides.api.ops.api.v1.scope_registry import ( CREATE, @@ -1029,12 +1030,33 @@ def test_system_update_privacy_declaration_invalid_duplicate( ] ), ( - # add 2 privacy declarations, one the same data use as existing + # add 2 privacy declarations, one the same data use and name as existing [ models.PrivacyDeclaration( name="declaration-name", data_categories=[], - data_use="provide", + data_use="third_party_sharing", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="Collect data for marketing", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), + ( + # add 2 privacy declarations, one the same data use and name as existing, other same data use + [ + models.PrivacyDeclaration( + name="Collect data for marketing", + data_categories=[], + data_use="advertising", data_subjects=[], data_qualifier="aggregated_data", dataset_references=[], @@ -1064,6 +1086,14 @@ def test_system_update_updates_declarations( system_update_request_body, update_declarations, ): + """ + Test to assert that our `PUT` endpoint acts in a fully declarative manner, putting the DB state of the + system's privacy requests in *exactly* the same state as specified on the request payload. + + This is executed against various different sets of input privacy declarations to ensure it works + in a variety of scenarios + """ + auth_header = generate_auth_header(scopes=[SYSTEM_UPDATE]) system_update_request_body.privacy_declarations = update_declarations result = _api.update( @@ -1098,6 +1128,172 @@ def test_system_update_updates_declarations( # and assert we don't have any extra response declarations assert len(db_decs) == 0 + @pytest.mark.parametrize( + "update_declarations", + [ + ( + [ # update a dec matching one existing dec + models.PrivacyDeclaration( + name="Collect data for marketing", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ) + ] + ), + ( + # update 2 privacy declarations both matching existing decs + [ + models.PrivacyDeclaration( + name="Collect data for marketing", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="Collect data for third party sharing", + data_categories=[], + data_use="third_party_sharing", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), + ( + # update 2 privacy declarations, one with matching name and data use, other only data use + [ + models.PrivacyDeclaration( + name="Collect data for marketing", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="declaration-name-2", + data_categories=[], + data_use="third_party_sharing", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), + ( + # update 2 privacy declarations, one with matching name and data use, other only data use but same data use + [ + models.PrivacyDeclaration( + name="Collect data for marketing", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="declaration-name-2", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), + ( + # add 2 new privacy declarations + [ + models.PrivacyDeclaration( + name="declaration-name", + data_categories=[], + data_use="provide", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="declaration-name-2", + data_categories=[], + data_use="provide", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), + ( + # specify no declarations, declarations should be cleared off the system + [] + ), + ], + ) + def test_system_update_manages_declaration_records( + self, + db, + test_config, + system_multiple_decs: System, + generate_auth_header, + system_update_request_body, + update_declarations, + ): + """ + Test to assert that existing privacy declaration records stay constant when necessary + """ + old_db_decs = [ + PrivacyDeclarationResponse.from_orm(dec) + for dec in system_multiple_decs.privacy_declarations + ] + old_decs_updated = [ + old_db_dec + for old_db_dec in old_db_decs + if any( + ( + old_db_dec.name == update_declaration.name + and old_db_dec.data_use == update_declaration.data_use + ) + for update_declaration in update_declarations + ) + ] + auth_header = generate_auth_header(scopes=[SYSTEM_UPDATE]) + system_update_request_body.privacy_declarations = update_declarations + _api.update( + url=test_config.cli.server_url, + headers=auth_header, + resource_type="system", + json_resource=system_update_request_body.json(exclude_none=True), + ) + + db.refresh(system_multiple_decs) + updated_decs: List[ + PrivacyDeclaration + ] = system_multiple_decs.privacy_declarations.copy() + + for old_dec_updated in old_decs_updated: + updated_dec = next( + updated_dec + for updated_dec in updated_decs + if updated_dec.name == old_dec_updated.name + and updated_dec.data_use == old_dec_updated.data_use + ) + assert updated_dec.id == old_dec_updated.id + updated_decs.remove(updated_dec) + old_db_decs.remove(old_dec_updated) + + # our old db decs that were _not_ updated should no longer be in the db + for old_db_dec in old_db_decs: + assert not any( + old_db_dec.id == updated_dec.id for updated_dec in updated_decs + ) + + # and just verify that we have same number of privacy declarations in db as specified in the update request + assert len(PrivacyDeclaration.all(db)) == len(update_declarations) + @pytest.mark.unit class TestSystemDelete: From ba3dd74517512adf57862dd6febbf32058574bbf Mon Sep 17 00:00:00 2001 From: Adam Sachs Date: Fri, 28 Apr 2023 13:44:21 -0700 Subject: [PATCH 4/5] update changelog --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 897f4b2e8be..9cf46d26075 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,6 +51,7 @@ The types of changes are: - Fix a typo in the Admin UI [#3166](https://github.com/ethyca/fides/pull/3166) - The `--local` flag is now respected for the `scan dataset db` command [#3096](https://github.com/ethyca/fides/pull/3096) - Fixing issue where connectors with external dataset references would fail to save [#3142](https://github.com/ethyca/fides/pull/3142) +- Ensure privacy declaration IDs are stable across updates through system API [#3188](https://github.com/ethyca/fides/pull/3188) ### Developer Experience From bd63960c8b70c32360ca63e20c5af920668f8789 Mon Sep 17 00:00:00 2001 From: Adam Sachs Date: Sun, 30 Apr 2023 18:24:51 -0700 Subject: [PATCH 5/5] improve test coverage --- tests/ctl/core/test_api.py | 149 ++++++++++++++++++++++++++++++++----- 1 file changed, 131 insertions(+), 18 deletions(-) diff --git a/tests/ctl/core/test_api.py b/tests/ctl/core/test_api.py index 5d96c45cf08..04198242929 100644 --- a/tests/ctl/core/test_api.py +++ b/tests/ctl/core/test_api.py @@ -895,20 +895,20 @@ def test_system_update_privacy_declaration_invalid_duplicate( system_update_request_body.privacy_declarations.append( models.PrivacyDeclaration( name="", # no name specified - data_categories=["user.payment"], # other fields can differ + data_categories=["user.payment"], data_use="provide", # identical data use data_subjects=["anonymous_user"], # other fields can differ - data_qualifier="aggregated", # other fields can differ + data_qualifier="aggregated", dataset_references=[], ) ) system_update_request_body.privacy_declarations.append( models.PrivacyDeclaration( name="", # no name specified - data_categories=["user.payment"], # other fields can differ + data_categories=["user.payment"], data_use="provide", # identicial data use - data_subjects=["anonymous_user"], # other fields can differ - data_qualifier="aggregated", # other fields can differ + data_subjects=["anonymous_user"], + data_qualifier="aggregated", dataset_references=[], ) ) @@ -928,20 +928,20 @@ def test_system_update_privacy_declaration_invalid_duplicate( system_update_request_body.privacy_declarations.append( models.PrivacyDeclaration( name="", # no name specified - data_categories=["user.payment"], # other fields can differ + data_categories=["user.payment"], data_use="provide", # identical data use - data_subjects=["anonymous_user"], # other fields can differ - data_qualifier="aggregated", # other fields can differ + data_subjects=["anonymous_user"], + data_qualifier="aggregated", dataset_references=[], ) ) system_update_request_body.privacy_declarations.append( models.PrivacyDeclaration( name="new declaration", # this name distinguishes the declaration from the above - data_categories=["user.payment"], # other fields can differ + data_categories=["user.payment"], data_use="provide", # identicial data use - data_subjects=["anonymous_user"], # other fields can differ - data_qualifier="aggregated", # other fields can differ + data_subjects=["anonymous_user"], + data_qualifier="aggregated", dataset_references=[], ) ) @@ -963,20 +963,20 @@ def test_system_update_privacy_declaration_invalid_duplicate( system_update_request_body.privacy_declarations.append( models.PrivacyDeclaration( name="new declaration 1", # specify a unique name here - data_categories=["user.payment"], # other fields can differ + data_categories=["user.payment"], data_use="advertising", # identical data use - data_subjects=["anonymous_user"], # other fields can differ - data_qualifier="aggregated", # other fields can differ + data_subjects=["anonymous_user"], + data_qualifier="aggregated", dataset_references=[], ) ) system_update_request_body.privacy_declarations.append( models.PrivacyDeclaration( name="new declaration 2", # this name distinguishes the declaration from the above - data_categories=["user.payment"], # other fields can differ + data_categories=["user.payment"], data_use="advertising", # identicial data use - data_subjects=["anonymous_user"], # other fields can differ - data_qualifier="aggregated", # other fields can differ + data_subjects=["anonymous_user"], + data_qualifier="aggregated", dataset_references=[], ) ) @@ -993,6 +993,50 @@ def test_system_update_privacy_declaration_invalid_duplicate( assert system.privacy_declarations[0].data_use == "advertising" assert system.privacy_declarations[1].data_use == "advertising" + # test that differeing data_use with same names on declarations succeeds + system_update_request_body.privacy_declarations = [] + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="new declaration 1", # identical name + data_categories=["user.payment"], + data_use="advertising", # differing data use + data_subjects=["anonymous_user"], + data_qualifier="aggregated", + dataset_references=[], + ) + ) + system_update_request_body.privacy_declarations.append( + models.PrivacyDeclaration( + name="new declaration 1", # identical name + data_categories=["user.payment"], + data_use="provide", # differing data use + data_subjects=["anonymous_user"], + data_qualifier="aggregated", + dataset_references=[], + ) + ) + result = _api.update( + url=test_config.cli.server_url, + headers=auth_header, + resource_type="system", + json_resource=system_update_request_body.json(exclude_none=True), + ) + assert result.status_code == HTTP_200_OK + # assert the system's privacy declarations have been updated + db.refresh(system) + # should be one declaration with advertising, one with provide + assert ( + system.privacy_declarations[0].data_use == "advertising" + and system.privacy_declarations[1].data_use == "provide" + ) or ( + system.privacy_declarations[1].data_use == "advertising" + and system.privacy_declarations[0].data_use == "provide" + ) + assert ( + system.privacy_declarations[0].name == "new declaration 1" + and system.privacy_declarations[1].name == "new declaration 1" + ) + @pytest.mark.parametrize( "update_declarations", [ @@ -1143,6 +1187,30 @@ def test_system_update_updates_declarations( ) ] ), + ( + [ # add a new single dec with same data use + models.PrivacyDeclaration( + name="declaration-name-1", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ) + ] + ), + ( + [ # add a new single dec with same data use, no name + models.PrivacyDeclaration( + name="", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ) + ] + ), ( # update 2 privacy declarations both matching existing decs [ @@ -1206,6 +1274,27 @@ def test_system_update_updates_declarations( ), ] ), + ( + # update 2 privacy declarations, one with only matching data use, other totally new + [ + models.PrivacyDeclaration( + name="declaration-name-1", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="declaration-name-2", + data_categories=[], + data_use="provide", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), ( # add 2 new privacy declarations [ @@ -1220,7 +1309,28 @@ def test_system_update_updates_declarations( models.PrivacyDeclaration( name="declaration-name-2", data_categories=[], - data_use="provide", + data_use="improve", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + ] + ), + ( + # add 2 new privacy declarations, same data uses as existing decs but no names + [ + models.PrivacyDeclaration( + name="", + data_categories=[], + data_use="advertising", + data_subjects=[], + data_qualifier="aggregated_data", + dataset_references=[], + ), + models.PrivacyDeclaration( + name="", + data_categories=[], + data_use="third_party_sharing", data_subjects=[], data_qualifier="aggregated_data", dataset_references=[], @@ -1281,7 +1391,10 @@ def test_system_update_manages_declaration_records( if updated_dec.name == old_dec_updated.name and updated_dec.data_use == old_dec_updated.data_use ) + # assert that the updated dec in the DB kept the same ID assert updated_dec.id == old_dec_updated.id + + # remove from our lists to check since we've confirmed ID stayed constant updated_decs.remove(updated_dec) old_db_decs.remove(old_dec_updated)