226 lines
6.2 KiB
Python
226 lines
6.2 KiB
Python
"""Pytest configuration and fixtures"""
|
|
import logging
|
|
|
|
import pytest
|
|
from faker import Faker
|
|
|
|
from app import create_app, db
|
|
from app.models import Order, OrderItem, Product, User
|
|
|
|
log = logging.getLogger(__name__)
|
|
fake = Faker()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def app():
|
|
"""Create application for testing with PostgreSQL database (session scope)"""
|
|
app = create_app(config_name="test")
|
|
app.config.update(
|
|
{
|
|
"TESTING": True,
|
|
# fmt: off
|
|
"WTF_CSRF_ENABLED": False,
|
|
"JWT_SECRET_KEY":
|
|
"test-secret-keytest-secret-keytest-secret-keytest-secret-key",
|
|
"SERVER_NAME": "localhost.localdomain",
|
|
# fmt: on
|
|
}
|
|
)
|
|
|
|
# Create tables once per session
|
|
with app.app_context():
|
|
print("--------db.create_all()------")
|
|
db.create_all()
|
|
yield app
|
|
# Cleanup after all tests
|
|
db.session.remove()
|
|
db.drop_all()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Test client for making requests"""
|
|
|
|
log.debug("Test finished - session dirty: %s", db.session.dirty)
|
|
log.debug("Test finished - session new: %s", db.session.new)
|
|
return app.test_client()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _cleanup_db_after_test(app):
|
|
"""
|
|
Automatically rollback and remove DB session after EACH test function.
|
|
|
|
- Runs after every test function (default scope='function')
|
|
- Depends on 'app' to ensure app context is available
|
|
- Rollback prevents data leakage between tests
|
|
- remove() returns connection to pool (prevents exhaustion)
|
|
"""
|
|
|
|
yield # Let the test run
|
|
with app.app_context():
|
|
db.session.rollback() # Undo all changes from the test
|
|
db.session.remove() # Return connection to pool
|
|
|
|
|
|
@pytest.fixture
|
|
def runner(app):
|
|
"""Test CLI runner"""
|
|
return app.test_cli_runner()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _rollback_db_session(app):
|
|
"""Automatically rollback db.session after each test for isolation.
|
|
|
|
This fixture uses the shared Flask-SQLAlchemy session but ensures
|
|
tests don't interfere with each other by rolling back after each test.
|
|
"""
|
|
yield
|
|
with app.app_context():
|
|
db.session.rollback()
|
|
db.session.remove()
|
|
|
|
|
|
@pytest.fixture
|
|
def db_session(app):
|
|
"""Database session for tests (function scope)"""
|
|
"""Provide the shared Flask-SQLAlchemy db.session for tests"""
|
|
with app.app_context():
|
|
yield db.session
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_user(db_session):
|
|
"""Create an admin user for testing"""
|
|
user = User(
|
|
email=fake.email(),
|
|
username=fake.user_name(),
|
|
first_name=fake.first_name(),
|
|
last_name=fake.last_name(),
|
|
is_admin=True,
|
|
is_active=True,
|
|
)
|
|
user.set_password("password123")
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def regular_user(db_session):
|
|
"""Create a regular user for testing"""
|
|
user = User(
|
|
email=fake.email(),
|
|
username=fake.user_name(),
|
|
first_name=fake.first_name(),
|
|
last_name=fake.last_name(),
|
|
is_admin=False,
|
|
is_active=True,
|
|
)
|
|
user.set_password("password123")
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def inactive_user(db_session):
|
|
"""Create an inactive user for testing"""
|
|
user = User(
|
|
email=fake.email(),
|
|
username=fake.user_name(),
|
|
first_name=fake.first_name(),
|
|
last_name=fake.last_name(),
|
|
is_admin=False,
|
|
is_active=False,
|
|
)
|
|
user.set_password("password123")
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def product(db_session):
|
|
"""Create a product for testing"""
|
|
product = Product(
|
|
name=fake.sentence(nb_words=4)[:-1], # Remove period
|
|
description=fake.paragraph(),
|
|
price=fake.pydecimal(left_digits=2, right_digits=2, positive=True),
|
|
stock=fake.pyint(min_value=0, max_value=100),
|
|
image_url=fake.url(),
|
|
)
|
|
db_session.add(product)
|
|
db_session.commit()
|
|
return product
|
|
|
|
|
|
@pytest.fixture
|
|
def products(db_session):
|
|
"""Create multiple products for testing"""
|
|
products = []
|
|
for _ in range(5):
|
|
product = Product(
|
|
name=fake.sentence(nb_words=4)[:-1],
|
|
description=fake.paragraph(),
|
|
price=fake.pydecimal(left_digits=2, right_digits=2, positive=True),
|
|
stock=fake.pyint(min_value=20, max_value=100),
|
|
image_url=fake.url(),
|
|
)
|
|
db_session.add(product)
|
|
products.append(product)
|
|
db_session.commit()
|
|
return products
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_headers(client, regular_user):
|
|
"""Get authentication headers for a regular user"""
|
|
response = client.post(
|
|
"/api/auth/login", json={"email": regular_user.email, "password": "password123"}
|
|
)
|
|
data = response.get_json()
|
|
response.close()
|
|
token = data["access_token"]
|
|
print(f"Auth headers token for user {regular_user.email}: {token[:50]}...")
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_headers(client, admin_user):
|
|
"""Get authentication headers for an admin user"""
|
|
response = client.post(
|
|
"/api/auth/login", json={"email": admin_user.email, "password": "password123"}
|
|
)
|
|
data = response.get_json()
|
|
token = data["access_token"]
|
|
print(f"Admin headers token for user {admin_user.email}: {token[:50]}...")
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.fixture
|
|
def order(db_session, regular_user, products):
|
|
print("-----order-created------")
|
|
"""Create an order for testing"""
|
|
order = Order(
|
|
user_id=regular_user.id, total_amount=0.0, shipping_address=fake.address()
|
|
)
|
|
db_session.add(order)
|
|
db_session.flush()
|
|
|
|
total_amount = 0
|
|
for i, product in enumerate(products[:2]):
|
|
quantity = fake.pyint(min_value=1, max_value=5)
|
|
order_item = OrderItem(
|
|
order_id=order.id,
|
|
product_id=product.id,
|
|
quantity=quantity,
|
|
price=product.price,
|
|
)
|
|
total_amount += float(product.price) * quantity
|
|
db_session.add(order_item)
|
|
|
|
order.total_amount = total_amount
|
|
db_session.commit()
|
|
return order
|