tests/conftest.py

import datetime
import pytest
import pkgutil
import importlib
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

from src.entity.fraud_details import FraudDetails
from src.entity.transaction import Transaction
from src.entity.api.transaction_api import TransactionApi
from src.main import app, provide_connection
from src.repository.common import get_session

from src.entity import Base

# ----------------------------------------------------------------
# Load all classes from src.entity
module = importlib.import_module('src.entity')

# Loop over all modules in the 'src.entity' package
package = importlib.import_module('src.entity')

for _, module_name, _ in pkgutil.walk_packages(package.__path__, package.__name__ + '.'):
    module = importlib.import_module(module_name)

@pytest.fixture
def client(db_session):
    """
    Create a test client for the FastAPI app.
    """
    # Override the get_session dependency to use the test database session
    def override_get_session():
        yield db_session

    # Override the dependency in the FastAPI app
    app.dependency_overrides[get_session] = override_get_session
    app.dependency_overrides[provide_connection] = override_get_session

    return TestClient(app)

# ----------------------------------------------------------------
# Fixtures for Database
# ----------------------------------------------------------------
@pytest.fixture
def db_session(postgresql):
    """
    Create a new database session for each test
    and tear it down after the test.
    """
    # Create a new database connection
    host = postgresql.info.host
    port = postgresql.info.port
    user = postgresql.info.user
    dbname = postgresql.info.dbname

    dsn = f"postgresql+psycopg://{user}@{host}:{port}/{dbname}"
    engine = create_engine(dsn, echo=True)

    # Create schema and tables once
    with engine.begin() as conn:
        conn.execute(text("CREATE SCHEMA IF NOT EXISTS public"))
    
    Base.metadata.create_all(engine)

    connection = engine.connect()
    connection.begin()

    SessionLocal = sessionmaker(bind=connection)
    session = SessionLocal()

    try:
        yield session
    except Exception:
        session.rollback()  # In case of an error, rollback the session
        raise
    finally:
        session.close()
        connection.close()

# ----------------------------------------------------------------
@pytest.fixture
def valid_transaction():
    return Transaction(
        customer_address_city_population=333497,
        customer_firstname='Jeff',
        customer_lastname='Elliott',
        customer_gender='M',
        customer_job='Mechanical engineer',
        customer_credit_card_number='2291163933867244',
        customer_address_street='351 Darlene Green',
        customer_address_city='Columbia',
        customer_address_state='SC',
        customer_address_zip='29209',
        customer_address_latitude=33.9659,
        customer_address_longitude=-80.9355,
        customer_dob='1968-03-19',
        transaction_number='2da90c7d74bd46a0caf3777415b3ebd3',
        transaction_amount=2.86,
        transaction_datetime=datetime.datetime(2020, 6, 21, 12, 14, 25),
        transaction_category='personal_care',
        merchant_name='fraud_Kirlin and Sons',
        merchant_address_latitude=33.986391,
        merchant_address_longitude=-81.200714,
    )

@pytest.fixture
def fraudulent_transaction():
    return Transaction(
        customer_address_city_population=23,
        customer_firstname='Brooke',
        customer_lastname='Smith',
        customer_gender='F',
        customer_job='Cytogeneticist',
        customer_credit_card_number='3560725013359375',
        customer_address_street='63542 Luna Brook Apt. 012',
        customer_address_city='Notrees',
        customer_address_state='TX',
        customer_address_zip='79759',
        customer_address_latitude=31.8599,
        customer_address_longitude=-102.7413,
        customer_dob='1969-09-15',
        transaction_number='16bf2e46c54369a8eab2214649506425',
        transaction_amount=2400000.84,
        transaction_datetime=datetime.datetime(2020, 6, 21, 22, 6, 39),
        transaction_category='health_fitness',
        merchant_name="fraud_Hamill-D'Amore",
        merchant_address_latitude=32.575873,
        merchant_address_longitude=-102.60429,
    )

@pytest.fixture
def transaction_api():
    return TransactionApi(
        transaction_number="123456789",
        transaction_timestamp=1633036800,
        transaction_amount=100.0,
        transaction_category="Electronics",
        merchant_name="Best Buy",
        merchant_latitude=37.7749,
        merchant_longitude=-122.4194,
        customer_credit_card_number="4111111111111111",
        customer_gender="M",
        customer_first_name="John",
        customer_last_name="Doe",
        customer_date_of_birth="1980-01-01",
        customer_job="Engineer",
        customer_street="123 Main St",
        customer_city="San Francisco",
        customer_state="CA",
        customer_postal_code=94105,
        customer_latitude=37.7749,
        customer_longitude=-122.4194,
        customer_city_population=870000,
    )

# ----------------------------------------------------------------
# Fixtures in database
# ----------------------------------------------------------------
@pytest.fixture
def valid_transaction_in_db(db_session, valid_transaction: Transaction):
    """
    Create a valid transaction and add it to the database
    """
    db_session.add(valid_transaction)
    db_session.commit()
    db_session.refresh(valid_transaction)

    return valid_transaction

@pytest.fixture
def fraudulent_transaction_in_db(db_session, fraudulent_transaction: Transaction):
    """
    Create a fraudulent transaction and add it to the database
    """
    fraud_details = FraudDetails(
        fraud_score=0.99,
        model_version="v1.0",
        notification_sent=False,
        notification_recipients=None,
        notification_datetime=None,
    )

    # Add the fraud details to the transaction
    fraudulent_transaction.fraud_details = fraud_details

    # Add the transaction to the database
    db_session.add(fraudulent_transaction)
    db_session.commit()
    db_session.refresh(fraudulent_transaction)

    return fraudulent_transaction