Delay env var loading till after setup

Author: chriscarrollsmith

.github/workflows/test.yml

@@ -47,26 +47,18 @@ jobs:
 
       - name: Set env variables for pytest
         run: |
-          echo "DB_USER=postgres" >> $GITHUB_ENV
-          echo "DB_PASSWORD=postgres" >> $GITHUB_ENV
           echo "DB_HOST=127.0.0.1" >> $GITHUB_ENV
           echo "DB_PORT=5432" >> $GITHUB_ENV
-          echo "DB_NAME=test_db" >> $GITHUB_ENV
-          echo "SECRET_KEY=$(openssl rand -base64 32)" >> $GITHUB_ENV
-          echo "BASE_URL=http://localhost:8000" >> $GITHUB_ENV
-          echo "RESEND_API_KEY=resend_api_key" >> $GITHUB_ENV
-          echo "[email protected]" >> $GITHUB_ENV
+          echo "DB_USER=postgres" >> $GITHUB_ENV
+          echo "DB_PASSWORD=postgres" >> $GITHUB_ENV
         
       - name: Verify environment variables
         run: |
           echo "Checking if required environment variables are set..."
-          [ -n "$DB_USER" ] && \
-          [ -n "$DB_PASSWORD" ] && \
           [ -n "$DB_HOST" ] && \
           [ -n "$DB_PORT" ] && \
-          [ -n "$DB_NAME" ] && \
-          [ -n "$SECRET_KEY" ] && \
-          [ -n "$RESEND_API_KEY" ]
+          [ -n "$DB_USER" ] && \
+          [ -n "$DB_PASSWORD" ]
 
       - name: Run type checking with mypy
         run: uv run mypy .

tests/conftest.py

@@ -1,18 +1,16 @@
 import pytest
+import os
 from typing import Generator
 from sqlmodel import create_engine, Session, select
 from sqlalchemy import Engine
 from fastapi.testclient import TestClient
 from dotenv import load_dotenv
-from utils.core.db import get_connection_url, tear_down_db, set_up_db, create_default_roles
+from utils.core.db import get_connection_url, tear_down_db, set_up_db, create_default_roles, ensure_database_exists
 from utils.core.models import User, PasswordResetToken, EmailUpdateToken, Organization, Role, Account, Invitation
 from utils.core.auth import get_password_hash, create_access_token, create_refresh_token
 from main import app
 from datetime import datetime, UTC, timedelta
 
-# Load environment variables
-load_dotenv(override=True)
-
 # Define a custom exception for test setup errors
 class SetupError(Exception):
     """Exception raised for errors in the test setup process."""
@@ -21,29 +19,44 @@ def __init__(self, message="An error occurred during test setup"):
         super().__init__(self.message)
 
 
-@pytest.fixture(scope="session")
-def engine() -> Engine:
+@pytest.fixture
+def env_vars(monkeypatch):
+    # Get valid db user, password, host, and port from .env
+    load_dotenv()
+    os.environ["DB_HOST"] = os.getenv("DB_HOST", "localhost")
+    os.environ["DB_PORT"] = os.getenv("DB_PORT", "5432")
+    os.environ["DB_USER"] = os.getenv("DB_USER", "appuser")
+    os.environ["DB_PASSWORD"] = os.getenv("DB_PASSWORD", "testpassword")
+
+    # monkeypatch remaining env vars
+    with monkeypatch.context() as m:
+        m.setenv("SECRET_KEY", "testsecretkey")
+        m.setenv("HOST_NAME", "Test Organization")
+        m.setenv("DB_HOST", os.getenv("DB_HOST", "localhost"))
+        m.setenv("DB_PORT", os.getenv("DB_PORT", "5432"))
+        m.setenv("DB_USER", os.getenv("DB_USER", "appuser"))
+        m.setenv("DB_PASSWORD", os.getenv("DB_PASSWORD", "testpassword"))
+        m.setenv("DB_NAME", "qual2db4-test-db")
+        m.setenv("RESEND_API_KEY", "test")
+        m.setenv("EMAIL_FROM", "[email protected]")
+        m.setenv("QUALTRICS_BASE_URL", "test")
+        m.setenv("QUALTRICS_API_TOKEN", "test")
+        yield
+
+
+@pytest.fixture
+def engine(env_vars) -> Engine:
     """
     Create a new SQLModel engine for the test database.
     Use PostgreSQL for testing to match production environment.
     """
     # Use PostgreSQL for testing to match production environment
+    ensure_database_exists(get_connection_url())
     engine = create_engine(get_connection_url())
-    return engine
+    set_up_db(drop=True)
 
+    yield engine
 
-@pytest.fixture(scope="session", autouse=True)
-def set_up_database(engine) -> Generator[None, None, None]:
-    """
-    Set up the test database before running the test suite.
-    Drop all tables and recreate them to ensure a clean state.
-    """
-    # Drop and recreate all tables using the helpers from db.py
-    tear_down_db()
-    set_up_db(drop=False)
-    
-    yield
-    
     # Clean up after tests
     tear_down_db()
 
