From f85e523398acfbe641f047c60703f9c3b2da2166 Mon Sep 17 00:00:00 2001 From: pbrassel <52356233+pbrassel@users.noreply.github.com> Date: Tue, 6 Oct 2026 12:18:08 +0200 Subject: [PATCH 1/3] feat: expand gpas client with more functionalities from the regular and the domain service The new functionalities include: - retrieve domains for configured pseudonym prefixes or suffixes - validation of pseudonyms - update the expiration date of pseudonyms - anonymization of pseudonyms and entries - check if a value is anonym or if a pseudonym is already anonymized - retrieve pseudonyms that match a specific prefix - retrieval of pseudonym trees and nets - check if pseudonyms are deletable - listing pseudonyms of a specified domain --- README.md | 4 +- mosaic_client/gpas/__init__.py | 44 ++- mosaic_client/gpas/client.py | 604 +++++++++++++++++++++++++++++---- mosaic_client/gpas/models.py | 125 ++++++- mosaic_client/gpas/schemas.py | 98 +++++- tests/conftest.py | 12 +- tests/helpers.py | 5 + tests/test_gpas_client.py | 252 +++++++++++--- 8 files changed, 1009 insertions(+), 135 deletions(-) diff --git a/README.md b/README.md index d348c2b..b32c750 100644 --- a/README.md +++ b/README.md @@ -211,9 +211,9 @@ gpas = GPASClient( domain_client="http://localhost:8080/gpas/DomainService?wsdl", ) -pseudonym = gpas.get_or_create_pseudonym_for(domain_name="default", value="value123") +pseudonym = gpas.get_or_create_pseudonyms_for(domain_name="default", value="value123") -print(f"Pseudonym: {pseudonym}") +print(f"Pseudonym: {pseudonym.pseudonym}") ``` ```text diff --git a/mosaic_client/gpas/__init__.py b/mosaic_client/gpas/__init__.py index dac0197..5de1479 100644 --- a/mosaic_client/gpas/__init__.py +++ b/mosaic_client/gpas/__init__.py @@ -1,8 +1,36 @@ from mosaic_client.gpas.client import GPASClient -from mosaic_client.gpas.models import Domain, DomainConfig, DomainResponse -from mosaic_client.gpas.schemas import DomainConfigSchema, DomainResponseSchema, DomainSchema +from mosaic_client.gpas.models import ( + AnonymizationResponse, + DeletionResponse, + Domain, + DomainConfig, + DomainResponse, + InsertPairException, + Pseudonym, + PseudonymNet, + PseudonymNetNode, + PseudonymTree, + ValueToPseudonyms, +) +from mosaic_client.gpas.schemas import ( + AnonymizationResponseSchema, + DeletionResponseSchema, + DomainConfigSchema, + DomainResponseSchema, + DomainSchema, + InsertPairExceptionSchema, + PseudonymNetNodeSchema, + PseudonymNetSchema, + PseudonymSchema, + PseudonymTreeSchema, + ValueToPseudonymsSchema, +) __all__ = [ + "AnonymizationResponse", + "AnonymizationResponseSchema", + "DeletionResponse", + "DeletionResponseSchema", "Domain", "DomainConfig", "DomainConfigSchema", @@ -10,4 +38,16 @@ "DomainResponseSchema", "DomainSchema", "GPASClient", + "InsertPairException", + "InsertPairExceptionSchema", + "Pseudonym", + "PseudonymNet", + "PseudonymNetNode", + "PseudonymNetNodeSchema", + "PseudonymNetSchema", + "PseudonymSchema", + "PseudonymTree", + "PseudonymTreeSchema", + "ValueToPseudonyms", + "ValueToPseudonymsSchema", ] diff --git a/mosaic_client/gpas/client.py b/mosaic_client/gpas/client.py index 93d96b3..a2e9312 100644 --- a/mosaic_client/gpas/client.py +++ b/mosaic_client/gpas/client.py @@ -2,10 +2,32 @@ This module contains functions for interacting with the gPAS SOAP interface. """ +from datetime import date + from zeep import Client -from mosaic_client.gpas.models import Domain, DomainResponse -from mosaic_client.gpas.schemas import DomainResponseSchema, DomainSchema +from mosaic_client.gpas.models import ( + AnonymizationResponse, + DeletionResponse, + Domain, + DomainResponse, + InsertPairException, + Pseudonym, + PseudonymNet, + PseudonymTree, + ValueToPseudonyms, +) +from mosaic_client.gpas.schemas import ( + AnonymizationResponseSchema, + DeletionResponseSchema, + DomainResponseSchema, + DomainSchema, + InsertPairExceptionSchema, + PseudonymNetSchema, + PseudonymSchema, + PseudonymTreeSchema, + ValueToPseudonymsSchema, +) from mosaic_client.helpers import WSDLClient, _cast_client, _read_key_value_list, _serialize_dict KeyValueTuple = tuple[str, str] @@ -13,53 +35,94 @@ def delete_entry(client: Client, domain_name: str, value: str) -> None: """ - Deletes a value and its associated pseudonym from the specified data domain. + Deletes a value and all of its associated pseudonyms from the specified data domain. :param client: Zeep client with gPAS service definitions :param domain_name: name of the domain where the value is present :param value: value to remove """ - client.service.deleteEntry(domainName=domain_name, value=value) + client.service.deleteAllEntriesForValue(domainName=domain_name, value=value) -def delete_entries(client: Client, domain_name: str, values: list[str]) -> list[KeyValueTuple]: +def delete_entries(client: Client, domain_name: str, values: list[str]) -> list[DeletionResponse]: """ - Deletes a list of values and their associated pseudonyms from the specified data domain. + Deletes a list of values and all associated pseudonyms from the specified data domain. :param client: Zeep client with gPAS service definitions - :param domain_name: name of the domain where the value is present + :param domain_name: name of the domain where the values are present :param values: list of values to remove - :return: list of key-value pairs where the key is the value that was requested to be deleted, and the value is - an indicator whether deletion was successful or not + :return: list of deletion response data class instances """ - response = client.service.deleteEntries(domainName=domain_name, values=values) - return _read_key_value_list(_serialize_dict(response)) + response = client.service.deleteAllEntriesForValues(domainName=domain_name, values=values) + return [DeletionResponseSchema().load(_serialize_dict(r)) for r in response] -def get_or_create_pseudonym_for(client: Client, domain_name: str, value: str) -> str: +def delete_pseudonym(client: Client, domain_name: str, pseudonym: str) -> None: """ - Requests a new pseudonym for a value in the specified data domain. + Deletes a pseudonym and the associated value from the specified data domain. :param client: Zeep client with gPAS service definitions - :param domain_name: name of the domain where the value is supposed to be inserted - :param value: value to insert - :return: pseudonym that is assigned to the value + :param domain_name: name of the domain where the pseudonym is present + :param pseudonym: pseudonym to delete """ - return client.service.getOrCreatePseudonymFor(domainName=domain_name, value=value) + client.service.deletePseudonym(psn=pseudonym, domainName=domain_name) -def get_or_create_pseudonym_for_list(client: Client, domain_name: str, values: list[str]) -> list[KeyValueTuple]: +def delete_pseudonyms(client: Client, domain_name: str, pseudonyms: list[str]) -> list[DeletionResponse]: """ - Request new pseudonyms for a list of values in the specified data domain. + Deletes a list of pseudonyms and their associated values from the specified data domain. :param client: Zeep client with gPAS service definitions - :param domain_name: name of the domain where the value is supposed to be inserted - :param values: list of values to insert - :return: list of key-value pairs where the key is the original value that is supposed to be pseudonymized, - and the value is the assigned pseudonym + :param domain_name: name of the domain where the pseudonyms are present + :param pseudonyms: list of pseudonyms to remove + :return: list of delete response data class instances """ - response = client.service.getOrCreatePseudonymForList(domainName=domain_name, values=values) - return _read_key_value_list(_serialize_dict(response)) + response = client.service.deletePseudonyms(domainName=domain_name, psns=pseudonyms) + return [DeletionResponseSchema().load(_serialize_dict(r)) for r in response] + + +def get_or_create_pseudonyms_for( + client: Client, + domain_name: str, + value: str, + min_number: int = 1, +) -> ValueToPseudonyms: + """ + Gets the pseudonyms or creates new pseudonyms for a given value in the specified domain. This function assures that + at least the specified number of pseudonyms exist for the given value in the domain. Raises a Fault if the domain is + full, expired or not found. Also raises a Fault if more than one pseudonym is requested for a domain that does not + allow multiple pseudonyms per value. + + :param client: Zeep client with gPAS service definitions + :param domain_name: name of the domain where the value should be got from or created in + :param value: value to get or create pseudonyms for + :param min_number: minimum number of pseudonyms that should exist for the given value, defaults to one + :return: value to pseudonyms data class instance + """ + psns = client.service.getOrCreatePseudonymsFor(domainName=domain_name, value=value, minNumber=min_number) + return ValueToPseudonyms(value=value, pseudonyms=psns) + + +def get_or_create_pseudonyms_for_list( + client: Client, + domain_name: str, + values: list[str], + min_number: int = 1, +) -> list[ValueToPseudonyms]: + """ + Gets the pseudonyms or creates new pseudonyms for a given list of values in the specified domain. This function + assures that at least the specified number of pseudonyms exist for each of the given values in the domain. Raises + a Fault if the domain is full, expired or not found. Also raises a Fault if more than one pseudonym per value is + requested for a domain that does not allow multiple pseudonyms per value. + + :param client: Zeep client with gPAS service definitions + :param domain_name: name of the domain where the values should be got from or created in + :param values: list of values to get or create pseudonyms for + :param min_number: minimum number of pseudonyms that should exist for each value, defaults to one + :return: list of value to pseudonyms data class instances + """ + response = client.service.getOrCreatePseudonymsForList(domainName=domain_name, values=values, minNumber=min_number) + return [ValueToPseudonymsSchema().load(_serialize_dict(r)) for r in response] def get_value_for(client: Client, domain_name: str, pseudonym: str) -> str: @@ -87,29 +150,32 @@ def get_value_for_list(client: Client, domain_name: str, pseudonyms: list[str]) return _read_key_value_list(_serialize_dict(response)) -def get_pseudonym_for(client: Client, domain_name: str, value: str) -> str: +def get_pseudonyms_for(client: Client, domain_name: str, value: str) -> ValueToPseudonyms: """ - Gets the pseudonym for a value in the specified data domain. + Get all pseudonyms for a value in the specified data domain. Note that a domain can be configured to allow multiple + pseudonyms per value. :param client: Zeep client with gPAS service definitions :param domain_name: name of the domain where the value is present :param value: value to resolve - :return: pseudonym assigned to the value + :return: value to pseudonyms data class instance """ - return client.service.getPseudonymFor(domainName=domain_name, value=value) + psns = client.service.getPseudonymsFor(domainName=domain_name, value=value) + return ValueToPseudonyms(value=value, pseudonyms=psns) -def get_pseudonym_for_list(client: Client, domain_name: str, values: list[str]) -> list[KeyValueTuple]: +def get_pseudonyms_for_list(client: Client, domain_name: str, values: list[str]) -> list[ValueToPseudonyms]: """ - Gets the pseudonyms for a list of values in the specified data domain. + Get all pseudonyms for each value in a list of values in the specified data domain. Note that a domain can be + specified to allow multiple pseudonyms per value. :param client: Zeep client with gPAS service definitions :param domain_name: name of the domain where the values are present :param values: values to resolve - :return: list of key-value pairs, structured as { value => pseudonym } + :return: list of value to pseudonyms data class instances """ - response = client.service.getPseudonymForList(domainName=domain_name, values=values) - return _read_key_value_list(_serialize_dict(response)) + response_soap = client.service.getPseudonymsForList(domainName=domain_name, values=values) + return [ValueToPseudonymsSchema().load(_serialize_dict(r)) for r in response_soap] def insert_value_pseudonym_pair(client: Client, domain_name: str, value: str, pseudonym: str) -> None: @@ -124,17 +190,22 @@ def insert_value_pseudonym_pair(client: Client, domain_name: str, value: str, ps client.service.insertValuePseudonymPair(domainName=domain_name, value=value, pseudonym=pseudonym) -def insert_value_pseudonym_pairs(client: Client, domain_name: str, pairs: list[KeyValueTuple]) -> None: +def insert_value_pseudonym_pairs( + client: Client, + domain_name: str, + pairs: list[KeyValueTuple], +) -> list[InsertPairException]: """ - Manually inserts a list of values and pseudonyms into the specified data domain. + Manually inserts a list of values and pseudonyms tuples into the specified data domain. :param client: Zeep client with gPAS service definitions :param domain_name: name of the domain to insert the pairs into :param pairs: list of key-value pairs, structured as { value => pseudonym } + :return: list of insert pair exception data class instances for cases that could not be inserted and raised an error """ # for some reason this is the only case where the "entry" key is mandatory. it is not returned by any other # endpoints where there's supposedly an "entry" key, e.g. deleteEntries (???) - client.service.insertValuePseudonymPairs( + response = client.service.insertValuePseudonymPairs( domainName=domain_name, pairs={ "entry": [ @@ -147,6 +218,11 @@ def insert_value_pseudonym_pairs(client: Client, domain_name: str, pairs: list[K }, ) + if response is None: + response = [] + + return [InsertPairExceptionSchema().load(_serialize_dict(r)) for r in response] + def add_domain(client: Client, domain: Domain) -> None: """ @@ -171,6 +247,38 @@ def get_domain(client: Client, domain_name: str) -> DomainResponse: return DomainResponseSchema().load(_serialize_dict(domain_soap)) +def get_domains_for_prefix(client: Client, prefix: str) -> list[DomainResponse]: + """ + Lists all domains that configured the specified pseudonym prefix. + + :param client: Zeep client with gPAS domain service definitions + :param prefix: configured pseudonym prefix + :return: list of domain response instances + """ + domains = client.service.getDomainsForPrefix(prefix=prefix) + + if domains is None: + domains = [] + + return [DomainResponseSchema().load(_serialize_dict(domain)) for domain in domains] + + +def get_domains_for_suffix(client: Client, suffix: str) -> list[DomainResponse]: + """ + Lists all domains that configured the specified pseudonym suffix. + + :param client: Zeep client with gPAS domain service definitions + :param suffix: configured pseudonym suffix + :return: list of domain response instances + """ + domains = client.service.getDomainsForSuffix(suffix=suffix) + + if domains is None: + domains = [] + + return [DomainResponseSchema().load(_serialize_dict(domain)) for domain in domains] + + def list_domains(client: Client) -> list[DomainResponse]: """ Lists all available domains in the gPAS instance. @@ -179,6 +287,10 @@ def list_domains(client: Client) -> list[DomainResponse]: :return: list of all available domains as domain response instances """ domains = _serialize_dict(client.service.listDomains()) + + if domains is None: + domains = [] + return [DomainResponseSchema().load(domain) for domain in domains] @@ -192,6 +304,179 @@ def delete_domain(client: Client, domain_name: str) -> None: client.service.deleteDomainWithPSNs(domainName=domain_name) +def validate_pseudonym(client: Client, pseudonym: str, domain_name: str) -> None: + """ + Validates a given pseudonym against the specified domain. Raise a Fault if the pseudonym is not valid. + + :param client: Zeep client with gPAS service definitions + :param pseudonym: pseudonym to validate + :param domain_name: domain for which the pseudonym should be validated + """ + client.service.validatePSN(psn=pseudonym, domainName=domain_name) + + +def update_pseudonym_expiration_date(client: Client, pseudonym: str, domain_name: str, expiration_date: date) -> None: + """ + Updates the expiration date of a pseudonym. Raises a Fault if there is no domain with the specified name, the + expiration date is invalid, there is no such pseudonym or the domain has not been configured for expirations. + + :param client: Zeep client with gPAS service definitions + :param pseudonym: pseudonym to update + :param domain_name: domain which the pseudonym belongs to + :param expiration_date: new expiration date of the pseudonym + """ + client.service.updatePseudonymExpirationDate( + psn=pseudonym, + domainName=domain_name, + newExpirationDate=expiration_date, + ) + + +def is_anonym(client: Client, value: str) -> bool: + """ + Checks if the given value is an Anonym. + + :param client: Zeep client with gPAS service definitions + :param value: value to check + :return: True if the given value is an Anonym, False otherwise + """ + return client.service.isAnonym(value) + + +def is_anonymized(client: Client, pseudonym: str, domain_name: str) -> bool: + """ + Checks if the given pseudonym of the specified domain is anonymized. Raises a Fault if the given pseudonym is not + found in the given domain or there is no domain with such a name. + + :param client: Zeep client with gPAS service definitions + :param pseudonym: pseudonym to check if it is anonymized + :param domain_name: domain for which the pseudonym should be checked + :return: True if the given pseudonym is anonymized, False otherwise + """ + return client.service.isAnonymised(psn=pseudonym, domainName=domain_name) + + +def anonymize_pseudonym(client: Client, pseudonym: str, domain_name: str) -> None: + """ + Anonymizes the value of the given pseudonym in the specified domain. Raises a Fault if the pseudonym is not found in + the given domain or there is no domain with such a name. + + :param client: Zeep client with gPAS service definitions + :param pseudonym: pseudonym to anonymize + :param domain_name: domain for which the pseudonym should be anonymized + """ + client.service.anonymisePseudonym(psn=pseudonym, domainName=domain_name) + + +def anonymize_pseudonyms( + client: Client, + pseudonyms: list[str], + domain_name: str, +) -> list[AnonymizationResponse]: + """ + Anonymizes the values of the given pseudonyms in the specified domain. Raises a Fault if there is no domain with + such a name. + + :param client: Zeep client with gPAS service definitions + :param pseudonyms: list of pseudonyms to anonymize + :param domain_name: domain for which the pseudonyms should be anonymized + :return: list of anonymization response data class instances + """ + response_soap = client.service.anonymisePseudonyms(psns=pseudonyms, domainName=domain_name) + return [AnonymizationResponseSchema().load(_serialize_dict(r)) for r in response_soap] + + +def anonymize_entry(client: Client, value: str, domain_name: str) -> None: + """ + Anonymizes a given value in the specified domain. If the specified domain allows multiple pseudonyms per value, all + values will be anonymized. Raises a Fault if the value is not found in the specified domain, there is no domain with + such a name, or the value is already anonymized. + + :param client: Zeep client with gPAS service definitions + :param value: value to anonymize + :param domain_name: domain for which the value should be anonymized + """ + client.service.anonymiseAllEntriesForValue(value=value, domainName=domain_name) + + +def anonymize_entries(client: Client, values: list[str], domain_name: str) -> list[AnonymizationResponse]: + """ + Anonymizes given values in the specified domain. If the specified domain allows multiple pseudonyms per value, all + values will be anonymized. Raises a Fault if there is no domain with such a name. + + :param client: Zeep client with gPAS service definitions + :param values: list of values to anonymize + :param domain_name: domain for which the values should be anonymized + :return: list of anonymization response data class instances + """ + response_soap = client.service.anonymiseAllEntriesForValues(values=values, domainName=domain_name) + return [AnonymizationResponseSchema().load(_serialize_dict(r)) for r in response_soap] + + +def get_pseudonyms_for_value_prefix(client: Client, value_prefix: str, domain_name: str) -> list[ValueToPseudonyms]: + """ + Returns all pseudonyms for each value that starts with the specified prefix in the specified domain. Raises a Fault + if there is no domain with such a name. + + :param client: Zeep client with gPAS service definitions + :param value_prefix: the prefix of values for which the pseudonyms should be retrieved + :param domain_name: domain for which the pseudonyms should be retrieved + :return: list of value to pseudonyms data class instances + """ + response_soap = client.service.getPseudonymsForValuePrefix(valuePrefix=value_prefix, domainName=domain_name) + return [ValueToPseudonymsSchema().load(_serialize_dict(r)) for r in response_soap] + + +def get_pseudonym_tree(client: Client, pseudonym: str, domain_name: str) -> PseudonymTree: + """ + Creates a pseudonym tree with all values that are somehow linked to the given pseudonym. Raises a Fault if the + pseudonym is not found in the specified domain, there is no domain with such a name, or if the value is already + anonymized. + + :param client: Zeep client with gPAS service definitions + :param pseudonym: pseudonym to create the tree from + :param domain_name: name of the domain for the given pseudonym + :return: pseudonym tree data class instance + """ + tree_soap = client.service.getPSNTreeForPSN(psn=pseudonym, domainName=domain_name) + return PseudonymTreeSchema().load(_serialize_dict(tree_soap)) + + +def get_pseudonym_net(client: Client, value_or_pseudonym: str) -> PseudonymNet: + """ + Creates a pseudonym net with all values and pseudonyms that are somehow linked to the given value or pseudonym. + + :param client: Zeep client with gPAS service definitions + :param value_or_pseudonym: value or pseudonym to create the net from + :return: pseudonym net data class instance + """ + net_soap = client.service.getPSNNetFor(valueOrPSN=value_or_pseudonym) + return PseudonymNetSchema().load(_serialize_dict(net_soap)) + + +def are_pseudonyms_deletable(client: Client, domain_name: str) -> bool: + """ + Checks if the deletion of pseudonyms is allowed for the specified domain. + + :param client: Zeep client with gPAS domain service definitions + :param domain_name: name of the domain to check + :return: True if pseudonyms are allowed to be deleted, False otherwise + """ + return client.service.arePSNDeletable(domainName=domain_name) + + +def list_pseudonyms(client: Client, domain_name: str) -> list[Pseudonym]: + """ + Retrieves all pseudonyms for the specified domain. Raises a Fault if there is no domain with such a name. + + :param client: Zeep client with gPAS domain service definitions + :param domain_name: name of the domain to retrieve pseudonyms for + :return: list of pseudonym data class instances + """ + psns = client.service.listPSNs(domainName=domain_name) + return [PseudonymSchema().load(_serialize_dict(psn)) for psn in psns] + + class GPASClient(WSDLClient): """ This class is a wrapper around the gPAS service functions. @@ -210,44 +495,74 @@ def __init__(self, client: Client | str, domain_client: Client | str): def delete_entry(self, domain_name: str, value: str) -> None: """ - Deletes a value and its associated pseudonym from the specified data domain. + Deletes a value and all of its associated pseudonyms from the specified data domain. :param domain_name: name of the domain where the value is present :param value: value to remove """ delete_entry(self._client, domain_name, value) - def delete_entries(self, domain_name: str, values: list[str]) -> list[KeyValueTuple]: + def delete_entries(self, domain_name: str, values: list[str]) -> list[DeletionResponse]: """ - Deletes a list of values and their associated pseudonyms from the specified data domain. + Deletes a list of values and all associated pseudonyms from the specified data domain. - :param domain_name: name of the domain where the value is present + :param domain_name: name of the domain where the values are present :param values: list of values to remove - :return: list of key-value pairs where the key is the value that was requested to be deleted, and the value is - an indicator whether deletion was successful or not + :return: list of deletion response data class instances """ return delete_entries(self._client, domain_name, values) - def get_or_create_pseudonym_for(self, domain_name: str, value: str) -> str: + def delete_pseudonym(self, domain_name: str, pseudonym: str) -> None: + """ + Deletes a pseudonym and the associated value from the specified data domain. + + :param domain_name: name of the domain where the pseudonym is present + :param pseudonym: pseudonym to delete """ - Requests a new pseudonym for a value in the specified data domain. + delete_pseudonym(self._client, domain_name, pseudonym) - :param domain_name: name of the domain where the value is supposed to be inserted - :param value: value to insert - :return: pseudonym that is assigned to the value + def delete_pseudonyms(self, domain_name: str, pseudonyms: list[str]) -> list[DeletionResponse]: """ - return get_or_create_pseudonym_for(self._client, domain_name, value) + Deletes a list of pseudonyms and their associated values from the specified data domain. - def get_or_create_pseudonym_for_list(self, domain_name: str, values: list[str]) -> list[KeyValueTuple]: + :param domain_name: name of the domain where the pseudonyms are present + :param pseudonyms: list of pseudonyms to remove + :return: list of delete response data class instances """ - Request new pseudonyms for a list of values in the specified data domain. + return delete_pseudonyms(self._client, domain_name, pseudonyms) - :param domain_name: name of the domain where the value is supposed to be inserted - :param values: list of values to insert - :return: list of key-value pairs where the key is the original value that is supposed to be pseudonymized, - and the value is the assigned pseudonym + def get_or_create_pseudonyms_for(self, domain_name: str, value: str, min_number: int = 1) -> ValueToPseudonyms: + """ + Gets the pseudonyms or creates new pseudonyms for a given value in the specified domain. This function assures + that at the least specified number of pseudonyms exist for the given value in the domain. Raises a Fault if the + domain is full, expired or not found. Also raises a Fault if more than one pseudonym is requested for a domain + that does not allow multiple pseudonyms per value. + + :param domain_name: name of the domain where the value should be got from or created in + :param value: value to get or create pseudonyms for + :param min_number: minimum number of pseudonyms that should exist for the given value, defaults to one + :return: value to pseudonyms data class instance + """ + return get_or_create_pseudonyms_for(self._client, domain_name, value, min_number) + + def get_or_create_pseudonyms_for_list( + self, + domain_name: str, + values: list[str], + min_number: int = 1, + ) -> list[ValueToPseudonyms]: """ - return get_or_create_pseudonym_for_list(self._client, domain_name, values) + Gets the pseudonyms or creates new pseudonyms for a given list of values in the specified domain. This function + assures that at least the specified number of pseudonyms exist for each of the given values in the domain. + Raises a Fault if the domain is full, expired or not found. Also raises a Fault if more than one pseudonym per + value is requested for a domain that does not allow multiple pseudonyms per value. + + :param domain_name: name of the domain where the values should be got from or created in + :param values: list of values to get or create pseudonyms for + :param min_number: minimum number of pseudonyms that should exist for each value, defaults to one + :return: list of value to pseudonyms data class instances + """ + return get_or_create_pseudonyms_for_list(self._client, domain_name, values, min_number) def get_value_for(self, domain_name: str, pseudonym: str) -> str: """ @@ -269,25 +584,27 @@ def get_value_for_list(self, domain_name: str, pseudonyms: list[str]) -> list[Ke """ return get_value_for_list(self._client, domain_name, pseudonyms) - def get_pseudonym_for(self, domain_name: str, value: str) -> str: + def get_pseudonyms_for(self, domain_name: str, value: str) -> ValueToPseudonyms: """ - Gets the pseudonym for a value in the specified data domain. + Get all pseudonyms for a value in the specified data domain. Note that a domain can be configured to allow + multiple pseudonyms per value. :param domain_name: name of the domain where the value is present :param value: value to resolve - :return: pseudonym assigned to the value + :return: value to pseudonyms data class instance """ - return get_pseudonym_for(self._client, domain_name, value) + return get_pseudonyms_for(self._client, domain_name, value) - def get_pseudonym_for_list(self, domain_name: str, values: list[str]) -> list[KeyValueTuple]: + def get_pseudonyms_for_list(self, domain_name: str, values: list[str]) -> list[ValueToPseudonyms]: """ - Gets the pseudonyms for a list of values in the specified data domain. + Get all pseudonyms for each value in a list of values in the specified data domain. Note that a domain can be + specified to allow multiple pseudonyms per value. :param domain_name: name of the domain where the values are present :param values: values to resolve - :return: list of key-value pairs, structured as { value => pseudonym } + :return: list of value to pseudonyms data class instances """ - return get_pseudonym_for_list(self._client, domain_name, values) + return get_pseudonyms_for_list(self._client, domain_name, values) def insert_value_pseudonym_pair(self, domain_name: str, value: str, pseudonym: str) -> None: """ @@ -299,14 +616,16 @@ def insert_value_pseudonym_pair(self, domain_name: str, value: str, pseudonym: s """ insert_value_pseudonym_pair(self._client, domain_name, value, pseudonym) - def insert_value_pseudonym_pairs(self, domain_name: str, pairs: list[KeyValueTuple]) -> None: + def insert_value_pseudonym_pairs(self, domain_name: str, pairs: list[KeyValueTuple]) -> list[InsertPairException]: """ - Manually inserts a list of values and pseudonyms into the specified data domain. + Manually inserts a list of values and pseudonyms tuples into the specified data domain. :param domain_name: name of the domain to insert the pairs into :param pairs: list of key-value pairs, structured as { value => pseudonym } + :return: list of insert pair exception data class instances for cases that could not be inserted and raised an + error """ - insert_value_pseudonym_pairs(self._client, domain_name, pairs) + return insert_value_pseudonym_pairs(self._client, domain_name, pairs) def add_domain(self, domain: Domain) -> None: """ @@ -325,6 +644,24 @@ def get_domain(self, domain_name: str) -> DomainResponse: """ return get_domain(self._domain_client, domain_name) + def get_domains_for_prefix(self, prefix: str) -> list[DomainResponse]: + """ + Lists all domains that configured the specified pseudonym prefix. + + :param prefix: configured pseudonym prefix + :return: list of domain response instances + """ + return get_domains_for_prefix(self._domain_client, prefix) + + def get_domains_for_suffix(self, suffix: str) -> list[DomainResponse]: + """ + Lists all domains that configured the specified pseudonym suffix. + + :param suffix: configured pseudonym suffix + :return: list of domain response instances + """ + return get_domains_for_suffix(self._domain_client, suffix) + def list_domains(self) -> list[DomainResponse]: """ Lists all available domains in the gPAS instance. @@ -340,3 +677,136 @@ def delete_domain(self, domain_name: str) -> None: :param domain_name: name of the domain to delete """ delete_domain(self._domain_client, domain_name) + + def validate_pseudonym(self, pseudonym: str, domain_name: str) -> None: + """ + Validates a given pseudonym against the specified domain. Raise a Fault if the pseudonym is not valid. + + :param pseudonym: pseudonym to validate + :param domain_name: domain for which the pseudonym should be validated + """ + validate_pseudonym(self._client, pseudonym, domain_name) + + def update_pseudonym_expiration_date(self, pseudonym: str, domain_name: str, expiration_date: date) -> None: + """ + Updates the expiration date of a pseudonym. Raises a Fault if there is no domain with the specified name, the + expiration date is invalid, there is no such pseudonym or the domain has not been configured for expirations. + + :param pseudonym: pseudonym to update + :param domain_name: domain which the pseudonym belongs to + :param expiration_date: new expiration date of the pseudonym + """ + update_pseudonym_expiration_date(self._client, pseudonym, domain_name, expiration_date) + + def is_anonym(self, value: str) -> bool: + """ + Checks if the given value is an Anonym. + + :param value: value to check + :return: True if the given value is an Anonym, False otherwise + """ + return is_anonym(self._client, value) + + def is_anonymized(self, pseudonym: str, domain_name: str) -> bool: + """ + Checks if the given pseudonym of the specified domain is anonymized. Raises a Fault if the given pseudonym is + not found in the given domain or there is no domain with such a name. + + :param pseudonym: pseudonym to check if it is anonymized + :param domain_name: domain for which the pseudonym should be checked + :return: True if the given pseudonym is anonymized, False otherwise + """ + return is_anonymized(self._client, pseudonym, domain_name) + + def anonymize_pseudonym(self, pseudonym: str, domain_name: str) -> None: + """ + Anonymizes the value of the given pseudonym in the specified domain. Raises a Fault if the pseudonym is not + found in the given domain or there is no domain with such a name. + + :param pseudonym: pseudonym to anonymize + :param domain_name: domain for which the pseudonym should be anonymized + """ + anonymize_pseudonym(self._client, pseudonym, domain_name) + + def anonymize_pseudonyms(self, pseudonyms: list[str], domain_name: str) -> list[AnonymizationResponse]: + """ + Anonymizes the values of the given pseudonyms in the specified domain. Raises a Fault if there is no domain with + such a name. + + :param pseudonyms: list of pseudonyms to anonymize + :param domain_name: domain for which the pseudonyms should be anonymized + :return: list of anonymization response data class instances + """ + return anonymize_pseudonyms(self._client, pseudonyms, domain_name) + + def anonymize_entry(self, value: str, domain_name: str) -> None: + """ + Anonymizes a given value in the specified domain. If the specified domain allows multiple pseudonyms per value, + all values will be anonymized. Raises a Fault if the value is not found in the specified domain, there is no + domain with such a name, or the value is already anonymized. + + :param value: value to anonymize + :param domain_name: domain for which the value should be anonymized + """ + anonymize_entry(self._client, value, domain_name) + + def anonymize_entries(self, values: list[str], domain_name: str) -> list[AnonymizationResponse]: + """ + Anonymizes given values in the specified domain. If the specified domain allows multiple pseudonyms per value, + all values will be anonymized. Raises a Fault if there is no domain with such a name. + + :param values: list of values to anonymize + :param domain_name: domain for which the values should be anonymized + :return: list of anonymization response data class instances + """ + return anonymize_entries(self._client, values, domain_name) + + def get_pseudonyms_for_value_prefix(self, value_prefix: str, domain_name: str) -> list[ValueToPseudonyms]: + """ + Returns all pseudonyms for each value that starts with the specified prefix in the specified domain. Raises a + Fault if there is no domain with such a name. + + :param value_prefix: the prefix of values for which the pseudonyms should be retrieved + :param domain_name: domain for which the pseudonyms should be retrieved + :return: list of value to pseudonyms data class instances + """ + return get_pseudonyms_for_value_prefix(self._client, value_prefix, domain_name) + + def get_pseudonym_tree(self, pseudonym: str, domain_name: str) -> PseudonymTree: + """ + Creates a pseudonym tree with all values that are somehow linked to the given pseudonym. Raises a Fault if the + pseudonym is not found in the specified domain, there is no domain with such a name, or if the value is already + anonymized. + + :param pseudonym: pseudonym to create the tree from + :param domain_name: name of the domain for the given pseudonym + :return: pseudonym tree data class instance + """ + return get_pseudonym_tree(self._client, pseudonym, domain_name) + + def get_pseudonym_net(self, value_or_pseudonym: str) -> PseudonymNet: + """ + Creates a pseudonym net with all values and pseudonyms that are somehow linked to the given value or pseudonym. + + :param value_or_pseudonym: value or pseudonym to create the net from + :return: pseudonym net data class instance + """ + return get_pseudonym_net(self._client, value_or_pseudonym) + + def are_pseudonyms_deletable(self, domain_name: str) -> bool: + """ + Checks if the deletion of pseudonyms is allowed for the specified domain. + + :param domain_name: name of the domain to check + :return: True if pseudonyms are allowed to be deleted, False otherwise + """ + return are_pseudonyms_deletable(self._domain_client, domain_name) + + def list_pseudonyms(self, domain_name: str) -> list[Pseudonym]: + """ + Retrieves all pseudonyms for the specified domain. Raises a Fault if there is no domain with such a name. + + :param domain_name: name of the domain to retrieve pseudonyms for + :return: list of pseudonym data class instances + """ + return list_pseudonyms(self._domain_client, domain_name) diff --git a/mosaic_client/gpas/models.py b/mosaic_client/gpas/models.py index e886df7..30fceaa 100644 --- a/mosaic_client/gpas/models.py +++ b/mosaic_client/gpas/models.py @@ -1,18 +1,15 @@ from dataclasses import dataclass, field -from datetime import datetime +from datetime import date, datetime from typing import Literal -Alphabet = ( - Literal[ - "org.emau.icmvc.ganimed.ttp.psn.alphabets.Hex", - "org.emau.icmvc.ganimed.ttp.psn.alphabets.Numbers", - "org.emau.icmvc.ganimed.ttp.psn.alphabets.NumbersWithoutZero", - "org.emau.icmvc.ganimed.ttp.psn.alphabets.NumbersX", - "org.emau.icmvc.ganimed.ttp.psn.alphabets.Symbol31", - "org.emau.icmvc.ganimed.ttp.psn.alphabets.Symbol32", - ] - | str -) +Alphabet = Literal[ + "org.emau.icmvc.ganimed.ttp.psn.alphabets.Hex", + "org.emau.icmvc.ganimed.ttp.psn.alphabets.Numbers", + "org.emau.icmvc.ganimed.ttp.psn.alphabets.NumbersWithoutZero", + "org.emau.icmvc.ganimed.ttp.psn.alphabets.NumbersX", + "org.emau.icmvc.ganimed.ttp.psn.alphabets.Symbol31", + "org.emau.icmvc.ganimed.ttp.psn.alphabets.Symbol32", +] ForceCache = Literal["DEFAULT", "OFF", "ON"] ValidateViaParents = Literal["CASCADE_DELETE", "ENSURE_EXISTS", "OFF", "VALIDATE"] @@ -46,7 +43,7 @@ class Domain: label: str name: str config: DomainConfig = field(default_factory=DomainConfig) - alphabet: Alphabet = "org.emau.icmvc.ganimed.ttp.psn.alphabets.Numbers" + alphabet: Alphabet | str = "org.emau.icmvc.ganimed.ttp.psn.alphabets.Numbers" check_digit_class: str = "org.emau.icmvc.ganimed.ttp.psn.generator.Verhoeff" comment: str | None = None parent_domain_names: list[str] = field(default_factory=list) @@ -68,3 +65,105 @@ class DomainResponse(Domain): update_date: datetime create_date_string: str update_date_string: str + + +@dataclass(frozen=True) +class AnonymizationResponse: + """ + In case of anonymizing a list of values, the response for each value is stored in this data class. The key is either + represents the value or the pseudonym for which the value was anonymized. + """ + + key: str + result: Literal["SUCCESS", "NOT_FOUND", "ALREADY_ANONYMISED", "ERROR"] + + +@dataclass(frozen=True) +class DeletionResponse: + """ + In case of deleting a list of entries, the response for each value is stored in this data class. + """ + + key: str + result: Literal["SUCCESS", "NOT_FOUND", "ERROR"] + + +@dataclass(frozen=True) +class ValueToPseudonyms: + """ + Describes the relation from one value to its various pseudonyms. + """ + + value: str + pseudonyms: list[str] + + @property + def pseudonym(self) -> str: + if len(self.pseudonyms) != 1: + raise ValueError("There are more than one pseudonyms for this value.") + return self.pseudonyms[0] + + +@dataclass(frozen=True) +class PseudonymTree: + """ + Describes the relation of a pseudonym across related domains. + """ + + domain_name: str + original_value: str | None + pseudonym: str | None + expiration_date: date | None + level: int + path: str | None + children: list["PseudonymTree"] + + +@dataclass(frozen=True) +class PseudonymNetNode(PseudonymTree): + """ + Describes a node in a pseudonym/value net. + """ + + children: list["PseudonymNetNode"] + circle_children: list["PseudonymNetNode"] + + +@dataclass(frozen=True) +class PseudonymNet: + """ + Describes all related values and pseudonyms for a specific pseudonym or value as a root. + """ + + root: PseudonymNetNode + nodes: list[PseudonymNetNode] + + +@dataclass(frozen=True) +class InsertPairException: + """ + Container that describes an error that could occur while inserting a value, pseudonym pair. + """ + + message: str + value: str + pseudonym: str + domain: str + error_type: Literal[ + "DIFFERENT_PSEUDONYM_FOR_VALUE_EXISTS", + "DIFFERENT_VALUE_FOR_PSEUDONYM_EXISTS", + "PSEUDONYM_INVALID", + "VALUE_INVALID", + ] + + +@dataclass(frozen=True) +class Pseudonym: + """ + Container for holding metadata related to a pseudonym. + """ + + domain_name: str + original_value: str + pseudonym: str + expiration_date: date diff --git a/mosaic_client/gpas/schemas.py b/mosaic_client/gpas/schemas.py index 5f9c693..1bfc6d9 100644 --- a/mosaic_client/gpas/schemas.py +++ b/mosaic_client/gpas/schemas.py @@ -1,6 +1,18 @@ from marshmallow import Schema, fields, post_load -from mosaic_client.gpas.models import Domain, DomainConfig, DomainResponse +from mosaic_client.gpas.models import ( + AnonymizationResponse, + DeletionResponse, + Domain, + DomainConfig, + DomainResponse, + InsertPairException, + Pseudonym, + PseudonymNet, + PseudonymNetNode, + PseudonymTree, + ValueToPseudonyms, +) class DomainConfigSchema(Schema): @@ -51,3 +63,87 @@ class DomainResponseSchema(DomainSchema): @post_load def make_domain(self, data, **kwargs) -> DomainResponse: return DomainResponse(**data) + + +class AnonymizationResponseSchema(Schema): + key = fields.Str(required=True) + result = fields.Str(required=True, data_key="value") + + @post_load + def make_anonymization_response(self, data, **kwargs) -> AnonymizationResponse: + return AnonymizationResponse(**data) + + +class DeletionResponseSchema(Schema): + key = fields.Str(required=True) + result = fields.Str(required=True, data_key="value") + + @post_load + def make_deletion_response(self, data, **kwargs) -> DeletionResponse: + return DeletionResponse(**data) + + +class ValueToPseudonymsSchema(Schema): + value = fields.Str(required=True) + pseudonyms = fields.List(fields.Str(), data_key="psn", required=True) + + @post_load + def make_value_to_pseudonyms(self, data, **kwargs) -> ValueToPseudonyms: + return ValueToPseudonyms(**data) + + +class PseudonymTreeSchema(Schema): + domain_name = fields.Str(required=True, data_key="domainName") + original_value = fields.Str(data_key="originalValue", load_default=None) + pseudonym = fields.Str(load_default=None) + expiration_date = fields.Date(data_key="expirationDate", load_default=None) + level = fields.Int(required=True) + path = fields.Str(load_default=None) + children = fields.List(fields.Nested("PseudonymTreeSchema")) + + @post_load + def make_pseudonym_tree(self, data, **kwargs) -> PseudonymTree: + return PseudonymTree(**data) + + +class PseudonymNetNodeSchema(PseudonymTreeSchema): + children = fields.List(fields.Nested("PseudonymNetNodeSchema")) + circle_children = fields.List(fields.Nested("PseudonymNetNodeSchema"), data_key="circleChildren", load_default=list) + + make_pseudonym_tree = None # Otherwise this will be called on load(). + + @post_load + def make_pseudonym_net_node(self, data, **kwargs) -> PseudonymNetNode: + return PseudonymNetNode(**data) + + +class PseudonymNetSchema(Schema): + root = fields.Nested(PseudonymNetNodeSchema, required=True) + nodes = fields.List(fields.Nested(PseudonymNetNodeSchema)) + + @post_load + def make_pseudonym_net(self, data, **kwargs) -> PseudonymNet: + return PseudonymNet(**data) + + +class InsertPairExceptionSchema(Schema): + domain = fields.Str(required=True) + pseudonym = fields.Str(required=True) + value = fields.Str(required=True) + message = fields.Str(required=True) + error_type = fields.Str(required=True, data_key="errorType") + + @post_load + def make_insert_pair_exception(self, data, **kwargs) -> InsertPairException: + return InsertPairException(**data) + + +class PseudonymSchema(Schema): + domain_name = fields.Str(required=True, data_key="domainName") + original_value = fields.Str(required=True, data_key="originalValue") + pseudonym = fields.Str(required=True) + expiration_date = fields.Date(data_key="expirationDate", load_default=None) + + @post_load + def make_pseudonym(self, data, **kwargs) -> Pseudonym: + return Pseudonym(**data) diff --git a/tests/conftest.py b/tests/conftest.py index c2530b5..b685cfe 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,7 +12,7 @@ from mosaic_client.epix import Domain as EPIXDomain from mosaic_client.epix import IdentifierDomain, Person, Source from mosaic_client.gpas import Domain as GPASDomain -from mosaic_client.gpas import DomainConfig +from mosaic_client.gpas import DomainConfig, ValueToPseudonyms from tests.helpers import random_identity, random_string @@ -257,16 +257,16 @@ def person(epix_client, epix_domain, epix_source) -> Iterator[Person]: @pytest.fixture(scope="session") -def value_psn_pair_factory(gpas_client, gpas_domain) -> Callable[[], tuple[str, str]]: - def factory() -> tuple[str, str]: +def value_psn_pair_factory(gpas_client, gpas_domain) -> Callable[[], ValueToPseudonyms]: + def factory() -> ValueToPseudonyms: value = random_string() - psn = gpas_client.get_or_create_pseudonym_for(domain_name=gpas_domain, value=value) + relation = gpas_client.get_or_create_pseudonyms_for(domain_name=gpas_domain, value=value) - return value, psn + return relation return factory @pytest.fixture() -def value_psn_pair(value_psn_pair_factory) -> tuple[str, str]: +def value_psn_pair(value_psn_pair_factory) -> ValueToPseudonyms: return value_psn_pair_factory() diff --git a/tests/helpers.py b/tests/helpers.py index 05dff3f..24bfae9 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -19,3 +19,8 @@ def random_identity() -> Identity: birth_date=datetime.datetime.now(tz=datetime.UTC), gender=random.choice(["m", "f"]), ) + + +def random_date(max_days: int = 100) -> datetime.date: + delta = random.randint(1, max_days) + return datetime.datetime.now(tz=datetime.UTC) + datetime.timedelta(days=delta) diff --git a/tests/test_gpas_client.py b/tests/test_gpas_client.py index da9df7c..44a1285 100644 --- a/tests/test_gpas_client.py +++ b/tests/test_gpas_client.py @@ -1,112 +1,148 @@ import pytest from zeep.exceptions import Fault -from tests.helpers import random_string +from mosaic_client.gpas import ( + DeletionResponse, + Domain, + DomainConfig, + DomainResponse, + Pseudonym, + PseudonymNet, + PseudonymTree, + ValueToPseudonyms, +) +from tests.helpers import random_date, random_string def test_get_or_create_pseudonym_for(gpas_client, gpas_domain): - psn = gpas_client.get_or_create_pseudonym_for(domain_name=gpas_domain, value=random_string()) + relation = gpas_client.get_or_create_pseudonyms_for(domain_name=gpas_domain, value=random_string()) - assert isinstance(psn, str) + assert isinstance(relation, ValueToPseudonyms) def test_get_or_create_pseudonym_for_list(gpas_client, gpas_domain): values = [random_string() for _ in range(5)] - key_value_pairs = gpas_client.get_or_create_pseudonym_for_list( + response = gpas_client.get_or_create_pseudonyms_for_list( domain_name=gpas_domain, values=values, ) - assert set(values) == {kv[0] for kv in key_value_pairs} - assert all(isinstance(kv[1], str) for kv in key_value_pairs) + assert set(values) == {item.value for item in response} + assert all(isinstance(item.pseudonyms, list) for item in response) def test_get_value_for(gpas_client, gpas_domain, value_psn_pair): - value, psn = value_psn_pair - fetched_value = gpas_client.get_value_for(domain_name=gpas_domain, pseudonym=psn) + fetched_value = gpas_client.get_value_for(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) - assert fetched_value == value + assert fetched_value == value_psn_pair.value def test_get_value_for_list(gpas_client, gpas_domain, value_psn_pair_factory): pairs = [value_psn_pair_factory() for _ in range(5)] fetched_pairs = gpas_client.get_value_for_list( domain_name=gpas_domain, - pseudonyms=[pair[1] for pair in pairs], + pseudonyms=[pair.pseudonym for pair in pairs], ) - assert set(pairs) == {(psn, value) for value, psn in fetched_pairs} + assert {pair.value for pair in pairs} == {value for _, value in fetched_pairs} def test_get_pseudonym_for(gpas_client, gpas_domain, value_psn_pair): - value, psn = value_psn_pair - fetched_psn = gpas_client.get_pseudonym_for(domain_name=gpas_domain, value=value) + fetched = gpas_client.get_pseudonyms_for(domain_name=gpas_domain, value=value_psn_pair.value) - assert fetched_psn == psn + assert fetched.pseudonyms == value_psn_pair.pseudonyms def test_get_pseudonym_for_list(gpas_client, gpas_domain, value_psn_pair_factory): pairs = [value_psn_pair_factory() for _ in range(5)] - fetched_pairs = gpas_client.get_or_create_pseudonym_for_list( + fetched_pairs = gpas_client.get_or_create_pseudonyms_for_list( domain_name=gpas_domain, - values=[pair[0] for pair in pairs], + values=[pair.value for pair in pairs], ) - assert set(pairs) == set(fetched_pairs) + assert {pair.pseudonym for pair in pairs} == {pair.pseudonym for pair in fetched_pairs} def test_delete_entry(gpas_client, gpas_domain, value_psn_pair): - value, _ = value_psn_pair - gpas_client.delete_entry(domain_name=gpas_domain, value=value) + gpas_client.delete_entry(domain_name=gpas_domain, value=value_psn_pair.value) with pytest.raises(Fault) as e: - gpas_client.get_pseudonym_for(domain_name=gpas_domain, value=value) + gpas_client.get_pseudonyms_for(domain_name=gpas_domain, value=value_psn_pair.value) - assert str(e.value) == f"value {value} for domain {gpas_domain} not found" + assert str(e.value) == f"value {value_psn_pair.value} for domain {gpas_domain} not found" def test_delete_entries(gpas_client, gpas_domain, value_psn_pair_factory): pairs = [value_psn_pair_factory() for _ in range(5)] - gpas_client.delete_entries(domain_name=gpas_domain, values=[pair[0] for pair in pairs]) + gpas_client.delete_entries(domain_name=gpas_domain, values=[pair.value for pair in pairs]) - for value, _ in pairs: + for pair in pairs: with pytest.raises(Fault) as e: - gpas_client.get_pseudonym_for(domain_name=gpas_domain, value=value) + gpas_client.get_pseudonyms_for(domain_name=gpas_domain, value=pair.value) - assert str(e.value) == f"value {value} for domain {gpas_domain} not found" + assert str(e.value) == f"value {pair.value} for domain {gpas_domain} not found" + + +def test_delete_pseudonym(gpas_client, gpas_domain, value_psn_pair): + gpas_client.delete_pseudonym(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) + + with pytest.raises(Fault) as e: + gpas_client.get_value_for(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) + + assert str(e.value) == f"value for pseudonym {value_psn_pair.pseudonym} not found within domain {gpas_domain}" + + +def test_delete_pseudonyms(gpas_client, gpas_domain, value_psn_pair_factory): + pairs = [value_psn_pair_factory() for _ in range(5)] + responses = gpas_client.delete_pseudonyms(domain_name=gpas_domain, pseudonyms=[pair.pseudonym for pair in pairs]) + + assert all(isinstance(response, DeletionResponse) for response in responses) + assert all(response.result == "SUCCESS" for response in responses) def test_insert_value_pseudonym_pair(gpas_client, gpas_domain, value_psn_pair): - value, psn = value_psn_pair # Let gPAS create the pseudonym so it is valid for the domain when inserting it again. + # Let gPAS create the pseudonym so it is valid for the domain when inserting it again. # Delete the entry first to check if the insertion really works. - gpas_client.delete_entry(domain_name=gpas_domain, value=value) + gpas_client.delete_entry(domain_name=gpas_domain, value=value_psn_pair.value) - gpas_client.insert_value_pseudonym_pair(domain_name=gpas_domain, value=value, pseudonym=psn) + gpas_client.insert_value_pseudonym_pair( + domain_name=gpas_domain, + value=value_psn_pair.value, + pseudonym=value_psn_pair.pseudonym, + ) - assert gpas_client.get_pseudonym_for(domain_name=gpas_domain, value=value) == psn - assert gpas_client.get_value_for(domain_name=gpas_domain, pseudonym=psn) == value + assert gpas_client.get_pseudonyms_for(domain_name=gpas_domain, value=value_psn_pair.value) == value_psn_pair + assert ( + gpas_client.get_value_for( + domain_name=gpas_domain, + pseudonym=value_psn_pair.pseudonym, + ) + == value_psn_pair.value + ) def test_insert_value_pseudonym_pairs(gpas_client, gpas_domain, value_psn_pair_factory): pairs = [value_psn_pair_factory() for _ in range(5)] - gpas_client.delete_entries(domain_name=gpas_domain, values=[pair[0] for pair in pairs]) + gpas_client.delete_entries(domain_name=gpas_domain, values=[pair.value for pair in pairs]) - gpas_client.insert_value_pseudonym_pairs(domain_name=gpas_domain, pairs=pairs) + gpas_client.insert_value_pseudonym_pairs( + domain_name=gpas_domain, + pairs=[(pair.value, pair.pseudonym) for pair in pairs], + ) - assert set( - gpas_client.get_pseudonym_for_list( - domain_name=gpas_domain, - values=[pair[0] for pair in pairs], - ) - ) == set(pairs) - assert set( - gpas_client.get_value_for_list( - domain_name=gpas_domain, - pseudonyms=[pair[1] for pair in pairs], - ) - ) == {(psn, value) for value, psn in pairs} + search_for_pseudonyms = gpas_client.get_pseudonyms_for_list( + domain_name=gpas_domain, + values=[pair.value for pair in pairs], + ) + assert {relation.pseudonym for relation in search_for_pseudonyms} == {pair.pseudonym for pair in pairs} + + search_for_values = gpas_client.get_value_for_list( + domain_name=gpas_domain, + pseudonyms=[pair.pseudonym for pair in pairs], + ) + assert {value for _, value in search_for_values} == {pair.value for pair in pairs} def test_get_domain(gpas_client, gpas_domain): @@ -121,3 +157,131 @@ def test_list_domains(gpas_client, gpas_domain): assert gpas_domain in [d.name for d in fetched_domains] assert gpas_domain in [d.label for d in fetched_domains] + + +def test_validate_pseudonym(gpas_client, gpas_domain, value_psn_pair): + # This should just not raise an error. + gpas_client.validate_pseudonym(pseudonym=value_psn_pair.pseudonym, domain_name=gpas_domain) + + with pytest.raises(Fault) as e: + psn = random_string() + gpas_client.validate_pseudonym(pseudonym=psn, domain_name=gpas_domain) + + assert "invalid value" in str(e.value) + + +def test_is_anonym(gpas_client, gpas_domain, value_psn_pair): + assert gpas_client.is_anonym(value=value_psn_pair.value) is False + + +def test_is_anonymized(gpas_client, gpas_domain, value_psn_pair): + assert gpas_client.is_anonymized(pseudonym=value_psn_pair.pseudonym, domain_name=gpas_domain) is False + + +def test_update_pseudonym_expiration_date(gpas_client, gpas_domain, value_psn_pair): + with pytest.raises(Fault) as e: + gpas_client.update_pseudonym_expiration_date( + domain_name=gpas_domain, + pseudonym=value_psn_pair.pseudonym, + expiration_date=random_date(), + ) + + assert str(e.value) == f"the domain {gpas_domain} does not allow pseudonyms with expiration date" + + +def test_anonymize_pseudonym(gpas_client, gpas_domain, value_psn_pair): + gpas_client.anonymize_pseudonym(pseudonym=value_psn_pair.pseudonym, domain_name=gpas_domain) + + assert gpas_client.is_anonymized(pseudonym=value_psn_pair.pseudonym, domain_name=gpas_domain) is True + + # Delete anonymized entry because otherwise no new pseudonyms can be added. + gpas_client.delete_pseudonym(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) + + +def test_anonymize_pseudonyms(gpas_client, gpas_domain, value_psn_pair_factory): + pseudonyms = [value_psn_pair_factory().pseudonym for _ in range(5)] + gpas_client.anonymize_pseudonyms(pseudonyms=pseudonyms, domain_name=gpas_domain) + + assert all(gpas_client.is_anonymized(pseudonym=psn, domain_name=gpas_domain) for psn in pseudonyms) + + # Delete anonymized entries because otherwise no new pseudonyms can be added. + gpas_client.delete_pseudonyms(domain_name=gpas_domain, pseudonyms=pseudonyms) + + +def test_anonymize_entry(gpas_client, gpas_domain, value_psn_pair): + gpas_client.anonymize_entry(domain_name=gpas_domain, value=value_psn_pair.value) + + assert gpas_client.is_anonymized(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) is True + + # Delete anonymized entry because otherwise no new pseudonyms can be added. + gpas_client.delete_pseudonym(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) + + +def test_anonymize_entries(gpas_client, gpas_domain, value_psn_pair_factory): + pairs = [value_psn_pair_factory() for _ in range(5)] + gpas_client.anonymize_entries(domain_name=gpas_domain, values=[pair.value for pair in pairs]) + + assert all(gpas_client.is_anonymized(domain_name=gpas_domain, pseudonym=pair.pseudonym) for pair in pairs) + + # Delete anonymized entries because otherwise no new pseudonyms can be added. + gpas_client.delete_pseudonyms(domain_name=gpas_domain, pseudonyms=[pair.pseudonym for pair in pairs]) + + +def test_get_pseudonyms_for_value_prefix(gpas_client, gpas_domain, value_psn_pair): + fetched_pairs = gpas_client.get_pseudonyms_for_value_prefix( + domain_name=gpas_domain, + value_prefix=value_psn_pair.value[0:3], + ) + + assert len(fetched_pairs) >= 1 + assert all(isinstance(pair, ValueToPseudonyms) for pair in fetched_pairs) + assert any(pair.value == value_psn_pair.value for pair in fetched_pairs) + + +def test_get_pseudonym_tree(gpas_client, gpas_domain, value_psn_pair): + tree = gpas_client.get_pseudonym_tree(domain_name=gpas_domain, pseudonym=value_psn_pair.pseudonym) + + assert isinstance(tree, PseudonymTree) + + +def test_get_pseudonym_net(gpas_client, value_psn_pair): + net = gpas_client.get_pseudonym_net(value_or_pseudonym=value_psn_pair.value) + + assert isinstance(net, PseudonymNet) + + +def test_get_domains_for_prefix(gpas_client, gpas_domain): + prefix, name = random_string(), random_string() + gpas_client.add_domain(domain=Domain(name=name, label=name, config=DomainConfig(psn_prefix=prefix))) + + domains = gpas_client.get_domains_for_prefix(prefix=prefix) + + assert len(domains) == 1 + assert isinstance(domains[0], DomainResponse) + assert domains[0].config.psn_prefix == prefix + + gpas_client.delete_domain(domain_name=name) + + +def test_get_domains_for_suffix(gpas_client, gpas_domain): + suffix, name = random_string(), random_string() + gpas_client.add_domain(domain=Domain(name=name, label=name, config=DomainConfig(psn_suffix=suffix))) + + domains = gpas_client.get_domains_for_suffix(suffix=suffix) + + assert len(domains) == 1 + assert isinstance(domains[0], DomainResponse) + assert domains[0].config.psn_suffix == suffix + + gpas_client.delete_domain(domain_name=name) + + +def test_are_pseudonyms_deletable(gpas_client, gpas_domain): + assert gpas_client.are_pseudonyms_deletable(domain_name=gpas_domain) is True + + +def test_list_pseudonyms(gpas_client, gpas_domain, value_psn_pair): + psns = gpas_client.list_pseudonyms(domain_name=gpas_domain) + + assert all(isinstance(psn, Pseudonym) for psn in psns) + assert any(psn.pseudonym == value_psn_pair.pseudonym for psn in psns) From 6abbc6e413e4f112c2833fdc2d9b3a68a7385735 Mon Sep 17 00:00:00 2001 From: pbrassel <52356233+pbrassel@users.noreply.github.com> Date: Tue, 6 Oct 2026 12:18:48 +0200 Subject: [PATCH 2/3] chore: bump package version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index ca496e1..7f2b073 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "mosaic-python-client" -version = "0.2.0" +version = "0.3.0" description = "Zeep client for interacting with the SOAP interfaces provided by E-PIX and gPAS of the MOSAIC suite by the THS Greifswald." authors = [ {name = "Maximilian Jugl"}, From a3531aef1a9808ae2594d6eacd0b0b909cd3c2d8 Mon Sep 17 00:00:00 2001 From: pbrassel <52356233+pbrassel@users.noreply.github.com> Date: Tue, 6 Oct 2026 12:20:59 +0200 Subject: [PATCH 3/3] test(fix): let `random_date()` return a `date` object --- tests/helpers.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/helpers.py b/tests/helpers.py index 24bfae9..232af4c 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -23,4 +23,4 @@ def random_identity() -> Identity: def random_date(max_days: int = 100) -> datetime.date: delta = random.randint(1, max_days) - return datetime.datetime.now(tz=datetime.UTC) + datetime.timedelta(days=delta) + return datetime.datetime.now(tz=datetime.UTC).date() + datetime.timedelta(days=delta)