Create Stripe customer factories, use in tests

tests/data/__init__.py

@@ -0,0 +1,8 @@
+from base64 import b64encode
+from typing import Optional
+
+
+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 ''}"

tests/factories/__init__.py

@@ -0,0 +1 @@
+from . import models, stripe

tests/factories/models.py

@@ -0,0 +1,111 @@
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import factory
+from factory.alchemy import SQLAlchemyModelFactory
+
+from ctms import models
+from ctms.database import ScopedSessionLocal
+from tests.data import fake_stripe_id
+
+
+class BaseSQLAlchemyModelFactory(SQLAlchemyModelFactory):
+    class Meta:
+        abstract = True
+        sqlalchemy_session = ScopedSessionLocal
+
+
+class NewsletterFactory(BaseSQLAlchemyModelFactory):
+    class Meta:
+        model = models.Newsletter
+
+    name = factory.Sequence(lambda n: f"newsletter-{n}")
+    subscribed = True
+    format = "T"
+    lang = factory.Faker("language_code")
+    source = factory.Faker("url")
+
+    email = factory.SubFactory(factory="tests.factories.models.EmailFactory")
+
+
+class WaitlistFactory(BaseSQLAlchemyModelFactory):
+    class Meta:
+        model = models.Waitlist
+
+    name = factory.Sequence(lambda n: f"waitlist-{n}")
+    source = factory.Faker("url")
+    fields = {}
+
+    email = factory.SubFactory(factory="tests.factories.models.EmailFactory")
+
+
+class FirefoxAccountFactory(BaseSQLAlchemyModelFactory):
+    class Meta:
+        model = models.FirefoxAccount
+
+    fxa_id = factory.LazyFunction(lambda: uuid4().hex)
+    primary_email = factory.SelfAttribute("email.primary_email")
+    created_date = factory.Faker("date")
+    lang = factory.Faker("language_code")
+    first_service = None
+    account_deleted = False
+
+    email = factory.SubFactory(factory="tests.factories.models.EmailFactory")
+
+
+class EmailFactory(BaseSQLAlchemyModelFactory):
+    class Meta:
+        model = models.Email
+
+    email_id = factory.Faker("uuid4")
+    primary_email = factory.Faker("email")
+    basket_token = factory.Faker("uuid4")
+    first_name = factory.Faker("first_name")
+    last_name = factory.Faker("last_name")
+    mailing_country = factory.Faker("country_code")
+    email_format = "T"
+    email_lang = factory.Faker("language_code")
+    double_opt_in = False
+    has_opted_out_of_email = False
+
+    @factory.post_generation
+    def newsletters(self, create, extracted, **kwargs):
+        if not create:
+            return
+        if extracted:
+            for _ in range(extracted):
+                NewsletterFactory(email=self, **kwargs)
+
+    @factory.post_generation
+    def fxa(self, create, extracted, **kwargs):
+        if not create:
+            return
+        if extracted:
+            FirefoxAccountFactory(email=self, **kwargs)
+
+
+class StripeCustomerFactory(BaseSQLAlchemyModelFactory):
+    class Meta:
+        model = models.StripeCustomer
+
+    stripe_id = factory.LazyFunction(lambda: fake_stripe_id("cus", "customer"))
+    fxa_id = factory.SelfAttribute("fxa.fxa_id")
+    default_source_id = factory.LazyFunction(
+        lambda: fake_stripe_id("card", "default_payment")
+    )
+    invoice_settings_default_payment_method_id = factory.LazyFunction(
+        lambda: fake_stripe_id("pm", "default_payment")
+    )
+    stripe_created = factory.LazyFunction(lambda: datetime.now(timezone.utc))
+    deleted = False
+
+    fxa = factory.SubFactory(factory=FirefoxAccountFactory)
+
+
+__all__ = (
+    "EmailFactory",
+    "FirefoxAccountFactory",
+    "NewsletterFactory",
+    "StripeCustomerFactory",
+    "WaitlistFactory",
+)

tests/factories/stripe.py

@@ -0,0 +1,39 @@
+from uuid import uuid4
+
+import factory
+
+from tests.data import fake_stripe_id
+
+
+class StripeCustomerDataFactory(factory.DictFactory):
+    class Params:
+        fxa_id = factory.LazyFunction(lambda: uuid4().hex)
+
+    id = factory.LazyFunction(lambda: fake_stripe_id("cus", "customer"))
+    object = "customer"
+    address = None
+    balance = 0
+    created = factory.Faker("unix_time")
+    currency = "usd"
+    default_source = None
+    delinquent = False
+    description = factory.LazyAttribute(lambda o: o.fxa_id)
+    discount = None
+    email = factory.Faker("email")
+    invoice_prefix = factory.Faker("bothify", text="###???###")
+    invoice_settings = {
+        "custom_fields": None,
+        "default_payment_method": fake_stripe_id("pm", "payment_method"),
+        "footer": None,
+    }
+    livemode = False
+    metadata = factory.Dict({"userid": factory.SelfAttribute("..description")})
+    name = factory.Faker("name")
+    next_invoice_sequence = factory.Faker("pyint")
+    phone = None
+    preferred_locales = []
+    shipping = None
+    tax_exempt = "none"
+
+
+__all__ = ("StripeCustomerDataFactory",)

