Start to use factories to generate test data

ctms/database.py

@@ -1,5 +1,5 @@
 from sqlalchemy import create_engine, engine
-from sqlalchemy.orm import declarative_base, sessionmaker
+from sqlalchemy.orm import declarative_base, scoped_session, sessionmaker
 
 from . import config
 
@@ -17,5 +17,7 @@ def engine_factory(settings):
 
 engine = engine_factory(config.Settings())
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+# Used for testing
+ScopedSessionLocal = scoped_session(SessionLocal)
 
 Base = declarative_base()

pyproject.toml

@@ -49,6 +49,8 @@ SQLAlchemy-Utils = "^0.40.0"
 pylint = "^2.16.0"
 pylint-pytest = "^1.1.2"
 types-requests = "^2.28.11"
+factory-boy = "^3.2.1"
+pytest-factoryboy = "^2.5.1"
 
 [tool.pytest.ini_options]
 testpaths = [

tests/unit/conftest.py

@@ -12,8 +12,8 @@
 from fastapi.testclient import TestClient
 from prometheus_client import CollectorRegistry
 from pydantic import PostgresDsn
+from pytest_factoryboy import register
 from sqlalchemy import create_engine, event
-from sqlalchemy.orm import Session
 from sqlalchemy_utils.functions import create_database, database_exists, drop_database
 
 from ctms.app import app, get_api_client, get_db, get_metrics
@@ -37,6 +37,7 @@
     get_stripe_products,
     get_waitlists_by_email_id,
 )
+from ctms.database import ScopedSessionLocal, SessionLocal
 from ctms.schemas import (
     ApiClientSchema,
     ContactSchema,
@@ -51,6 +52,8 @@
     SAMPLE_STRIPE_DATA,
 )
 
+from . import factories
+
 MY_FOLDER = os.path.dirname(__file__)
 TEST_FOLDER = os.path.dirname(MY_FOLDER)
 APP_FOLDER = os.path.dirname(TEST_FOLDER)
@@ -104,10 +107,11 @@ def engine(pytestconfig):
         drop_database(test_db_url)
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 def connection(engine):
     """Return a connection to the database that rolls back automatically."""
     conn = engine.connect()
+    SessionLocal.configure(bind=conn)
     yield conn
     conn.close()
 
@@ -119,18 +123,23 @@ def dbsession(connection):
     Adapted from https://docs.sqlalchemy.org/en/14/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites
     """
     transaction = connection.begin()
-    session = Session(autocommit=False, autoflush=False, bind=connection)
+    session = ScopedSessionLocal()
     nested = connection.begin_nested()
 
     # If the application code calls session.commit, it will end the nested
     # transaction. Need to start a new one when that happens.
     @event.listens_for(session, "after_transaction_end")
-    def end_savepoint(session, transaction):
+    def end_savepoint(*args):
         nonlocal nested
         if not nested.is_active:
             nested = connection.begin_nested()
 
     yield session
+    # a nescessary addition to the example in the documentation linked above.
+    # Without this, the listener is not removed after each test ends and
+    # SQLAlchemy emits warnings:
+    #     `SAWarning: nested transaction already deassociated from connection`
+    event.remove(session, "after_transaction_end", end_savepoint)
     session.close()
     transaction.rollback()
 
@@ -162,6 +171,10 @@ def minimal_contact(dbsession):
     return contact
 
 
+register(factories.EmailFactory)
+register(factories.NewsletterFactory)
+
+
 @pytest.fixture
 def maximal_contact(dbsession):
     email_id = UUID("67e52c77-950f-4f28-accb-bb3ea1a2c51a")

tests/unit/factories.py

@@ -0,0 +1,54 @@
+import factory
+from factory.alchemy import SQLAlchemyModelFactory
+
+from ctms import models
+from ctms.database import ScopedSessionLocal
+
+
+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.unit.factories.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)
+
+
+__all__ = (
+    "NewsletterFactory",
+    "EmailFactory",
+)

tests/unit/test_api_get.py

@@ -6,9 +6,12 @@
 from ctms.models import Email
 
 
-def test_get_ctms_for_minimal_contact(client, minimal_contact):
+def test_get_ctms_for_minimal_contact(client, dbsession, email_factory):
     """GET /ctms/{email_id} returns a contact with most fields unset."""
-    email_id = minimal_contact.email.email_id
+    contact = email_factory(newsletters=1)
+    newsletter = contact.newsletters[0]
+    dbsession.commit()
+    email_id = str(contact.email_id)
     resp = client.get(f"/ctms/{email_id}")
     assert resp.status_code == 200
     assert resp.json() == {
@@ -27,20 +30,20 @@ def test_get_ctms_for_minimal_contact(client, minimal_contact):
             "username": None,
         },
         "email": {
-            "basket_token": "142e20b6-1ef5-43d8-b5f4-597430e956d7",
-            "create_timestamp": "2014-01-22T15:24:00+00:00",
-            "double_opt_in": False,
-            "email_format": "H",
-            "email_id": "93db83d4-4119-4e0c-af87-a713786fa81d",
-            "email_lang": "en",
-            "first_name": None,
-            "has_opted_out_of_email": False,
-            "last_name": None,
-            "mailing_country": "us",
-            "primary_email": "ctms-user@example.com",
-            "sfdc_id": "001A000001aABcDEFG",
-            "unsubscribe_reason": None,
-            "update_timestamp": "2020-01-22T15:24:00+00:00",
+            "basket_token": str(contact.basket_token),
+            "create_timestamp": contact.create_timestamp.isoformat(),
+            "double_opt_in": contact.double_opt_in,
+            "email_format": contact.email_format,
+            "email_id": str(contact.email_id),
+            "email_lang": contact.email_lang,
+            "first_name": contact.first_name,
+            "has_opted_out_of_email": contact.has_opted_out_of_email,
+            "last_name": contact.last_name,
+            "mailing_country": contact.mailing_country,
+            "primary_email": contact.primary_email,
+            "sfdc_id": None,
+            "unsubscribe_reason": contact.unsubscribe_reason,
+            "update_timestamp": contact.update_timestamp.isoformat(),
         },
         "fxa": {
             "created_date": None,
@@ -57,37 +60,13 @@ def test_get_ctms_for_minimal_contact(client, minimal_contact):
         },
         "newsletters": [
             {
-                "format": "H",
-                "lang": "en",
-                "name": "app-dev",
-                "source": None,
-                "subscribed": True,
-                "unsub_reason": None,
-            },
-            {
-                "format": "H",
-                "lang": "en",
-                "name": "maker-party",
-                "source": None,
-                "subscribed": True,
-                "unsub_reason": None,
-            },
-            {
-                "format": "H",
-                "lang": "en",
-                "name": "mozilla-foundation",
-                "source": None,
-                "subscribed": True,
-                "unsub_reason": None,
-            },
-            {
-                "format": "H",
-                "lang": "en",
-                "name": "mozilla-learning-network",
-                "source": None,
-                "subscribed": True,
-                "unsub_reason": None,
-            },
+                "format": newsletter.format,
+                "lang": newsletter.lang,
+                "name": newsletter.name,
+                "source": newsletter.source,
+                "subscribed": newsletter.subscribed,
+                "unsub_reason": newsletter.unsub_reason,
+            }
         ],
         "status": "ok",
         "vpn_waitlist": {"geo": None, "platform": None},