@@ -57,20 +70,7 @@ def session(engine) -> Generator[Session, None, None]:
         yield session
 
 
-@pytest.fixture(autouse=True)
-def clean_db(session: Session) -> None:
-    """
-    Cleans up the database tables before each test.
-    """
-    # Don't delete permissions as they are required for tests
-    for model in (PasswordResetToken, EmailUpdateToken, User, Role, Organization, Account):
-        for record in session.exec(select(model)).all():
-            session.delete(record)
-
-    session.commit()
-
-
-@pytest.fixture()
+@pytest.fixture
 def test_account(session: Session) -> Account:
     """
     Creates a test account in the database.
@@ -85,7 +85,7 @@ def test_account(session: Session) -> Account:
     return account
 
 
-@pytest.fixture()
+@pytest.fixture
 def test_user(session: Session, test_account: Account) -> User:
     """
     Creates a test user in the database linked to the test account.
@@ -103,7 +103,7 @@ def test_user(session: Session, test_account: Account) -> User:
     return user
 
 
-@pytest.fixture()
+@pytest.fixture
 def unauth_client(session: Session) -> Generator[TestClient, None, None]:
     """
     Provides a TestClient instance without authentication.
@@ -112,7 +112,7 @@ def unauth_client(session: Session) -> Generator[TestClient, None, None]:
     yield client
 
 
-@pytest.fixture()
+@pytest.fixture
 def auth_client(session: Session, test_account: Account, test_user: User) -> Generator[TestClient, None, None]:
     """
     Provides a TestClient instance with valid authentication tokens.

tests/routers/core/test_account.py

@@ -1,4 +1,3 @@
-import pytest
 from fastapi.testclient import TestClient
 from starlette.datastructures import URLPath
 from sqlmodel import Session, select
@@ -16,9 +15,6 @@
     get_password_hash
 )
 
-# --- Fixture setup ---
-
-
 # --- API Endpoint Tests ---
 
 
@@ -124,7 +120,7 @@ def test_password_reset_flow(unauth_client: TestClient, session: Session, test_a
 
     # Verify content
     assert call_args["to"] == [test_account.email]
-    assert call_args["from"] == "noreply@promptlytechnologies.com"
+    assert call_args["from"] == "test@example.com"
     assert "Password Reset Request" in call_args["subject"]
     assert "reset_password" in call_args["html"]
 
@@ -259,7 +255,7 @@ def test_request_email_update_success(auth_client: TestClient, test_account: Acc
 
     # Verify email content
     assert call_args["to"] == [test_account.email]
-    assert call_args["from"] == "noreply@promptlytechnologies.com"
+    assert call_args["from"] == "test@example.com"
     assert "Confirm Email Update" in call_args["subject"]
     assert "confirm_email_update" in call_args["html"]
     assert new_email in call_args["html"]

utils/core/auth.py

@@ -5,7 +5,6 @@
 import uuid
 import logging
 import resend
-from dotenv import load_dotenv
 from sqlmodel import Session, select
 from bcrypt import gensalt, hashpw, checkpw
 from datetime import UTC, datetime, timedelta
@@ -15,9 +14,6 @@
 from fastapi import Cookie
 from utils.core.models import PasswordResetToken, EmailUpdateToken, Account
 
-load_dotenv(override=True)
-resend.api_key = os.environ["RESEND_API_KEY"]
-
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
 logger.addHandler(logging.StreamHandler())
@@ -27,7 +23,6 @@
 
 
 templates = Jinja2Templates(directory="templates")
-SECRET_KEY = os.getenv("SECRET_KEY")
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 REFRESH_TOKEN_EXPIRE_DAYS = 30
@@ -120,7 +115,7 @@ def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -
         expire = datetime.now(
             UTC) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    encoded_jwt = jwt.encode(to_encode, os.getenv("SECRET_KEY"), algorithm=ALGORITHM)
     return encoded_jwt
 
 
@@ -132,13 +127,13 @@ def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None)
     else:
         expire = datetime.now(UTC) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
     to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    encoded_jwt = jwt.encode(to_encode, os.getenv("SECRET_KEY"), algorithm=ALGORITHM)
     return encoded_jwt
 
 
 def validate_token(token: str, token_type: str = "access") -> Optional[dict]:
     try:
-        decoded_token = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        decoded_token = jwt.decode(token, os.getenv("SECRET_KEY"), algorithms=[ALGORITHM])
 
         # Check if the token has expired
         if decoded_token["exp"] < datetime.now(UTC).timestamp():
@@ -182,7 +177,7 @@ def send_reset_email(email: str, session: Session) -> None:
             .where(
                 PasswordResetToken.account_id == account.id,
                 PasswordResetToken.expires_at > datetime.now(UTC),
-                PasswordResetToken.used == False
+                not PasswordResetToken.used
             )
         ).first()
 