tests/unit/conftest.py

@@ -1,7 +1,6 @@
 """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
@@ -26,32 +25,29 @@
 from ctms.crud import (
     create_api_client,
     create_contact,
-    create_stripe_customer,
     create_stripe_price,
     create_stripe_subscription,
     create_stripe_subscription_item,
     get_all_acoustic_fields,
     get_all_acoustic_newsletters_mapping,
     get_amo_by_email_id,
+    get_contact_by_email_id,
     get_contacts_by_any_id,
-    get_email,
     get_fxa_by_email_id,
     get_mofo_by_email_id,
     get_newsletters_by_email_id,
-    get_stripe_products,
     get_waitlists_by_email_id,
 )
 from ctms.database import ScopedSessionLocal, SessionLocal
 from ctms.schemas import (
     ApiClientSchema,
     ContactSchema,
-    StripeCustomerCreateSchema,
     StripePriceCreateSchema,
     StripeSubscriptionCreateSchema,
     StripeSubscriptionItemCreateSchema,
 )
-
-from . import factories
+from tests import factories
+from tests.data import fake_stripe_id
 
 MY_FOLDER = os.path.dirname(__file__)
 TEST_FOLDER = os.path.dirname(MY_FOLDER)
@@ -69,13 +65,6 @@
     ("default_newsletter_contact_data", {"newsletters"}),
 ]
 
-
-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")
@@ -182,9 +171,13 @@ def end_savepoint(*args):
     transaction.rollback()
 
 
-register(factories.EmailFactory)
-register(factories.NewsletterFactory)
-register(factories.WaitlistFactory)
+# Database models
+register(factories.models.EmailFactory)
+register(factories.models.NewsletterFactory)
+register(factories.models.StripeCustomerFactory)
+register(factories.models.WaitlistFactory)
+# Stripe REST API payloads
+register(factories.stripe.StripeCustomerDataFactory)
 
 
 @pytest.fixture
@@ -692,20 +685,6 @@ def stripe_test_json(request):
     return data
 
 
-@pytest.fixture
-def stripe_customer_data():
-    return {
-        "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(
-            "pm", "payment_method"
-        ),
-    }
-
-
 @pytest.fixture
 def stripe_price_data():
     return {
@@ -776,24 +755,6 @@ def stripe_invoice_line_item_data():
     }
 
 
-@pytest.fixture
-def raw_stripe_customer_data(stripe_customer_data) -> dict:
-    """Return minimal Stripe customer data."""
-    return {
-        "id": stripe_customer_data["stripe_id"],
-        "object": "customer",
-        "created": unix_timestamp(stripe_customer_data["stripe_created"]),
-        "description": stripe_customer_data["fxa_id"],
-        "email": "fxa_email@example.com",
-        "default_source": stripe_customer_data["default_source_id"],
-        "invoice_settings": {
-            "default_payment_method": stripe_customer_data[
-                "invoice_settings_default_payment_method_id"
-            ],
-        },
-    }
-
-
 @pytest.fixture
 def raw_stripe_price_data(stripe_price_data):
     """Return minimal Stripe price data."""
@@ -897,24 +858,18 @@ def raw_stripe_invoice_data(
     }
 
 
-@pytest.fixture
-def contact_with_stripe_customer(dbsession, example_contact, stripe_customer_data):
-    """Return the example contact with an associated Stripe Customer account."""
-    create_stripe_customer(
-        dbsession, StripeCustomerCreateSchema(**stripe_customer_data)
-    )
-    dbsession.commit()
-    return example_contact
-
-
 @pytest.fixture
 def contact_with_stripe_subscription(
     dbsession,
-    contact_with_stripe_customer,
+    example_contact,
+    stripe_customer_factory,
     stripe_price_data,
     stripe_subscription_data,
     stripe_subscription_item_data,
 ):
+    stripe_customer = stripe_customer_factory(
+        stripe_id=FAKE_STRIPE_CUSTOMER_ID, fxa_id=example_contact.fxa.fxa_id
+    )
     create_stripe_price(dbsession, StripePriceCreateSchema(**stripe_price_data))
     create_stripe_subscription(
         dbsession, StripeSubscriptionCreateSchema(**stripe_subscription_data)
@@ -926,9 +881,9 @@ def contact_with_stripe_subscription(
         ),
     )
     dbsession.commit()
-    email = get_email(dbsession, contact_with_stripe_customer.email.email_id)
-    contact_with_stripe_customer.products = get_stripe_products(email)
-    return contact_with_stripe_customer
+
+    contact = get_contact_by_email_id(dbsession, stripe_customer.email.email_id)
+    return ContactSchema(**contact)
 
 
 @pytest.fixture