diff --git a/tests/fixtures/saas/stripe_fixtures.py b/tests/fixtures/saas/stripe_fixtures.py index 7ff6bb27dc..993178fd2a 100644 --- a/tests/fixtures/saas/stripe_fixtures.py +++ b/tests/fixtures/saas/stripe_fixtures.py @@ -21,6 +21,7 @@ load_dataset_with_replacement, ) from tests.ops.test_helpers.vault_client import get_secrets +from tests.ops.integration_tests.saas.connector_runner import ConnectorRunner, generate_random_email secrets = get_secrets("stripe") @@ -35,12 +36,12 @@ def stripe_secrets(saas_config): } -@pytest.fixture(scope="session") +@pytest.fixture def stripe_identity_email(saas_config): return pydash.get(saas_config, "stripe.identity_email") or secrets["identity_email"] -@pytest.fixture(scope="session") +@pytest.fixture def stripe_identity_phone_number(saas_config): return ( pydash.get(saas_config, "stripe.identity_phone_number") @@ -50,7 +51,7 @@ def stripe_identity_phone_number(saas_config): @pytest.fixture(scope="session") def stripe_erasure_identity_email(): - return f"{cryptographic_util.generate_secure_random_string(13)}@email.com" + return generate_random_email() @pytest.fixture @@ -115,18 +116,300 @@ def stripe_dataset_config( ctl_dataset.delete(db=db) +class StripeTestClient: + + def __init__(self, stripe_secrets: Dict[str, Any]): + self.base_url = f"https://{stripe_secrets['domain']}" + self.headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Authorization": f"Bearer {stripe_secrets['api_key']}", + } + + def create_customer(self, customer_data: Dict[str, Any]) -> Dict[str, Any]: + + response = requests.post( + url=f"{self.base_url}/v1/customers", + headers=self.headers, + data=multidimensional_urlencode(customer_data), + ) + assert response.ok + return response.json() + + def create_dispute(self, customer_id, customer_data): + # create dispute by adding a fraudulent card and charging it + response = requests.post( + url=f"{self.base_url}/v1/customers/{customer_id}", + headers=self.headers, + data=multidimensional_urlencode({"source": "tok_createDispute"}), + ) + assert response.ok + card = response.json()["sources"]["data"][0] + card_id = card["id"] + + # update card name to have something to mask + response = requests.post( + url=f"{self.base_url}/v1/customers/{customer_id}/sources/{card_id}", + headers=self.headers, + data=multidimensional_urlencode({"name": customer_data["name"]}), + ) + assert response.ok + + + # charge + response = requests.post( + url=f"{self.base_url}/v1/charges", + headers=self.headers, + data=multidimensional_urlencode( + { + "customer": customer_id, + "source": card_id, + "amount": 1000, + "currency": "usd", + } + ), + ) + assert response.ok + + # charge + response = requests.post( + url=f"{self.base_url}/v1/charges", + headers=self.headers, + data=multidimensional_urlencode( + { + "customer": customer_id, + "source": card_id, + "amount": 1000, + "currency": "usd", + } + ), + ) + assert response.ok + + def create_bank_account(self, customer_id, customer_data): + response = requests.post( + url=f"{self.base_url}/v1/customers/{customer_id}/sources", + headers=self.headers, + data=multidimensional_urlencode({"source": "btok_us_verified"}), + ) + assert response.ok + bank_account = response.json() + bank_account_id = bank_account["id"] + # update bank account holder name to have something to mask + response = requests.post( + url=f"{self.base_url}/v1/customers/{customer_id}/sources/{bank_account_id}", + headers=self.headers, + data=multidimensional_urlencode({"account_holder_name": customer_data["name"]}), + ) + assert response.ok + + def create_invoice(self, customer_id): + # invoice item + response = requests.post( + url=f"{self.base_url}/v1/invoiceitems", + headers=self.headers, + params={"customer": customer_id}, + data=multidimensional_urlencode({"amount": 200, "currency": "usd"}), + ) + assert response.ok + + # pulls in the previously created invoice item automatically to create the invoice + response = requests.post( + url=f"{self.base_url}/v1/invoices", + headers=self.headers, + params={"customer": customer_id}, + ) + assert response.ok + invoice = response.json() + invoice_id = invoice["id"] + + # finalize invoice + response = requests.post( + url=f"{self.base_url}/v1/invoices/{invoice_id}/finalize", headers=self.headers + ) + assert response.ok + + return invoice + + def create_credit_note(self, invoice): + response = requests.post( + url=f"{self.base_url}/v1/credit_notes", + headers=self.headers, + params={"invoice": invoice["id"]}, + data=multidimensional_urlencode( + { + "lines[0]": { + "type": "invoice_line_item", + "invoice_line_item": invoice["lines"]["data"][0]["id"], + "quantity": 1, + } + } + ), + ) + assert response.ok + + def create_balance_transaction(self, customer_id): + response = requests.post( + url=f"{self.base_url}/v1/customers/{customer_id}/balance_transactions", + headers=self.headers, + data=multidimensional_urlencode({"amount": -500, "currency": "usd"}), + ) + assert response.ok + + def create_payment_intent(self, customer_id): + response = requests.post( + url=f"{self.base_url}/v1/payment_intents", + headers=self.headers, + data=multidimensional_urlencode( + { + "customer": customer_id, + "amount": 2000, + "currency": "usd", + "payment_method_types[]": "card", + "confirm": True, + } + ), + ) + assert response.ok + + response = requests.post( + url=f"{self.base_url}/v1/setup_intents", + params={"customer": customer_id, "payment_method_types[]": "card"}, + headers=self.headers, + ) + assert response.ok + + def create_payment_method(self, customer_id, customer_name): + response = requests.post( + url=f"{self.base_url}/v1/payment_methods", + headers=self.headers, + data=multidimensional_urlencode( + { + "type": "card", + "card": { + "number": 4242424242424242, + "exp_month": 4, + "exp_year": datetime.today().year + 1, + "cvc": 314, + }, + "billing_details": {"name": customer_name}, + } + ), + ) + assert response.ok + payment_method = response.json() + payment_method_id = payment_method["id"] + + response = requests.post( + url=f"{self.base_url}/v1/payment_methods/{payment_method_id}/attach", + params={"customer": customer_id}, + headers=self.headers, + ) + assert response.ok + + def create_subscription(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/prices", + params={"type": "recurring"}, + headers=self.headers, + ) + assert response.ok + price = response.json()["data"][0] + price_id = price["id"] + + response = requests.post( + url=f"{self.base_url}/v1/subscriptions", + headers=self.headers, + data=multidimensional_urlencode( + {"customer": customer_id, "items[0]": {"price": price_id}} + ), + ) + assert response.ok + + def create_tax(self, customer_id): + # tax id + response = requests.post( + url=f"{self.base_url}/v1/customers/{customer_id}/tax_ids", + headers=self.headers, + data=multidimensional_urlencode({"type": "us_ein", "value": "000000000"}), + ) + assert response.ok + + def delete_customer(self, customer_id): + response = requests.delete( + url=f"{self.base_url}/v1/customers/{customer_id}", headers=self.headers + ) + assert response.ok + + def get_customer(self, email): + response = requests.get( + url=f"{self.base_url}/v1/customers", + headers=self.headers, + params={"email": email}, + ) + customer = response.json()["data"][0] + return customer + + def get_card(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/customers/{customer_id}/sources", + headers=self.headers, + params={"object": "card"}, + ) + cards = response.json()["data"] + return cards + + def get_payment_method(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/customers/{customer_id}/payment_methods", + headers=self.headers, + params={"type": "card"}, + ) + payment_methods = response.json()["data"] + return payment_methods + + def get_bank_account(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/customers/{customer_id}/sources", + headers=self.headers, + params={"object": "bank_account"}, + ) + bank_account = response.json()["data"][0] + return bank_account + + def get_tax_ids(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/customers/{customer_id}/tax_ids", headers=self.headers + ) + tax_ids = response.json()["data"] + return tax_ids + + def get_invoice_items(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/invoiceitems", + headers=self.headers, + params={"customer": {customer_id}}, + ) + invoice_item = response.json()["data"] + return invoice_item + + def get_subscription(self, customer_id): + response = requests.get( + url=f"{self.base_url}/v1/customers/{customer_id}/subscriptions", headers=self.headers + ) + subscriptions = response.json()["data"] + return subscriptions + +@pytest.fixture(scope="function") +def stripe_test_client( + stripe_secrets, +) -> Generator: + test_client = StripeTestClient(stripe_secrets) + yield test_client + @pytest.fixture(scope="function") def stripe_create_erasure_data( - stripe_connection_config: ConnectionConfig, stripe_erasure_identity_email + stripe_test_client: StripeTestClient, stripe_erasure_identity_email: str ) -> Generator: - stripe_secrets = stripe_connection_config.secrets - - base_url = f"https://{stripe_secrets['domain']}" - - headers = { - "Content-Type": "application/x-www-form-urlencoded", - "Authorization": f"Bearer {stripe_secrets['api_key']}", - } # customer customer_data = { @@ -156,197 +439,277 @@ def stripe_create_erasure_data( }, } - response = requests.post( - url=f"{base_url}/v1/customers", - headers=headers, - data=multidimensional_urlencode(customer_data), - ) - assert response.ok - customer = response.json() + + customer = stripe_test_client.create_customer(customer_data) customer_id = customer["id"] - # create dispute by adding a fraudulent card and charging it - response = requests.post( - url=f"{base_url}/v1/customers/{customer['id']}", - headers=headers, - data=multidimensional_urlencode({"source": "tok_createDispute"}), - ) - assert response.ok - card = response.json()["sources"]["data"][0] - card_id = card["id"] - - # update card name to have something to mask - response = requests.post( - url=f"{base_url}/v1/customers/{customer_id}/sources/{card_id}", - headers=headers, - data=multidimensional_urlencode({"name": customer_data["name"]}), - ) - assert response.ok - - # charge - response = requests.post( - url=f"{base_url}/v1/charges", - headers=headers, - data=multidimensional_urlencode( - { - "customer": customer_id, - "source": card_id, - "amount": 1000, - "currency": "usd", - } - ), - ) - assert response.ok + stripe_test_client.create_dispute(customer_id, customer_data) - # bank account - response = requests.post( - url=f"{base_url}/v1/customers/{customer_id}/sources", - headers=headers, - data=multidimensional_urlencode({"source": "btok_us_verified"}), - ) - assert response.ok - bank_account = response.json() - bank_account_id = bank_account["id"] - # update bank account holder name to have something to mask - response = requests.post( - url=f"{base_url}/v1/customers/{customer_id}/sources/{bank_account_id}", - headers=headers, - data=multidimensional_urlencode({"account_holder_name": customer_data["name"]}), - ) - assert response.ok - - # invoice item - response = requests.post( - url=f"{base_url}/v1/invoiceitems", - headers=headers, - params={"customer": customer_id}, - data=multidimensional_urlencode({"amount": 200, "currency": "usd"}), - ) - assert response.ok + stripe_test_client.create_bank_account(customer_id, customer_data) - # pulls in the previously created invoice item automatically to create the invoice - response = requests.post( - url=f"{base_url}/v1/invoices", - headers=headers, - params={"customer": customer_id}, - ) - assert response.ok - invoice = response.json() - invoice_id = invoice["id"] + invoice = stripe_test_client.create_invoice(customer_id) - # finalize invoice - response = requests.post( - url=f"{base_url}/v1/invoices/{invoice_id}/finalize", headers=headers - ) - assert response.ok - - # credit note - response = requests.post( - url=f"{base_url}/v1/credit_notes", - headers=headers, - params={"invoice": invoice_id}, - data=multidimensional_urlencode( - { - "lines[0]": { - "type": "invoice_line_item", - "invoice_line_item": invoice["lines"]["data"][0]["id"], - "quantity": 1, - } - } - ), - ) - assert response.ok + stripe_test_client.create_credit_note(invoice) - # customer balance transaction - response = requests.post( - url=f"{base_url}/v1/customers/{customer_id}/balance_transactions", - headers=headers, - data=multidimensional_urlencode({"amount": -500, "currency": "usd"}), - ) - assert response.ok - - # payment intent - response = requests.post( - url=f"{base_url}/v1/payment_intents", - headers=headers, - data=multidimensional_urlencode( - { - "customer": customer_id, - "amount": 2000, - "currency": "usd", - "payment_method_types[]": "card", - "confirm": True, - } - ), - ) - assert response.ok - - # create and attach payment method to customer - response = requests.post( - url=f"{base_url}/v1/payment_methods", - headers=headers, - data=multidimensional_urlencode( - { - "type": "card", - "card": { - "number": 4242424242424242, - "exp_month": 4, - "exp_year": datetime.today().year + 1, - "cvc": 314, - }, - "billing_details": {"name": customer_data["name"]}, - } - ), - ) - assert response.ok - payment_method = response.json() - payment_method_id = payment_method["id"] - - response = requests.post( - url=f"{base_url}/v1/payment_methods/{payment_method_id}/attach", - params={"customer": customer_id}, - headers=headers, - ) - assert response.ok + stripe_test_client.create_balance_transaction(customer_id) - # setup intent - response = requests.post( - url=f"{base_url}/v1/setup_intents", - params={"customer": customer_id, "payment_method_types[]": "card"}, - headers=headers, - ) - assert response.ok + stripe_test_client.create_payment_intent(customer_id) - # get an existing price and use it to create a subscription - response = requests.get( - url=f"{base_url}/v1/prices", - params={"type": "recurring"}, - headers=headers, - ) - assert response.ok - price = response.json()["data"][0] - price_id = price["id"] - - response = requests.post( - url=f"{base_url}/v1/subscriptions", - headers=headers, - data=multidimensional_urlencode( - {"customer": customer_id, "items[0]": {"price": price_id}} - ), - ) - assert response.ok + stripe_test_client.create_payment_method(customer_id, customer_data["name"]) - # tax id - response = requests.post( - url=f"{base_url}/v1/customers/{customer_id}/tax_ids", - headers=headers, - data=multidimensional_urlencode({"type": "us_ein", "value": "000000000"}), - ) - assert response.ok + stripe_test_client.create_subscription(customer_id) + + stripe_test_client.create_tax(customer_id) yield customer - response = requests.delete( - url=f"{base_url}/v1/customers/{customer_id}", headers=headers + stripe_test_client.delete_customer(customer_id) + +# @pytest.fixture(scope="function") +# def stripe_create_erasure_data( +# stripe_connection_config: ConnectionConfig, stripe_erasure_identity_email +# ) -> Generator: +# stripe_secrets = stripe_connection_config.secrets + +# base_url = f"https://{stripe_secrets['domain']}" + +# headers = { +# "Content-Type": "application/x-www-form-urlencoded", +# "Authorization": f"Bearer {stripe_secrets['api_key']}", +# } + +# # customer +# customer_data = { +# "address": { +# "city": "Anaheim", +# "country": "US", +# "line1": "123 Fake St", +# "line2": "Apt 1", +# "postal_code": "92882", +# "state": "CA", +# }, +# "balance": 0, +# "description": "RTF Test Customer", +# "email": stripe_erasure_identity_email, +# "name": "Ethyca RTF", +# "preferred_locales": ["en-US"], +# "shipping": { +# "address": { +# "city": "Anaheim", +# "country": "US", +# "line1": "123 Fake St", +# "line2": "Apt 1", +# "postal_code": "92882", +# "state": "CA", +# }, +# "name": "Ethyca RTF", +# }, +# } + +# response = requests.post( +# url=f"{base_url}/v1/customers", +# headers=headers, +# data=multidimensional_urlencode(customer_data), +# ) +# assert response.ok +# customer = response.json() +# customer_id = customer["id"] + +# # create dispute by adding a fraudulent card and charging it +# response = requests.post( +# url=f"{base_url}/v1/customers/{customer['id']}", +# headers=headers, +# data=multidimensional_urlencode({"source": "tok_createDispute"}), +# ) +# assert response.ok +# card = response.json()["sources"]["data"][0] +# card_id = card["id"] + +# # update card name to have something to mask +# response = requests.post( +# url=f"{base_url}/v1/customers/{customer_id}/sources/{card_id}", +# headers=headers, +# data=multidimensional_urlencode({"name": customer_data["name"]}), +# ) +# assert response.ok + +# # charge +# response = requests.post( +# url=f"{base_url}/v1/charges", +# headers=headers, +# data=multidimensional_urlencode( +# { +# "customer": customer_id, +# "source": card_id, +# "amount": 1000, +# "currency": "usd", +# } +# ), +# ) +# assert response.ok + +# # bank account +# response = requests.post( +# url=f"{base_url}/v1/customers/{customer_id}/sources", +# headers=headers, +# data=multidimensional_urlencode({"source": "btok_us_verified"}), +# ) +# assert response.ok +# bank_account = response.json() +# bank_account_id = bank_account["id"] +# # update bank account holder name to have something to mask +# response = requests.post( +# url=f"{base_url}/v1/customers/{customer_id}/sources/{bank_account_id}", +# headers=headers, +# data=multidimensional_urlencode({"account_holder_name": customer_data["name"]}), +# ) +# assert response.ok + +# # invoice item +# response = requests.post( +# url=f"{base_url}/v1/invoiceitems", +# headers=headers, +# params={"customer": customer_id}, +# data=multidimensional_urlencode({"amount": 200, "currency": "usd"}), +# ) +# assert response.ok + +# # pulls in the previously created invoice item automatically to create the invoice +# response = requests.post( +# url=f"{base_url}/v1/invoices", +# headers=headers, +# params={"customer": customer_id}, +# ) +# assert response.ok +# invoice = response.json() +# invoice_id = invoice["id"] + +# # finalize invoice +# response = requests.post( +# url=f"{base_url}/v1/invoices/{invoice_id}/finalize", headers=headers +# ) +# assert response.ok + +# # credit note +# response = requests.post( +# url=f"{base_url}/v1/credit_notes", +# headers=headers, +# params={"invoice": invoice_id}, +# data=multidimensional_urlencode( +# { +# "lines[0]": { +# "type": "invoice_line_item", +# "invoice_line_item": invoice["lines"]["data"][0]["id"], +# "quantity": 1, +# } +# } +# ), +# ) +# assert response.ok + +# # customer balance transaction +# response = requests.post( +# url=f"{base_url}/v1/customers/{customer_id}/balance_transactions", +# headers=headers, +# data=multidimensional_urlencode({"amount": -500, "currency": "usd"}), +# ) +# assert response.ok + +# # payment intent +# response = requests.post( +# url=f"{base_url}/v1/payment_intents", +# headers=headers, +# data=multidimensional_urlencode( +# { +# "customer": customer_id, +# "amount": 2000, +# "currency": "usd", +# "payment_method_types[]": "card", +# "confirm": True, +# } +# ), +# ) +# assert response.ok + +# # create and attach payment method to customer +# response = requests.post( +# url=f"{base_url}/v1/payment_methods", +# headers=headers, +# data=multidimensional_urlencode( +# { +# "type": "card", +# "card": { +# "number": 4242424242424242, +# "exp_month": 4, +# "exp_year": datetime.today().year + 1, +# "cvc": 314, +# }, +# "billing_details": {"name": customer_data["name"]}, +# } +# ), +# ) +# assert response.ok +# payment_method = response.json() +# payment_method_id = payment_method["id"] + +# response = requests.post( +# url=f"{base_url}/v1/payment_methods/{payment_method_id}/attach", +# params={"customer": customer_id}, +# headers=headers, +# ) +# assert response.ok + +# # setup intent +# response = requests.post( +# url=f"{base_url}/v1/setup_intents", +# params={"customer": customer_id, "payment_method_types[]": "card"}, +# headers=headers, +# ) +# assert response.ok + +# # get an existing price and use it to create a subscription +# response = requests.get( +# url=f"{base_url}/v1/prices", +# params={"type": "recurring"}, +# headers=headers, +# ) +# assert response.ok +# price = response.json()["data"][0] +# price_id = price["id"] + +# response = requests.post( +# url=f"{base_url}/v1/subscriptions", +# headers=headers, +# data=multidimensional_urlencode( +# {"customer": customer_id, "items[0]": {"price": price_id}} +# ), +# ) +# assert response.ok + +# # tax id +# response = requests.post( +# url=f"{base_url}/v1/customers/{customer_id}/tax_ids", +# headers=headers, +# data=multidimensional_urlencode({"type": "us_ein", "value": "000000000"}), +# ) +# assert response.ok + +# yield customer + +# response = requests.delete( +# url=f"{base_url}/v1/customers/{customer_id}", headers=headers +# ) +# assert response.ok + +@pytest.fixture +def stripe_runner( + db, + cache, + stripe_secrets, +) -> ConnectorRunner: + return ConnectorRunner( + db, + cache, + "stripe", + stripe_secrets, ) - assert response.ok diff --git a/tests/ops/integration_tests/saas/connector_runner.py b/tests/ops/integration_tests/saas/connector_runner.py index 0f9e6e80e9..ff022da0f1 100644 --- a/tests/ops/integration_tests/saas/connector_runner.py +++ b/tests/ops/integration_tests/saas/connector_runner.py @@ -49,6 +49,7 @@ load_dataset_with_replacement, ) from fides.config import get_config +from fides.api.models.application_config import ApplicationConfig CONFIG = get_config() @@ -168,6 +169,7 @@ async def strict_erasure_request( # store the existing masking_strict value so we can reset it at the end of the test masking_strict = CONFIG.execution.masking_strict CONFIG.execution.masking_strict = True + ApplicationConfig.update_config_set(self.db, CONFIG) access_results, erasure_results = await self._base_erasure_request( access_policy, erasure_policy, identities, privacy_request_id @@ -175,6 +177,7 @@ async def strict_erasure_request( # reset masking_strict value CONFIG.execution.masking_strict = masking_strict + ApplicationConfig.update_config_set(self.db, CONFIG) return access_results, erasure_results async def non_strict_erasure_request( @@ -194,6 +197,7 @@ async def non_strict_erasure_request( # store the existing masking_strict value so we can reset it at the end of the test masking_strict = CONFIG.execution.masking_strict CONFIG.execution.masking_strict = False + ApplicationConfig.update_config_set(self.db, CONFIG) access_results, erasure_results = await self._base_erasure_request( access_policy, @@ -205,6 +209,7 @@ async def non_strict_erasure_request( # reset masking_strict value CONFIG.execution.masking_strict = masking_strict + ApplicationConfig.update_config_set(self.db, CONFIG) return access_results, erasure_results async def old_consent_request( diff --git a/tests/ops/integration_tests/saas/test_stripe_task.py b/tests/ops/integration_tests/saas/test_stripe_task.py index 1c51330a88..8c82273107 100644 --- a/tests/ops/integration_tests/saas/test_stripe_task.py +++ b/tests/ops/integration_tests/saas/test_stripe_task.py @@ -12,1205 +12,199 @@ from tests.conftest import access_runner_tester, erasure_runner_tester from tests.ops.graph.graph_test_util import assert_rows_match from tests.ops.test_helpers.cache_secrets_helper import clear_cache_identities +from tests.ops.integration_tests.saas.connector_runner import ConnectorRunner +from fides.api.models.policy import Policy @pytest.mark.integration_saas -def test_stripe_connection_test(stripe_connection_config) -> None: - get_connector(stripe_connection_config).test_connection() - - -@pytest.mark.integration_saas -@pytest.mark.asyncio -@pytest.mark.parametrize( - "dsr_version", - ["use_dsr_3_0", "use_dsr_2_0"], -) -async def test_stripe_access_request_task_with_email( - db, - policy, - dsr_version, - request, - privacy_request, - stripe_connection_config, - stripe_dataset_config, - stripe_identity_email, -) -> None: - """Full access request based on the Stripe SaaS config""" - request.getfixturevalue(dsr_version) # REQUIRED to test both DSR 3.0 and 2.0 - - identity = Identity(**{"email": stripe_identity_email}) - privacy_request.cache_identity(identity) - - dataset_name = stripe_connection_config.get_saas_config().fides_key - merged_graph = stripe_dataset_config.get_graph() - graph = DatasetGraph(merged_graph) - - v = access_runner_tester( - privacy_request, - policy, - graph, - [stripe_connection_config], - {"email": stripe_identity_email}, - db, - ) - - # verify all collections are returned with the expected number of rows and fields - - assert_rows_match( - v[f"{dataset_name}:bank_account"], - min_size=1, - keys=[ - "account_holder_name", - "account_holder_type", - "account_type", - "bank_name", - "country", - "currency", - "customer", - "fingerprint", - "id", - "last4", - "object", - "routing_number", - "status", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:card"], - min_size=1, - keys=[ - "address_city", - "address_country", - "address_line1", - "address_line1_check", - "address_line2", - "address_state", - "address_zip", - "address_zip_check", - "brand", - "country", - "customer", - "cvc_check", - "dynamic_last4", - "exp_month", - "exp_year", - "fingerprint", - "funding", - "id", - "last4", - "name", - "object", - "tokenization_method", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:charge"], - min_size=2, - keys=[ - "amount", - "amount_captured", - "amount_refunded", - "application", - "application_fee", - "application_fee_amount", - "balance_transaction", - "billing_details", - "calculated_statement_descriptor", - "captured", - "created", - "currency", - "customer", - "description", - "disputed", - "failure_balance_transaction", - "failure_code", - "failure_message", - "fraud_details", - "id", - "invoice", - "livemode", - "object", - "on_behalf_of", - "order", - "paid", - "payment_intent", - "payment_method", - "payment_method_details", - "receipt_email", - "receipt_number", - "receipt_url", - "refunded", - "review", - "shipping", - "source_transfer", - "statement_descriptor", - "statement_descriptor_suffix", - "status", - "transfer_data", - "transfer_group", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:credit_note"], - min_size=1, - keys=[ - "amount", - "created", - "currency", - "customer", - "customer_balance_transaction", - "discount_amount", - "discount_amounts", - "id", - "invoice", - "livemode", - "memo", - "number", - "object", - "out_of_band_amount", - "pdf", - "reason", - "refund", - "status", - "subtotal", - "tax_amounts", - "total", - "type", - "voided_at", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:customer"], - min_size=1, - keys=[ - "address", - "balance", - "created", - "currency", - "default_source", - "delinquent", - "description", - "discount", - "email", - "id", - "invoice_prefix", - "invoice_settings", - "livemode", - "name", - "next_invoice_sequence", - "object", - "phone", - "preferred_locales", - "shipping", - "tax_exempt", - "test_clock", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:customer_balance_transaction"], - min_size=2, - keys=[ - "amount", - "created", - "credit_note", - "currency", - "customer", - "description", - "ending_balance", - "id", - "invoice", - "livemode", - "object", - "type", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:dispute"], - min_size=3, - keys=[ - "amount", - "balance_transactions", - "charge", - "created", - "currency", - "evidence", - "evidence_details", - "id", - "is_charge_refundable", - "livemode", - "object", - "payment_intent", - "reason", - "status", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:invoice"], - min_size=1, - keys=[ - "account_country", - "account_name", - "amount_due", - "amount_paid", - "amount_remaining", - "application_fee_amount", - "attempt_count", - "attempted", - "auto_advance", - "automatic_tax", - "billing_reason", - "charge", - "collection_method", - "created", - "currency", - "custom_fields", - "customer", - "customer_address", - "customer_email", - "customer_name", - "customer_phone", - "customer_shipping", - "customer_tax_exempt", - "customer_tax_ids", - "default_payment_method", - "default_source", - "default_tax_rates", - "description", - "discount", - "discounts", - "due_date", - "ending_balance", - "footer", - "hosted_invoice_url", - "id", - "invoice_pdf", - "last_finalization_error", - "livemode", - "next_payment_attempt", - "number", - "object", - "on_behalf_of", - "paid", - "paid_out_of_band", - "payment_intent", - "payment_settings", - "period_end", - "period_start", - "post_payment_credit_notes_amount", - "pre_payment_credit_notes_amount", - "quote", - "receipt_number", - "starting_balance", - "statement_descriptor", - "status", - "status_transitions", - "subscription", - "subtotal", - "tax", - "test_clock", - "total", - "total_tax_amounts", - "transfer_data", - "webhooks_delivered_at", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:invoice_item"], - min_size=1, - keys=[ - "amount", - "currency", - "customer", - "date", - "description", - "discountable", - "discounts", - "id", - "invoice", - "livemode", - "object", - "period", - "price", - "proration", - "quantity", - "subscription", - "tax_rates", - "test_clock", - "unit_amount", - "unit_amount_decimal", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:payment_intent"], - min_size=1, - keys=[ - "amount", - "amount_capturable", - "amount_received", - "application", - "application_fee_amount", - "automatic_payment_methods", - "canceled_at", - "cancellation_reason", - "capture_method", - "client_secret", - "confirmation_method", - "created", - "currency", - "customer", - "description", - "id", - "invoice", - "last_payment_error", - "livemode", - "next_action", - "object", - "on_behalf_of", - "payment_method", - "payment_method_options", - "payment_method_types", - "processing", - "receipt_email", - "review", - "setup_future_usage", - "shipping", - "statement_descriptor", - "statement_descriptor_suffix", - "status", - "transfer_data", - "transfer_group", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:payment_method"], - min_size=2, - keys=[ - "billing_details", - "created", - "customer", - "id", - "livemode", - "object", - "type", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:tax_id"], - min_size=1, - keys=[ - "country", - "created", - "customer", - "id", - "livemode", - "object", - "type", - "value", - "verification", - ], - ) - - # verify we only returned data for our identity email - assert v[f"{dataset_name}:customer"][0]["email"] == stripe_identity_email - customer_id: str = v[f"{dataset_name}:customer"][0]["id"] - - for bank_account in v[f"{dataset_name}:bank_account"]: - assert bank_account["customer"] == customer_id - - for card in v[f"{dataset_name}:card"]: - assert card["customer"] == customer_id - - charge_ids: List[str] = [] - for charge in v[f"{dataset_name}:charge"]: - assert charge["customer"] == customer_id - charge_ids.append(charge["id"]) - - payment_intent_ids: List[str] = [] - for payment_intent in v[f"{dataset_name}:payment_intent"]: - assert payment_intent["customer"] == customer_id - payment_intent_ids.append(payment_intent["id"]) - - for credit_note in v[f"{dataset_name}:credit_note"]: - assert credit_note["customer"] == customer_id - - for bank_account in v[f"{dataset_name}:bank_account"]: - assert bank_account["customer"] == customer_id - - for customer_balance_transaction in v[ - f"{dataset_name}:customer_balance_transaction" - ]: - assert customer_balance_transaction["customer"] == customer_id - - # disputes are retrieved by charge.id or payment_intent.id - for dispute in v[f"{dataset_name}:dispute"]: - assert ( - dispute["charge"] in charge_ids - or dispute["payment_intent"] in payment_intent_ids - ) - - for invoice in v[f"{dataset_name}:invoice"]: - assert invoice["customer"] == customer_id - - for invoice_item in v[f"{dataset_name}:invoice_item"]: - assert invoice_item["customer"] == customer_id - - for payment_method in v[f"{dataset_name}:payment_method"]: - assert payment_method["customer"] == customer_id - - for subscription in v[f"{dataset_name}:subscription"]: - assert subscription["customer"] == customer_id - - for tax_id in v[f"{dataset_name}:tax_id"]: - assert tax_id["customer"] == customer_id - - # verify we keep the expected fields after filtering by the user data category - target_categories = {"user"} - filtered_results = filter_data_categories(v, target_categories, graph) - - assert set(filtered_results.keys()) == { - f"{dataset_name}:bank_account", - f"{dataset_name}:card", - f"{dataset_name}:charge", - f"{dataset_name}:credit_note", - f"{dataset_name}:customer", - f"{dataset_name}:customer_balance_transaction", - f"{dataset_name}:dispute", - f"{dataset_name}:invoice", - f"{dataset_name}:invoice_item", - f"{dataset_name}:payment_intent", - f"{dataset_name}:payment_method", - f"{dataset_name}:tax_id", - } - - assert set(filtered_results[f"{dataset_name}:bank_account"][0].keys()) == { - "account_holder_name", - "bank_name", - "country", - "last4", - "routing_number", - } - - assert set(filtered_results[f"{dataset_name}:card"][0].keys()) == { - "address_city", - "address_country", - "address_line1", - "address_line2", - "address_state", - "address_zip", - "country", - "dynamic_last4", - "last4", - "name", - } - - assert set(filtered_results[f"{dataset_name}:charge"][0].keys()) == { - "amount", - "amount_captured", - "amount_refunded", - "billing_details", - "payment_method_details", - "receipt_email", - "source", - } - - assert set(filtered_results[f"{dataset_name}:credit_note"][0].keys()) == { - "amount", - "discount_amount", - "subtotal", - "total", - } - - assert set(filtered_results[f"{dataset_name}:customer"][0].keys()) == { - "address", - "balance", - "email", - "name", - "phone", - "preferred_locales", - "shipping", - } - - assert set( - filtered_results[f"{dataset_name}:customer_balance_transaction"][0].keys() - ) == {"ending_balance"} - - assert set(filtered_results[f"{dataset_name}:dispute"][0].keys()) == { - "amount", - "evidence", - } - - assert set(filtered_results[f"{dataset_name}:invoice"][0].keys()) == { - "account_country", - "account_name", - "amount_due", - "amount_paid", - "amount_remaining", - "customer_address", - "customer_email", - "customer_name", - "customer_phone", - "customer_shipping", - "discount", - "starting_balance", - "subtotal", - "total", - } - - assert set(filtered_results[f"{dataset_name}:invoice_item"][0].keys()) == { - "amount", - "unit_amount", - "unit_amount_decimal", - } - - assert set(filtered_results[f"{dataset_name}:payment_intent"][0].keys()) == { - "amount", - "amount_capturable", - "amount_received", - "receipt_email", - "shipping", - } - - assert set(filtered_results[f"{dataset_name}:payment_method"][0].keys()) == { - "billing_details", - "card", - } - - assert set(filtered_results[f"{dataset_name}:tax_id"][0].keys()) == { - "country", - "verification", - } - - -@pytest.mark.integration_saas -@pytest.mark.asyncio -@pytest.mark.parametrize( - "dsr_version", - ["use_dsr_3_0", "use_dsr_2_0"], -) -async def test_stripe_access_request_task_with_phone_number( - db, - policy, - dsr_version, - request, - privacy_request, - stripe_connection_config, - stripe_dataset_config, - stripe_identity_email, - stripe_identity_phone_number, -) -> None: - """Full access request based on the Stripe SaaS config""" - request.getfixturevalue(dsr_version) # REQUIRED to test both DSR 3.0 and 2.0 - # The Privacy request fixture we're using already has an email/phone cached - # so I'm clearing that first - clear_cache_identities(privacy_request.id) - - identity = Identity(**{"phone_number": stripe_identity_phone_number}) - privacy_request.cache_identity(identity) - - dataset_name = stripe_connection_config.get_saas_config().fides_key - merged_graph = stripe_dataset_config.get_graph() - graph = DatasetGraph(merged_graph) - - v = access_runner_tester( - privacy_request, - policy, - graph, - [stripe_connection_config], - {"phone_number": stripe_identity_phone_number}, - db, - ) - - assert_rows_match( - v[f"{dataset_name}:customer"], - min_size=1, - keys=[ - "address", - "balance", - "created", - "currency", - "default_source", - "delinquent", - "description", - "discount", - "email", - "id", - "invoice_prefix", - "invoice_settings", - "livemode", - "name", - "next_invoice_sequence", - "object", - "phone", - "preferred_locales", - "shipping", - "tax_exempt", - "test_clock", - ], - ) - - # verify we only returned data for our identity phone number and that - # it is the same customer that we retrieved using the identity email - assert v[f"{dataset_name}:customer"][0]["phone"] == stripe_identity_phone_number - assert v[f"{dataset_name}:customer"][0]["email"] == stripe_identity_email - - -@pytest.mark.integration_saas -@pytest.mark.asyncio -@pytest.mark.parametrize( - "dsr_version", - ["use_dsr_3_0", "use_dsr_2_0"], -) -async def test_stripe_erasure_request_task( - db, - privacy_request, - dsr_version, - request, - erasure_policy_string_rewrite, - stripe_connection_config, - stripe_dataset_config, - stripe_erasure_identity_email, - stripe_create_erasure_data, -) -> None: - """Full erasure request based on the Stripe SaaS config""" - request.getfixturevalue(dsr_version) # REQUIRED to test both DSR 3.0 and 2.0 - - privacy_request.policy_id = erasure_policy_string_rewrite.id - privacy_request.save(db) - - identity = Identity(**{"email": stripe_erasure_identity_email}) - privacy_request.cache_identity(identity) - - dataset_name = stripe_connection_config.get_saas_config().fides_key - merged_graph = stripe_dataset_config.get_graph() - graph = DatasetGraph(merged_graph) - - v = access_runner_tester( - privacy_request, - erasure_policy_string_rewrite, - graph, - [stripe_connection_config], - {"email": stripe_erasure_identity_email}, - db, - ) - - # verify staged data is available for erasure - assert_rows_match( - v[f"{dataset_name}:bank_account"], - min_size=1, - keys=[ - "account_holder_name", - "account_holder_type", - "account_type", - "bank_name", - "country", - "currency", - "customer", - "fingerprint", - "id", - "last4", - "object", - "routing_number", - "status", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:card"], - min_size=1, - keys=[ - "address_city", - "address_country", - "address_line1", - "address_line1_check", - "address_line2", - "address_state", - "address_zip", - "address_zip_check", - "brand", - "country", - "customer", - "cvc_check", - "dynamic_last4", - "exp_month", - "exp_year", - "fingerprint", - "funding", - "id", - "last4", - "name", - "object", - "tokenization_method", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:charge"], - min_size=2, - keys=[ - "amount", - "amount_captured", - "amount_refunded", - "application", - "application_fee", - "application_fee_amount", - "balance_transaction", - "billing_details", - "calculated_statement_descriptor", - "captured", - "created", - "currency", - "customer", - "description", - "disputed", - "failure_balance_transaction", - "failure_code", - "failure_message", - "fraud_details", - "id", - "invoice", - "livemode", - "object", - "on_behalf_of", - "order", - "paid", - "payment_intent", - "payment_method", - "payment_method_details", - "receipt_email", - "receipt_number", - "receipt_url", - "refunded", - "review", - "shipping", - "source_transfer", - "statement_descriptor", - "statement_descriptor_suffix", - "status", - "transfer_data", - "transfer_group", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:credit_note"], - min_size=1, - keys=[ - "amount", - "created", - "currency", - "customer", - "customer_balance_transaction", - "discount_amount", - "discount_amounts", - "id", - "invoice", - "livemode", - "memo", - "number", - "object", - "out_of_band_amount", - "pdf", - "reason", - "refund", - "status", - "subtotal", - "tax_amounts", - "total", - "type", - "voided_at", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:customer"], - min_size=1, - keys=[ - "address", - "balance", - "created", - "currency", - "default_source", - "delinquent", - "description", - "discount", - "email", - "id", - "invoice_prefix", - "invoice_settings", - "livemode", - "name", - "next_invoice_sequence", - "object", - "phone", - "preferred_locales", - "shipping", - "tax_exempt", - "test_clock", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:customer_balance_transaction"], - min_size=2, - keys=[ - "amount", - "created", - "credit_note", - "currency", - "customer", - "description", - "ending_balance", - "id", - "invoice", - "livemode", - "object", - "type", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:dispute"], - min_size=3, - keys=[ - "amount", - "balance_transactions", - "charge", - "created", - "currency", - "evidence", - "evidence_details", - "id", - "is_charge_refundable", - "livemode", - "object", - "payment_intent", - "reason", - "status", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:invoice"], - min_size=1, - keys=[ - "account_country", - "account_name", - "amount_due", - "amount_paid", - "amount_remaining", - "application_fee_amount", - "attempt_count", - "attempted", - "auto_advance", - "automatic_tax", - "billing_reason", - "charge", - "collection_method", - "created", - "currency", - "custom_fields", - "customer", - "customer_address", - "customer_email", - "customer_name", - "customer_phone", - "customer_shipping", - "customer_tax_exempt", - "customer_tax_ids", - "default_payment_method", - "default_source", - "default_tax_rates", - "description", - "discount", - "discounts", - "due_date", - "ending_balance", - "footer", - "hosted_invoice_url", - "id", - "invoice_pdf", - "last_finalization_error", - "livemode", - "next_payment_attempt", - "number", - "object", - "on_behalf_of", - "paid", - "paid_out_of_band", - "payment_intent", - "payment_settings", - "period_end", - "period_start", - "post_payment_credit_notes_amount", - "pre_payment_credit_notes_amount", - "quote", - "receipt_number", - "starting_balance", - "statement_descriptor", - "status", - "status_transitions", - "subscription", - "subtotal", - "tax", - "test_clock", - "total", - "total_tax_amounts", - "transfer_data", - "webhooks_delivered_at", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:invoice_item"], - min_size=1, - keys=[ - "amount", - "currency", - "customer", - "date", - "description", - "discountable", - "discounts", - "id", - "invoice", - "livemode", - "object", - "period", - "price", - "proration", - "quantity", - "subscription", - "tax_rates", - "test_clock", - "unit_amount", - "unit_amount_decimal", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:payment_intent"], - min_size=1, - keys=[ - "amount", - "amount_capturable", - "amount_received", - "application", - "application_fee_amount", - "automatic_payment_methods", - "canceled_at", - "cancellation_reason", - "capture_method", - "client_secret", - "confirmation_method", - "created", - "currency", - "customer", - "description", - "id", - "invoice", - "last_payment_error", - "livemode", - "next_action", - "object", - "on_behalf_of", - "payment_method", - "payment_method_options", - "payment_method_types", - "processing", - "receipt_email", - "review", - "setup_future_usage", - "shipping", - "statement_descriptor", - "statement_descriptor_suffix", - "status", - "transfer_data", - "transfer_group", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:payment_method"], - min_size=2, - keys=[ - "billing_details", - "created", - "customer", - "id", - "livemode", - "object", - "type", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:subscription"], - min_size=1, - keys=[ - "application_fee_percent", - "automatic_tax", - "billing_cycle_anchor", - "billing_thresholds", - "cancel_at", - "cancel_at_period_end", - "canceled_at", - "collection_method", - "created", - "current_period_end", - "current_period_start", - "customer", - "days_until_due", - "default_payment_method", - "default_source", - "default_tax_rates", - "discount", - "ended_at", - "id", - "latest_invoice", - "livemode", - "next_pending_invoice_item_invoice", - "object", - "pause_collection", - "payment_settings", - "pending_invoice_item_interval", - "pending_setup_intent", - "pending_update", - "schedule", - "start_date", - "status", - "test_clock", - "transfer_data", - "trial_end", - "trial_start", - ], - ) - - assert_rows_match( - v[f"{dataset_name}:tax_id"], - min_size=1, - keys=[ - "country", - "created", - "customer", - "id", - "livemode", - "object", - "type", - "value", - "verification", - ], - ) - - # Run erasure with masking_strict = False so both update and delete actions can be used - masking_strict = CONFIG.execution.masking_strict - CONFIG.execution.masking_strict = False - - x = erasure_runner_tester( - privacy_request, +class TestStripeConnector: + + def test_stripe_connection_test(self, stripe_runner: ConnectorRunner) -> None: + stripe_runner.test_connection() + + # @pytest.mark.parametrize( + # "dsr_version", + # ["use_dsr_3_0", "use_dsr_2_0"], + # ) + # async def test_stripe_access_request_task_with_email( + # self, + # stripe_runner: ConnectorRunner, + # policy: Policy, + # dsr_version, + # request, + # stripe_identity_email, + # ) -> None: + # """Full access request based on the Stripe SaaS config""" + # request.getfixturevalue(dsr_version) # REQUIRED to test both DSR 3.0 and 2.0 + + # dataset_name = stripe_runner.dataset_config.fides_key + + # access_results = await stripe_runner.access_request( + # access_policy=policy, identities={"email": stripe_identity_email} + # ) + + # # verify we only returned data for our identity email + + # assert len(access_results[f"{dataset_name}:customer"]) == 1 + + # assert access_results[f"{dataset_name}:customer"][0]["email"] == stripe_identity_email + # customer_id: str = access_results[f"{dataset_name}:customer"][0]["id"] + + # for bank_account in access_results[f"{dataset_name}:bank_account"]: + # assert bank_account["customer"] == customer_id + + # for card in access_results[f"{dataset_name}:card"]: + # assert card["customer"] == customer_id + + # charge_ids: List[str] = [] + # for charge in access_results[f"{dataset_name}:charge"]: + # assert charge["customer"] == customer_id + # charge_ids.append(charge["id"]) + + # payment_intent_ids: List[str] = [] + # for payment_intent in access_results[f"{dataset_name}:payment_intent"]: + # assert payment_intent["customer"] == customer_id + # payment_intent_ids.append(payment_intent["id"]) + + # for credit_note in access_results[f"{dataset_name}:credit_note"]: + # assert credit_note["customer"] == customer_id + + # for bank_account in access_results[f"{dataset_name}:bank_account"]: + # assert bank_account["customer"] == customer_id + + # for customer_balance_transaction in access_results[ + # f"{dataset_name}:customer_balance_transaction" + # ]: + # assert customer_balance_transaction["customer"] == customer_id + + # # disputes are retrieved by charge.id or payment_intent.id + # for dispute in access_results[f"{dataset_name}:dispute"]: + # assert ( + # dispute["charge"] in charge_ids + # or dispute["payment_intent"] in payment_intent_ids + # ) + + # for invoice in access_results[f"{dataset_name}:invoice"]: + # assert invoice["customer"] == customer_id + + # for invoice_item in access_results[f"{dataset_name}:invoice_item"]: + # assert invoice_item["customer"] == customer_id + + # for payment_method in access_results[f"{dataset_name}:payment_method"]: + # assert payment_method["customer"] == customer_id + + # for subscription in access_results[f"{dataset_name}:subscription"]: + # assert subscription["customer"] == customer_id + + # for tax_id in access_results[f"{dataset_name}:tax_id"]: + # assert tax_id["customer"] == customer_id + + + # @pytest.mark.parametrize( + # "dsr_version", + # ["use_dsr_3_0", "use_dsr_2_0"], + # ) + # async def test_stripe_access_request_task_with_phone_number( + # self, + # stripe_runner: ConnectorRunner, + # policy: Policy, + # dsr_version, + # request, + # stripe_identity_email, + # stripe_identity_phone_number, + # ) -> None: + # """Full access request based on the Stripe SaaS config""" + # request.getfixturevalue(dsr_version) # REQUIRED to test both DSR 3.0 and 2.0 + # # The Privacy request fixture we're using already has an email/phone cached + # # so I'm clearing that first + # dataset_name = stripe_runner.dataset_config.fides_key + + # access_results = await stripe_runner.access_request( + # access_policy=policy, identities={"phone_number": stripe_identity_phone_number} + # ) + + # # verify we only returned data for our identity phone number and that + # # it is the same customer that we retrieved using the identity email + # assert access_results[f"{dataset_name}:customer"][0]["phone"] == stripe_identity_phone_number + # assert access_results[f"{dataset_name}:customer"][0]["email"] == stripe_identity_email + + + @pytest.mark.integration_saas + @pytest.mark.asyncio + @pytest.mark.parametrize( + "dsr_version", + ["use_dsr_3_0", "use_dsr_2_0"], + ) + async def test_non_strict_erasure_request_with_email( + self, + stripe_runner: ConnectorRunner, + policy: Policy, + dsr_version, + request, erasure_policy_string_rewrite, - graph, - [stripe_connection_config], - {"email": stripe_erasure_identity_email}, - get_cached_data_for_erasures(privacy_request.id), - db, - ) - - # verify masking request was issued for endpoints with both update/delete actions - assert x == { - f"{dataset_name}:customer": 1, - f"{dataset_name}:tax_id": 1, - f"{dataset_name}:invoice_item": 1, - f"{dataset_name}:charge": 0, - f"{dataset_name}:invoice": 2, - f"{dataset_name}:card": 1, - f"{dataset_name}:customer_balance_transaction": 0, - f"{dataset_name}:payment_intent": 0, - f"{dataset_name}:payment_method": 3, - f"{dataset_name}:credit_note": 0, - f"{dataset_name}:bank_account": 1, - f"{dataset_name}:subscription": 1, - f"{dataset_name}:dispute": 0, - } - - stripe_secrets = stripe_connection_config.secrets - base_url = f"https://{stripe_secrets['domain']}" - headers = { - "Content-Type": "application/x-www-form-urlencoded", - "Authorization": f"Bearer {stripe_secrets['api_key']}", - } - - # customer - response = requests.get( - url=f"{base_url}/v1/customers", - headers=headers, - params={"email": stripe_erasure_identity_email}, - ) - customer = response.json()["data"][0] - customer_id = customer["id"] - assert customer["shipping"]["name"] == "MASKED" - - # card - response = requests.get( - url=f"{base_url}/v1/customers/{customer_id}/sources", - headers=headers, - params={"object": "card"}, - ) - cards = response.json()["data"] - assert cards == [] - - # payment method - response = requests.get( - url=f"{base_url}/v1/customers/{customer_id}/payment_methods", - headers=headers, - params={"type": "card"}, - ) - payment_methods = response.json()["data"] - for payment_method in payment_methods: - assert payment_method["billing_details"]["name"] == "MASKED" - - # bank account - response = requests.get( - url=f"{base_url}/v1/customers/{customer_id}/sources", - headers=headers, - params={"object": "bank_account"}, - ) - bank_account = response.json()["data"][0] - assert bank_account["account_holder_name"] == "MASKED" - - # tax_id - response = requests.get( - url=f"{base_url}/v1/customers/{customer_id}/tax_ids", headers=headers - ) - tax_ids = response.json()["data"] - assert tax_ids == [] - - # invoice_item - response = requests.get( - url=f"{base_url}/v1/invoiceitems", - headers=headers, - params={"customer": {customer_id}}, - ) - invoice_item = response.json()["data"] - # Can't delete an invoice item that is attached to an invoice that is no longer editable - assert len(invoice_item) == 1 + stripe_erasure_identity_email, + stripe_test_client, + stripe_create_erasure_data, + ) -> None: + """Full erasure request based on the Stripe SaaS config""" + request.getfixturevalue(dsr_version) # REQUIRED to test both DSR 3.0 and 2.0 + + dataset_name = stripe_runner.dataset_config.fides_key + + ( + _, + erasure_results, + ) = await stripe_runner.non_strict_erasure_request( + access_policy=policy, + erasure_policy=erasure_policy_string_rewrite, + identities={"email": stripe_erasure_identity_email}, + ) - # subscription - response = requests.get( - url=f"{base_url}/v1/customers/{customer_id}/subscriptions", headers=headers - ) - subscriptions = response.json()["data"] - assert subscriptions == [] - # reset - CONFIG.execution.masking_strict = masking_strict + # verify masking request was issued for endpoints with both update/delete actions + assert erasure_results == { + f"{dataset_name}:customer": 1, + f"{dataset_name}:tax_id": 1, + f"{dataset_name}:invoice_item": 1, + f"{dataset_name}:charge": 0, + f"{dataset_name}:invoice": 2, + f"{dataset_name}:card": 1, + f"{dataset_name}:customer_balance_transaction": 0, + f"{dataset_name}:payment_intent": 0, + f"{dataset_name}:payment_method": 3, + f"{dataset_name}:credit_note": 0, + f"{dataset_name}:bank_account": 1, + f"{dataset_name}:subscription": 1, + f"{dataset_name}:dispute": 0, + } + + # customer + customer = stripe_test_client.get_customer(stripe_erasure_identity_email) + customer_id = customer["id"] + assert customer["shipping"]["name"] == "MASKED" + + # card + cards = stripe_test_client.get_card(customer_id) + assert cards == [] + + # payment method + payment_methods = stripe_test_client.get_payment_method(customer_id) + for payment_method in payment_methods: + assert payment_method["billing_details"]["name"] == "MASKED" + + # bank account + bank_account = stripe_test_client.get_bank_account(customer_id) + assert bank_account["account_holder_name"] == "MASKED" + + # tax_id + tax_ids = stripe_test_client.get_tax_ids(customer_id) + assert tax_ids == [] + + # invoice_item + invoice_item = stripe_test_client.get_invoice_items(customer_id) + # Can't delete an invoice item that is attached to an invoice that is no longer editable + assert len(invoice_item) == 1 + + # subscription + subscriptions = stripe_test_client.get_subscription(customer_id) + assert subscriptions == []