Finish moving Stripe data out of sample_data.py

tests/unit/conftest.py

@@ -1,6 +1,7 @@
 """pytest fixtures for the CTMS app"""
 import json
 import os.path
+from base64 import b64encode
 from datetime import datetime, timezone
 from glob import glob
 from time import mktime
@@ -49,7 +50,6 @@
     StripeSubscriptionCreateSchema,
     StripeSubscriptionItemCreateSchema,
 )
-from tests.unit.sample_data import FAKE_STRIPE_ID
 
 from . import factories
 
@@ -70,6 +70,18 @@
 ]
 
 
+def fake_stripe_id(prefix: str, seed: str, suffix: Optional[str] = None) -> str:
+    """Create a fake Stripe ID for testing"""
+    body = b64encode(seed.encode()).decode().replace("=", "")
+    return f"{prefix}_{body}{suffix if suffix else ''}"
+
+
+FAKE_STRIPE_CUSTOMER_ID = fake_stripe_id("cus", "customer")
+FAKE_STRIPE_INVOICE_ID = fake_stripe_id("in", "invoice")
+FAKE_STRIPE_PRICE_ID = fake_stripe_id("price", "price")
+FAKE_STRIPE_SUBSCRIPTION_ID = fake_stripe_id("sub", "subscription")
+
+
 def _gather_examples(schema_class) -> dict[str, str]:
     """Gather the examples from a schema definition"""
     examples = {}