@@ -204,6 +199,7 @@ def send_reset_email(email: str, session: Session) -> None:
                 "emails/reset_email.html")
             html_content: str = template.render({"reset_url": reset_url})
 
+            resend.api_key = os.getenv("RESEND_API_KEY")
             params: resend.Emails.SendParams = {
                 "from": os.getenv("EMAIL_FROM", ""),
                 "to": [email],
@@ -242,7 +238,7 @@ def send_email_update_confirmation(
         .where(
             EmailUpdateToken.account_id == account_id,
             EmailUpdateToken.expires_at > datetime.now(UTC),
-            EmailUpdateToken.used == False
+            not EmailUpdateToken.used
         )
     ).first()
 
@@ -266,6 +262,7 @@ def send_email_update_confirmation(
             "new_email": new_email
         })
 
+        resend.api_key = os.getenv("RESEND_API_KEY")
         params: resend.Emails.SendParams = {
             "from": os.getenv("EMAIL_FROM", ""),
             "to": [current_email],

utils/core/db.py

@@ -1,15 +1,11 @@
 import os
 import logging
 from typing import Union, Sequence
-from dotenv import load_dotenv
 from sqlalchemy.engine import URL
-from sqlmodel import create_engine, Session, SQLModel, select
+from sqlmodel import create_engine, Session, SQLModel, select, text
 from utils.core.models import Role, Permission, RolePermissionLink
 from utils.core.enums import ValidPermissions
 
-# Load environment variables from a .env file
-load_dotenv()
-
 # Set up a logger for error reporting
 logger = logging.getLogger("uvicorn.error")
 
@@ -23,6 +19,19 @@
 # --- Database connection functions ---
 
 
+def ensure_database_exists(url: URL) -> None:
+    dbname = url.database
+    server_url = url.set(database="postgres")
+    engine = create_engine(server_url, isolation_level="AUTOCOMMIT")
+    with engine.connect() as conn:
+        exists = conn.execute(
+            text("SELECT 1 FROM pg_database WHERE datname = :n"),
+            {"n": dbname},
+        ).scalar()
+        if not exists:
+            conn.execute(text(f'CREATE DATABASE "{dbname}"'))
+            
+
 def get_connection_url() -> URL:
     """
     Constructs a SQLModel URL object for connecting to the PostgreSQL database.
@@ -49,10 +58,6 @@ def get_connection_url() -> URL:
     return database_url
 
 
-# Create the database engine using the connection URL
-engine = create_engine(get_connection_url())
-
-
 def assign_permissions_to_role(
         session: Session,
         role: Role,

utils/core/dependencies.py

@@ -8,7 +8,7 @@
     validate_token, create_access_token, create_refresh_token,
     oauth2_scheme_cookie, verify_password
 )
-from utils.core.db import engine
+from utils.core.db import create_engine, get_connection_url
 from utils.core.models import User, Role, PasswordResetToken, EmailUpdateToken, Account
 from exceptions.http_exceptions import AuthenticationError, CredentialsError, DataIntegrityError
 from exceptions.exceptions import NeedsNewTokens
@@ -21,6 +21,7 @@ def get_session() -> Generator[Session, None, None]:
     Yields:
         Session: A SQLModel session object for database operations.
     """
+    engine = create_engine(get_connection_url())
     with Session(engine) as session:
         yield session
 

utils/core/invitations.py

@@ -1,7 +1,6 @@
 import os
 from logging import getLogger, DEBUG
 import resend
-from dotenv import load_dotenv
 from sqlmodel import Session
 from jinja2.environment import Template
 from fastapi.templating import Jinja2Templates
@@ -10,10 +9,6 @@
 from exceptions.exceptions import EmailSendFailedError
 from exceptions.http_exceptions import DataIntegrityError
 
-# Load environment variables
-load_dotenv(override=True)
-resend.api_key = os.environ.get("RESEND_API_KEY")
-BASE_URL = os.getenv("BASE_URL", "")
 
 # Setup logging
 logger = getLogger("uvicorn.error")
@@ -33,7 +28,7 @@ def generate_invitation_link(token: str) -> str:
     Returns:
         The complete URL for accepting the invitation.
     """
-    return f"{BASE_URL}/invitations/accept?token={token}"
+    return f"{os.getenv('BASE_URL')}/invitations/accept?token={token}"
 
 
 def send_invitation_email(invitation: Invitation, session: Session) -> None:
@@ -45,6 +40,7 @@ def send_invitation_email(invitation: Invitation, session: Session) -> None:
         session: The database session (used here primarily for potential future needs or consistency,
                  though direct DB access might be minimal if invitation object is pre-loaded).
     """
+    resend.api_key = os.getenv("RESEND_API_KEY")
     if not resend.api_key:
         logger.error("Resend API key is not configured. Cannot send invitation email.")
         raise EmailSendFailedError("Resend API key is not configured.")