@@ -683,21 +695,23 @@ def stripe_test_json(request):
 @pytest.fixture
 def stripe_customer_data():
     return {
-        "stripe_id": FAKE_STRIPE_ID["Customer"],
+        "stripe_id": FAKE_STRIPE_CUSTOMER_ID,
         "stripe_created": datetime(2021, 10, 25, 15, 34, tzinfo=timezone.utc),
         # TODO magic string from fxa schema
         "fxa_id": "6eb6ed6ac3b64259968aa490c6c0b9df",
         "default_source_id": None,
-        "invoice_settings_default_payment_method_id": FAKE_STRIPE_ID["Payment Method"],
+        "invoice_settings_default_payment_method_id": fake_stripe_id(
+            "pm", "payment_method"
+        ),
     }
 
 
 @pytest.fixture
 def stripe_price_data():
     return {
-        "stripe_id": FAKE_STRIPE_ID["Price"],
+        "stripe_id": FAKE_STRIPE_PRICE_ID,
         "stripe_created": datetime(2020, 10, 27, 10, 45, tzinfo=timezone.utc),
-        "stripe_product_id": FAKE_STRIPE_ID["Product"],
+        "stripe_product_id": fake_stripe_id("prod", "product"),
         "active": True,
         "currency": "usd",
         "recurring_interval": "month",
@@ -709,9 +723,9 @@ def stripe_price_data():
 @pytest.fixture
 def stripe_subscription_data():
     return {
-        "stripe_id": FAKE_STRIPE_ID["Subscription"],
+        "stripe_id": FAKE_STRIPE_SUBSCRIPTION_ID,
         "stripe_created": datetime(2021, 9, 27, tzinfo=timezone.utc),
-        "stripe_customer_id": FAKE_STRIPE_ID["Customer"],
+        "stripe_customer_id": FAKE_STRIPE_CUSTOMER_ID,
         "default_source_id": None,
         "default_payment_method_id": None,
         "cancel_at_period_end": False,
@@ -727,19 +741,19 @@ def stripe_subscription_data():
 @pytest.fixture
 def stripe_subscription_item_data():
     return {
-        "stripe_id": FAKE_STRIPE_ID["Subscription Item"],
+        "stripe_id": FAKE_STRIPE_SUBSCRIPTION_ID,
         "stripe_created": datetime(2021, 9, 27, tzinfo=timezone.utc),
-        "stripe_subscription_id": FAKE_STRIPE_ID["Subscription"],
-        "stripe_price_id": FAKE_STRIPE_ID["Price"],
+        "stripe_subscription_id": FAKE_STRIPE_SUBSCRIPTION_ID,
+        "stripe_price_id": FAKE_STRIPE_PRICE_ID,
     }
 
 
 @pytest.fixture
 def stripe_invoice_data():
     return {
-        "stripe_id": FAKE_STRIPE_ID["Invoice"],
+        "stripe_id": FAKE_STRIPE_INVOICE_ID,
         "stripe_created": datetime(2021, 10, 28, tzinfo=timezone.utc),
-        "stripe_customer_id": FAKE_STRIPE_ID["Customer"],
+        "stripe_customer_id": FAKE_STRIPE_CUSTOMER_ID,
         "default_source_id": None,
         "default_payment_method_id": None,
         "currency": "usd",
@@ -751,11 +765,11 @@ def stripe_invoice_data():
 @pytest.fixture
 def stripe_invoice_line_item_data():
     return {
-        "stripe_id": FAKE_STRIPE_ID["(Invoice) Line Item"],
-        "stripe_price_id": FAKE_STRIPE_ID["Price"],
-        "stripe_invoice_id": FAKE_STRIPE_ID["Invoice"],
-        "stripe_subscription_id": FAKE_STRIPE_ID["Subscription"],
-        "stripe_subscription_item_id": FAKE_STRIPE_ID["Subscription Item"],
+        "stripe_id": fake_stripe_id("il", "invoice line item"),
+        "stripe_price_id": FAKE_STRIPE_PRICE_ID,
+        "stripe_invoice_id": FAKE_STRIPE_INVOICE_ID,
+        "stripe_subscription_id": FAKE_STRIPE_SUBSCRIPTION_ID,
+        "stripe_subscription_item_id": fake_stripe_id("si", "subscription_item"),
         "stripe_type": "subscription",
         "amount": 1000,
         "currency": "usd",

tests/unit/test_api_stripe.py

@@ -11,7 +11,7 @@
 
 from ctms.app import app, get_pubsub_claim
 from ctms.models import PendingAcousticRecord
-from tests.unit.sample_data import fake_stripe_id
+from tests.unit.conftest import fake_stripe_id
 
 
 def pubsub_wrap(data):

tests/unit/test_crud.py

@@ -59,7 +59,7 @@
     StripeSubscriptionCreateSchema,
     StripeSubscriptionItemCreateSchema,
 )
-from tests.unit.sample_data import FAKE_STRIPE_ID, fake_stripe_id
+from tests.unit.conftest import fake_stripe_id
 
 # Treat all SQLAlchemy warnings as errors
 pytestmark = pytest.mark.filterwarnings("error::sqlalchemy.exc.SAWarning")
@@ -79,21 +79,26 @@ def test_get_email(dbsession, example_contact):
     assert sorted(waitlists_names) == waitlists_names
 
 
-def test_get_email_with_stripe_customer(dbsession, contact_with_stripe_customer):
+def test_get_email_with_stripe_customer(
+    dbsession, stripe_customer_data, contact_with_stripe_customer
+):
+    """`stripe_customer_data` is used when building `contact_with_stripe_customer`"""
     email_id = contact_with_stripe_customer.email.email_id
     email = get_email(dbsession, email_id)
     assert email.email_id == email_id
     newsletter_names = [newsletter.name for newsletter in email.newsletters]
     assert newsletter_names == ["firefox-welcome", "mozilla-welcome"]
     assert sorted(newsletter_names) == newsletter_names
 
-    assert email.stripe_customer.stripe_id == FAKE_STRIPE_ID["Customer"]
+    assert email.stripe_customer.stripe_id == stripe_customer_data["stripe_id"]
     assert len(email.stripe_customer.subscriptions) == 0
 
 
 def test_get_email_with_stripe_subscription(
-    dbsession, contact_with_stripe_subscription
+    dbsession, contact_with_stripe_subscription, stripe_customer_data, stripe_price_data
 ):
+    """`stripe_customer_data` and `stripe_price_data` is used when building
+    `contact_with_stripe_subscription`"""
     email_id = contact_with_stripe_subscription.email.email_id
     email = get_email(dbsession, email_id)
     assert email.email_id == email_id
@@ -102,12 +107,12 @@ def test_get_email_with_stripe_subscription(
     assert newsletter_names == ["firefox-welcome", "mozilla-welcome"]
     assert sorted(newsletter_names) == newsletter_names
 
-    assert email.stripe_customer.stripe_id == FAKE_STRIPE_ID["Customer"]
+    assert email.stripe_customer.stripe_id == stripe_customer_data["stripe_id"]
     assert len(email.stripe_customer.subscriptions) == 1
     assert len(email.stripe_customer.subscriptions[0].subscription_items) == 1
     assert (
         email.stripe_customer.subscriptions[0].subscription_items[0].price.stripe_id
-        == FAKE_STRIPE_ID["Price"]
+        == stripe_price_data["stripe_id"]
     )
 
 
@@ -504,11 +509,11 @@ def test_get_acoustic_record_serial_stripe_subscriptions(
 
 
 def test_get_acoustic_record_stripe_subscription_cancelled(
-    dbsession, contact_with_stripe_subscription
+    dbsession, contact_with_stripe_subscription, stripe_subscription_data
 ):
     """A contact with a canceled Stripe subscription is in the canceled segement."""
     subscription = get_stripe_subscription_by_stripe_id(
-        dbsession, FAKE_STRIPE_ID["Subscription"]
+        dbsession, stripe_subscription_data["stripe_id"]
     )
     subscription.status = "canceled"
     subscription.ended_at = subscription.current_period_end
@@ -523,11 +528,11 @@ def test_get_acoustic_record_stripe_subscription_cancelled(
 
 
 def test_get_acoustic_record_stripe_subscription_other(
-    dbsession, contact_with_stripe_subscription
+    dbsession, contact_with_stripe_subscription, stripe_subscription_data
 ):
     """A contact with a canceled Stripe subscription is in the canceled segement."""
     subscription = get_stripe_subscription_by_stripe_id(
-        dbsession, FAKE_STRIPE_ID["Subscription"]
+        dbsession, stripe_subscription_data["stripe_id"]
     )
     subscription.status = "unpaid"
     dbsession.commit()