add pytest

This commit is contained in:
david 2026-02-24 17:36:31 +03:00
parent 068a1b38f9
commit a9b0fd34a2
21 changed files with 2169 additions and 129 deletions

143
.github/workflows/backend-tests.yml vendored Normal file
View file

@ -0,0 +1,143 @@
name: Backend Tests
on:
push:
branches: [ main, develop ]
paths:
- 'backend/**'
- '.github/workflows/backend-tests.yml'
pull_request:
branches: [ main, develop ]
paths:
- 'backend/**'
- '.github/workflows/backend-tests.yml'
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10', '3.11', '3.12']
services:
postgres:
image: postgres:15
env:
POSTGRES_USER: test_user
POSTGRES_PASSWORD: test_password
POSTGRES_DB: test_db
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
redis:
image: redis:7-alpine
options: >-
--health-cmd "redis-cli ping"
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 6379:6379
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
cache-dependency-path: 'backend/requirements/*.txt'
- name: Install dependencies
working-directory: ./backend
run: |
python -m pip install --upgrade pip
pip install -r requirements/base.txt
pip install -r requirements/dev.txt
- name: Lint with flake8 (if installed)
working-directory: ./backend
run: |
pip install flake8
# Stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
# Exit-zero treats all errors as warnings
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
continue-on-error: true
- name: Run tests with pytest
working-directory: ./backend
env:
DATABASE_URL: postgresql://test_user:test_password@localhost:5432/test_db
CELERY_BROKER_URL: redis://localhost:6379/0
CELERY_RESULT_BACKEND: redis://localhost:6379/0
SECRET_KEY: test-secret-key
JWT_SECRET_KEY: test-jwt-secret-key
FLASK_ENV: testing
run: |
pytest --cov=app --cov-report=xml --cov-report=html --cov-report=term
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
file: ./backend/coverage.xml
flags: backend
name: codecov-umbrella
fail_ci_if_error: false
- name: Upload coverage HTML as artifact
uses: actions/upload-artifact@v3
with:
name: coverage-report-python-${{ matrix.python-version }}
path: backend/htmlcov/
retention-days: 7
- name: Check coverage thresholds
working-directory: ./backend
run: |
coverage report --fail-under=80
security-scan:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
working-directory: ./backend
run: |
python -m pip install --upgrade pip
pip install bandit safety
- name: Run Bandit security linter
working-directory: ./backend
run: |
bandit -r app -f json -o bandit-report.json || true
continue-on-error: true
- name: Check for known security vulnerabilities
working-directory: ./backend
run: |
safety check --json --output safety-report.json || true
continue-on-error: true
- name: Upload security reports
uses: actions/upload-artifact@v3
with:
name: security-reports
path: |
backend/bandit-report.json
backend/safety-report.json

65
.pre-commit-config.yaml Normal file
View file

@ -0,0 +1,65 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
exclude: ^.+\.md$
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
args: ['--maxkb=1000']
- id: check-json
- id: check-toml
- id: check-merge-conflict
- id: debug-statements
language: python
- repo: https://github.com/psf/black
rev: 23.12.1
hooks:
- id: black
language_version: python3.11
args: ['--line-length=100']
- repo: https://github.com/pycqa/isort
rev: 5.13.2
hooks:
- id: isort
args: ['--profile=black', '--line-length=100']
- repo: https://github.com/pycqa/flake8
rev: 7.0.0
hooks:
- id: flake8
args: ['--max-line-length=100', '--extend-ignore=E203,W503']
additional_dependencies: [
flake8-docstrings,
flake8-bugbear,
flake8-comprehensions,
]
- repo: local
hooks:
- id: pytest
name: pytest
entry: pytest
language: system
pass_filenames: false
args: ['backend/', '-v', '--tb=short']
always_run: true
- id: type-check
name: mypy type check
entry: mypy
language: system
pass_filenames: false
args: ['backend/app/']
always_run: false
- id: security-check
name: bandit security check
entry: bandit
language: system
pass_filenames: false
args: ['-r', 'backend/app/', '-ll']
always_run: false

View file

@ -67,6 +67,48 @@ test: ## Run all tests
test-backend: ## Run backend tests only test-backend: ## Run backend tests only
cd backend && . venv/bin/activate && pytest cd backend && . venv/bin/activate && pytest
test-backend-cov: ## Run backend tests with coverage
cd backend && . venv/bin/activate && pytest --cov=app --cov-report=html --cov-report=term
test-backend-verbose: ## Run backend tests with verbose output
cd backend && . venv/bin/activate && pytest -v
test-backend-unit: ## Run backend unit tests only
cd backend && . venv/bin/activate && pytest -m unit
test-backend-integration: ## Run backend integration tests only
cd backend && . venv/bin/activate && pytest -m integration
test-backend-auth: ## Run backend authentication tests only
cd backend && . venv/bin/activate && pytest -m auth
test-backend-product: ## Run backend product tests only
cd backend && . venv/bin/activate && pytest -m product
test-backend-order: ## Run backend order tests only
cd backend && . venv/bin/activate && pytest -m order
test-backend-watch: ## Run backend tests in watch mode (auto-rerun on changes)
cd backend && . venv/bin/activate && pip install pytest-watch && pytest-watch
test-backend-parallel: ## Run backend tests in parallel (faster)
cd backend && . venv/bin/activate && pip install pytest-xdist && pytest -n auto
test-backend-coverage-report: ## Open backend coverage report in browser
cd backend && . venv/bin/activate && pytest --cov=app --cov-report=html && python -m webbrowser htmlcov/index.html
test-backend-failed: ## Re-run only failed backend tests
cd backend && . venv/bin/activate && pytest --lf
test-backend-last-failed: ## Run the tests that failed in the last run
cd backend && . venv/bin/activate && pytest --lf
test-backend-specific: ## Run specific backend test (usage: make test-backend-specific TEST=test_models.py)
cd backend && . venv/bin/activate && pytest tests/$(TEST)
test-backend-marker: ## Run backend tests by marker (usage: make test-backend-marker MARKER=auth)
cd backend && . venv/bin/activate && pytest -m $(MARKER)
test-frontend: ## Run frontend tests only test-frontend: ## Run frontend tests only
cd frontend && npm test cd frontend && npm test

23
backend/.coveragerc Normal file
View file

@ -0,0 +1,23 @@
[run]
source = app
omit =
*/tests/*
*/migrations/*
*/__pycache__/*
*/venv/*
*/instance/*
app/__init__.py
[report]
exclude_lines =
pragma: no cover
def __repr__
raise NotImplementedError
if __name__ == .__main__.:
if TYPE_CHECKING:
@abstractmethod
pass
precision = 2
[html]
directory = htmlcov

View file

@ -50,10 +50,17 @@ def create_app(config_name=None):
# Global error handlers # Global error handlers
@app.errorhandler(404) @app.errorhandler(404)
def not_found(error): def not_found(error):
print(f"404 Error: {error}")
return jsonify({"error": "Not found"}), 404 return jsonify({"error": "Not found"}), 404
@app.errorhandler(500) @app.errorhandler(500)
def internal_error(error): def internal_error(error):
print(f"500 Error: {error}")
return jsonify({"error": "Internal server error"}), 500 return jsonify({"error": "Internal server error"}), 500
@app.errorhandler(422)
def validation_error(error):
print(f"422 Error: {error}")
return jsonify({"error": "Validation error"}), 422
return app return app

View file

@ -6,7 +6,7 @@ class Config:
"""Base configuration""" """Base configuration"""
SECRET_KEY = os.environ.get("SECRET_KEY") or "dev-secret-key-change-in-production" SECRET_KEY = os.environ.get("SECRET_KEY") or "dev-secret-key-change-in-production"
SQLALCHEMY_TRACK_MODIFICATIONS = False SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY") or "jwt-secret-key-change-in-production" JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1) JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30) JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
CORS_ORIGINS = os.environ.get("CORS_ORIGINS", "*") CORS_ORIGINS = os.environ.get("CORS_ORIGINS", "*")

View file

@ -1,4 +1,4 @@
from datetime import datetime from datetime import datetime, UTC
from app import db from app import db
@ -11,8 +11,8 @@ class Order(db.Model):
status = db.Column(db.String(20), default="pending", index=True) status = db.Column(db.String(20), default="pending", index=True)
total_amount = db.Column(db.Numeric(10, 2), nullable=False) total_amount = db.Column(db.Numeric(10, 2), nullable=False)
shipping_address = db.Column(db.Text) shipping_address = db.Column(db.Text)
created_at = db.Column(db.DateTime, default=datetime.utcnow) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
# Relationships # Relationships
user = db.relationship("User", back_populates="orders") user = db.relationship("User", back_populates="orders")

View file

@ -1,4 +1,4 @@
from datetime import datetime from datetime import datetime, UTC
from app import db from app import db
@ -13,8 +13,8 @@ class Product(db.Model):
stock = db.Column(db.Integer, default=0) stock = db.Column(db.Integer, default=0)
image_url = db.Column(db.String(500)) image_url = db.Column(db.String(500))
is_active = db.Column(db.Boolean, default=True) is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
# Relationships # Relationships
order_items = db.relationship("OrderItem", back_populates="product", lazy="dynamic") order_items = db.relationship("OrderItem", back_populates="product", lazy="dynamic")

View file

@ -1,4 +1,4 @@
from datetime import datetime from datetime import datetime, UTC
from werkzeug.security import generate_password_hash, check_password_hash from werkzeug.security import generate_password_hash, check_password_hash
from app import db from app import db
@ -15,8 +15,8 @@ class User(db.Model):
last_name = db.Column(db.String(50)) last_name = db.Column(db.String(50))
is_active = db.Column(db.Boolean, default=True) is_active = db.Column(db.Boolean, default=True)
is_admin = db.Column(db.Boolean, default=False) is_admin = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) updated_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
# Relationships # Relationships
orders = db.relationship("Order", back_populates="user", lazy="dynamic") orders = db.relationship("Order", back_populates="user", lazy="dynamic")

View file

@ -54,8 +54,8 @@ def login():
if not user.is_active: if not user.is_active:
return jsonify({"error": "Account is inactive"}), 401 return jsonify({"error": "Account is inactive"}), 401
access_token = create_access_token(identity=user.id) access_token = create_access_token(identity=str(user.id))
refresh_token = create_refresh_token(identity=user.id) refresh_token = create_refresh_token(identity=str(user.id))
return jsonify({ return jsonify({
"user": user.to_dict(), "user": user.to_dict(),
@ -68,8 +68,8 @@ def login():
@jwt_required() @jwt_required()
def get_current_user(): def get_current_user():
"""Get current user""" """Get current user"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
user = User.query.get(user_id) user = db.session.get(User, user_id)
if not user: if not user:
return jsonify({"error": "User not found"}), 404 return jsonify({"error": "User not found"}), 404
@ -93,7 +93,9 @@ def get_products():
@api_bp.route("/products/<int:product_id>", methods=["GET"]) @api_bp.route("/products/<int:product_id>", methods=["GET"])
def get_product(product_id): def get_product(product_id):
"""Get a single product""" """Get a single product"""
product = Product.query.get_or_404(product_id) product = db.session.get(Product, product_id)
if not product:
return jsonify({"error": "Product not found"}), 404
return jsonify(product.to_dict()), 200 return jsonify(product.to_dict()), 200
@ -101,8 +103,8 @@ def get_product(product_id):
@jwt_required() @jwt_required()
def create_product(): def create_product():
"""Create a new product (admin only)""" """Create a new product (admin only)"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
user = User.query.get(user_id) user = db.session.get(User, user_id)
if not user or not user.is_admin: if not user or not user.is_admin:
return jsonify({"error": "Admin access required"}), 403 return jsonify({"error": "Admin access required"}), 403
@ -116,8 +118,7 @@ def create_product():
description=product_data.description, description=product_data.description,
price=product_data.price, price=product_data.price,
stock=product_data.stock, stock=product_data.stock,
image_url=product_data.image_url, image_url=product_data.image_url
category=product_data.category
) )
db.session.add(product) db.session.add(product)
@ -128,6 +129,7 @@ def create_product():
return jsonify(response.model_dump()), 201 return jsonify(response.model_dump()), 201
except ValidationError as e: except ValidationError as e:
print(f"Pydantic Validation Error: {e.errors()}")
return jsonify({"error": "Validation error", "details": e.errors()}), 400 return jsonify({"error": "Validation error", "details": e.errors()}), 400
@ -135,13 +137,16 @@ def create_product():
@jwt_required() @jwt_required()
def update_product(product_id): def update_product(product_id):
"""Update a product (admin only)""" """Update a product (admin only)"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
user = User.query.get(user_id) user = db.session.get(User, user_id)
if not user or not user.is_admin: if not user or not user.is_admin:
return jsonify({"error": "Admin access required"}), 403 return jsonify({"error": "Admin access required"}), 403
product = Product.query.get_or_404(product_id) product = db.session.get(Product, product_id)
if not product:
return jsonify({"error": "Product not found"}), 404
data = request.get_json() data = request.get_json()
product.name = data.get("name", product.name) product.name = data.get("name", product.name)
@ -149,7 +154,6 @@ def update_product(product_id):
product.price = data.get("price", product.price) product.price = data.get("price", product.price)
product.stock = data.get("stock", product.stock) product.stock = data.get("stock", product.stock)
product.image_url = data.get("image_url", product.image_url) product.image_url = data.get("image_url", product.image_url)
product.category = data.get("category", product.category)
product.is_active = data.get("is_active", product.is_active) product.is_active = data.get("is_active", product.is_active)
db.session.commit() db.session.commit()
@ -161,13 +165,16 @@ def update_product(product_id):
@jwt_required() @jwt_required()
def delete_product(product_id): def delete_product(product_id):
"""Delete a product (admin only)""" """Delete a product (admin only)"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
user = User.query.get(user_id) user = db.session.get(User, user_id)
if not user or not user.is_admin: if not user or not user.is_admin:
return jsonify({"error": "Admin access required"}), 403 return jsonify({"error": "Admin access required"}), 403
product = Product.query.get_or_404(product_id) product = db.session.get(Product, product_id)
if not product:
return jsonify({"error": "Product not found"}), 404
db.session.delete(product) db.session.delete(product)
db.session.commit() db.session.commit()
@ -179,7 +186,7 @@ def delete_product(product_id):
@jwt_required() @jwt_required()
def get_orders(): def get_orders():
"""Get all orders for current user""" """Get all orders for current user"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
orders = Order.query.filter_by(user_id=user_id).all() orders = Order.query.filter_by(user_id=user_id).all()
return jsonify([order.to_dict() for order in orders]), 200 return jsonify([order.to_dict() for order in orders]), 200
@ -188,7 +195,7 @@ def get_orders():
@jwt_required() @jwt_required()
def create_order(): def create_order():
"""Create a new order""" """Create a new order"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
data = request.get_json() data = request.get_json()
if not data or not data.get("items"): if not data or not data.get("items"):
@ -198,7 +205,7 @@ def create_order():
order_items = [] order_items = []
for item_data in data["items"]: for item_data in data["items"]:
product = Product.query.get(item_data["product_id"]) product = db.session.get(Product, item_data["product_id"])
if not product: if not product:
return jsonify({"error": f'Product {item_data["product_id"]} not found'}), 404 return jsonify({"error": f'Product {item_data["product_id"]} not found'}), 404
if product.stock < item_data["quantity"]: if product.stock < item_data["quantity"]:
@ -215,8 +222,7 @@ def create_order():
order = Order( order = Order(
user_id=user_id, user_id=user_id,
total_amount=total_amount, total_amount=total_amount,
shipping_address=data.get("shipping_address"), shipping_address=data.get("shipping_address")
notes=data.get("notes")
) )
db.session.add(order) db.session.add(order)
@ -241,11 +247,13 @@ def create_order():
@jwt_required() @jwt_required()
def get_order(order_id): def get_order(order_id):
"""Get a single order""" """Get a single order"""
user_id = get_jwt_identity() user_id = int(get_jwt_identity())
order = Order.query.get_or_404(order_id) order = db.session.get(Order, order_id)
if not order:
return jsonify({"error": "Order not found"}), 404
if order.user_id != user_id: if order.user_id != user_id:
user = User.query.get(user_id) user = db.session.get(User, user_id)
if not user or not user.is_admin: if not user or not user.is_admin:
return jsonify({"error": "Access denied"}), 403 return jsonify({"error": "Access denied"}), 403

View file

@ -1,5 +1,5 @@
"""Pydantic schemas for Product model""" """Pydantic schemas for Product model"""
from pydantic import BaseModel, Field, field_validator from pydantic import BaseModel, Field, field_validator, ConfigDict
from decimal import Decimal from decimal import Decimal
from datetime import datetime from datetime import datetime
from typing import Optional from typing import Optional
@ -7,24 +7,23 @@ from typing import Optional
class ProductCreateRequest(BaseModel): class ProductCreateRequest(BaseModel):
"""Schema for creating a new product""" """Schema for creating a new product"""
name: str = Field(..., min_length=1, max_length=200, description="Product name") model_config = ConfigDict(
description: Optional[str] = Field(None, description="Product description") json_schema_extra={
price: Decimal = Field(..., gt=0, description="Product price (must be greater than 0)")
stock: int = Field(default=0, ge=0, description="Product stock quantity")
image_url: Optional[str] = Field(None, max_length=500, description="Product image URL")
category: Optional[str] = Field(None, description="Product category")
class Config:
json_schema_extra = {
"example": { "example": {
"name": "Handcrafted Wooden Bowl", "name": "Handcrafted Wooden Bowl",
"description": "A beautiful handcrafted bowl made from oak", "description": "A beautiful handcrafted bowl made from oak",
"price": 45.99, "price": 45.99,
"stock": 10, "stock": 10,
"image_url": "https://example.com/bowl.jpg", "image_url": "https://example.com/bowl.jpg"
"category": "Woodwork"
} }
} }
)
name: str = Field(..., min_length=1, max_length=200, description="Product name")
description: Optional[str] = Field(None, description="Product description")
price: Decimal = Field(..., gt=0, description="Product price (must be greater than 0)")
stock: int = Field(default=0, ge=0, description="Product stock quantity")
image_url: Optional[str] = Field(None, max_length=500, description="Product image URL")
@field_validator("price") @field_validator("price")
@classmethod @classmethod
@ -37,19 +36,9 @@ class ProductCreateRequest(BaseModel):
class ProductResponse(BaseModel): class ProductResponse(BaseModel):
"""Schema for product response""" """Schema for product response"""
id: int model_config = ConfigDict(
name: str from_attributes=True,
description: Optional[str] = None json_schema_extra={
price: float
stock: int
image_url: Optional[str] = None
is_active: bool
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class Config:
from_attributes = True
json_schema_extra = {
"example": { "example": {
"id": 1, "id": 1,
"name": "Handcrafted Wooden Bowl", "name": "Handcrafted Wooden Bowl",
@ -62,3 +51,14 @@ class ProductResponse(BaseModel):
"updated_at": "2024-01-15T10:30:00" "updated_at": "2024-01-15T10:30:00"
} }
} }
)
id: int
name: str
description: Optional[str] = None
price: float
stock: int
image_url: Optional[str] = None
is_active: bool
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None

View file

@ -1,69 +0,0 @@
import os
from datetime import timedelta
class Config:
"""Base configuration"""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'
SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret-key-change-in-production'
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
# Celery Configuration
CELERY = {
"broker_url": os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379/0"),
"result_backend": os.environ.get("CELERY_RESULT_BACKEND", "redis://localhost:6379/0"),
"task_serializer": "json",
"result_serializer": "json",
"accept_content": ["json"],
"timezone": "UTC",
"enable_utc": True,
"task_track_started": True,
"task_time_limit": 30 * 60, # 30 minutes
"task_soft_time_limit": 25 * 60, # 25 minutes
"task_acks_late": True, # Acknowledge after task completion
"task_reject_on_worker_lost": True, # Re-queue if worker dies
"worker_prefetch_multiplier": 1, # Process one task at a time
"worker_max_tasks_per_child": 100, # Restart worker after 100 tasks
"broker_connection_retry_on_startup": True,
"broker_connection_max_retries": 5,
"result_expires": 3600, # Results expire in 1 hour
"task_default_queue": "default",
"task_default_exchange": "default",
"task_default_routing_key": "default",
}
class DevelopmentConfig(Config):
"""Development configuration"""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///dev.db'
class TestingConfig(Config):
"""Testing configuration"""
TESTING = True
SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
'sqlite:///test.db'
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
"""Production configuration"""
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'postgresql://user:password@localhost/proddb'
# Security headers
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
config_by_name = {
'dev': DevelopmentConfig,
'test': TestingConfig,
'prod': ProductionConfig
}

19
backend/pytest.ini Normal file
View file

@ -0,0 +1,19 @@
[pytest]
python_files = test_*.py
python_classes = Test*
python_functions = test_*
testpaths = tests
addopts =
-v
--strict-markers
--tb=short
--cov=app
--cov-report=term-missing
--cov-report=html
markers =
slow: Tests that are slow to run
integration: Integration tests
unit: Unit tests
auth: Authentication tests
product: Product-related tests
order: Order-related tests

View file

@ -9,3 +9,9 @@ Werkzeug==3.0.1
SQLAlchemy==2.0.23 SQLAlchemy==2.0.23
celery[redis]==5.3.6 celery[redis]==5.3.6
pydantic==2.5.3 pydantic==2.5.3
pytest==7.4.3
pytest-flask==1.3.0
pytest-cov==4.1.0
pytest-mock==3.12.0
factory-boy==3.3.0
faker==20.1.0

View file

@ -0,0 +1 @@
"""Tests package for Flask application"""

191
backend/tests/conftest.py Normal file
View file

@ -0,0 +1,191 @@
"""Pytest configuration and fixtures"""
import pytest
import tempfile
import os
from faker import Faker
from app import create_app, db
from app.models import User, Product, Order, OrderItem
fake = Faker()
@pytest.fixture(scope='function')
def app():
"""Create application for testing with isolated database"""
db_fd, db_path = tempfile.mkstemp()
app = create_app(config_name='test')
app.config.update({
'TESTING': True,
'SQLALCHEMY_DATABASE_URI': f'sqlite:///{db_path}',
'WTF_CSRF_ENABLED': False,
'JWT_SECRET_KEY': 'test-secret-keytest-secret-keytest-secret-keytest-secret-keytest-secret-key',
'SERVER_NAME': 'localhost.localdomain'
})
with app.app_context():
db.create_all()
yield app
db.session.remove()
db.drop_all()
os.close(db_fd)
os.unlink(db_path)
@pytest.fixture
def client(app):
"""Test client for making requests"""
return app.test_client()
@pytest.fixture
def runner(app):
"""Test CLI runner"""
return app.test_cli_runner()
@pytest.fixture
def db_session(app):
"""Database 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=0, 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()
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):
"""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

View file

@ -0,0 +1,208 @@
"""Test models"""
import pytest
from decimal import Decimal
from datetime import datetime
from app.models import User, Product, Order, OrderItem
class TestUserModel:
"""Test User model"""
@pytest.mark.unit
def test_user_creation(self, db_session):
"""Test creating a user"""
user = User(
email='test@example.com',
username='testuser',
first_name='Test',
last_name='User',
is_admin=False,
is_active=True
)
user.set_password('password123')
db_session.add(user)
db_session.commit()
assert user.id is not None
assert user.email == 'test@example.com'
assert user.username == 'testuser'
assert user.first_name == 'Test'
assert user.last_name == 'User'
@pytest.mark.unit
def test_user_password_hashing(self, db_session):
"""Test password hashing and verification"""
user = User(email='test@example.com', username='testuser')
user.set_password('password123')
db_session.add(user)
db_session.commit()
assert user.check_password('password123') is True
assert user.check_password('wrongpassword') is False
@pytest.mark.unit
def test_user_to_dict(self, db_session):
"""Test user serialization to dictionary"""
user = User(
email='test@example.com',
username='testuser',
first_name='Test',
last_name='User'
)
user.set_password('password123')
db_session.add(user)
db_session.commit()
user_dict = user.to_dict()
assert user_dict['email'] == 'test@example.com'
assert user_dict['username'] == 'testuser'
assert 'password' not in user_dict
assert 'password_hash' not in user_dict
@pytest.mark.unit
def test_user_repr(self, db_session):
"""Test user string representation"""
user = User(email='test@example.com', username='testuser')
user.set_password('password123')
db_session.add(user)
db_session.commit()
assert repr(user) == '<User testuser>'
class TestProductModel:
"""Test Product model"""
@pytest.mark.unit
def test_product_creation(self, db_session):
"""Test creating a product"""
product = Product(
name='Test Product',
description='A test product',
price=Decimal('99.99'),
stock=10,
image_url='https://example.com/product.jpg'
)
db_session.add(product)
db_session.commit()
assert product.id is not None
assert product.name == 'Test Product'
assert product.price == Decimal('99.99')
assert product.stock == 10
assert product.is_active is True
@pytest.mark.unit
def test_product_to_dict(self, db_session):
"""Test product serialization to dictionary"""
product = Product(
name='Test Product',
description='A test product',
price=Decimal('99.99'),
stock=10
)
db_session.add(product)
db_session.commit()
product_dict = product.to_dict()
assert product_dict['name'] == 'Test Product'
assert product_dict['price'] == 99.99
assert isinstance(product_dict['created_at'], str)
assert isinstance(product_dict['updated_at'], str)
@pytest.mark.unit
def test_product_defaults(self, db_session):
"""Test product default values"""
product = Product(
name='Test Product',
price=Decimal('9.99')
)
db_session.add(product)
db_session.commit()
assert product.stock == 0
assert product.is_active is True
assert product.description is None
assert product.image_url is None
@pytest.mark.unit
def test_product_repr(self, db_session):
"""Test product string representation"""
product = Product(name='Test Product', price=Decimal('9.99'))
db_session.add(product)
db_session.commit()
assert repr(product) == '<Product Test Product>'
class TestOrderModel:
"""Test Order model"""
@pytest.mark.unit
def test_order_creation(self, db_session, regular_user):
"""Test creating an order"""
order = Order(
user_id=regular_user.id,
total_amount=Decimal('199.99'),
shipping_address='123 Test St'
)
db_session.add(order)
db_session.commit()
assert order.id is not None
assert order.user_id == regular_user.id
assert order.total_amount == Decimal('199.99')
@pytest.mark.unit
def test_order_to_dict(self, db_session, regular_user):
"""Test order serialization to dictionary"""
order = Order(
user_id=regular_user.id,
total_amount=Decimal('199.99'),
shipping_address='123 Test St'
)
db_session.add(order)
db_session.commit()
order_dict = order.to_dict()
assert order_dict['user_id'] == regular_user.id
assert order_dict['total_amount'] == 199.99
assert isinstance(order_dict['created_at'], str)
class TestOrderItemModel:
"""Test OrderItem model"""
@pytest.mark.unit
def test_order_item_creation(self, db_session, order, product):
"""Test creating an order item"""
order_item = OrderItem(
order_id=order.id,
product_id=product.id,
quantity=2,
price=product.price
)
db_session.add(order_item)
db_session.commit()
assert order_item.id is not None
assert order_item.order_id == order.id
assert order_item.product_id == product.id
assert order_item.quantity == 2
@pytest.mark.unit
def test_order_item_to_dict(self, db_session, order, product):
"""Test order item serialization to dictionary"""
order_item = OrderItem(
order_id=order.id,
product_id=product.id,
quantity=2,
price=product.price
)
db_session.add(order_item)
db_session.commit()
item_dict = order_item.to_dict()
assert item_dict['order_id'] == order.id
assert item_dict['product_id'] == product.id
assert item_dict['quantity'] == 2

View file

@ -0,0 +1,319 @@
"""Test API routes"""
import pytest
import json
from decimal import Decimal
class TestAuthRoutes:
"""Test authentication routes"""
@pytest.mark.auth
def test_register_success(self, client):
"""Test successful user registration"""
response = client.post('/api/auth/register', json={
'email': 'newuser@example.com',
'password': 'password123',
'username': 'newuser',
'first_name': 'New',
'last_name': 'User'
})
assert response.status_code == 201
data = response.get_json()
assert data['email'] == 'newuser@example.com'
assert data['username'] == 'newuser'
assert 'password' not in data
assert 'password_hash' not in data
@pytest.mark.auth
def test_register_missing_fields(self, client):
"""Test registration with missing required fields"""
response = client.post('/api/auth/register', json={
'email': 'newuser@example.com'
})
assert response.status_code == 400
data = response.get_json()
assert 'error' in data
@pytest.mark.auth
def test_register_duplicate_email(self, client, regular_user):
"""Test registration with duplicate email"""
response = client.post('/api/auth/register', json={
'email': regular_user.email,
'password': 'password123'
})
assert response.status_code == 400
data = response.get_json()
assert 'already exists' in data['error'].lower()
@pytest.mark.auth
def test_login_success(self, client, regular_user):
"""Test successful login"""
response = client.post('/api/auth/login', json={
'email': regular_user.email,
'password': 'password123'
})
assert response.status_code == 200
data = response.get_json()
assert 'access_token' in data
assert 'refresh_token' in data
assert data['user']['email'] == regular_user.email
@pytest.mark.auth
@pytest.mark.parametrize("email,password,expected_status", [
("wrong@example.com", "password123", 401),
("user@example.com", "wrongpassword", 401),
(None, "password123", 400),
("user@example.com", None, 400),
])
def test_login_validation(self, client, regular_user, email, password, expected_status):
"""Test login with various invalid inputs"""
login_data = {}
if email is not None:
login_data['email'] = email
if password is not None:
login_data['password'] = password
response = client.post('/api/auth/login', json=login_data)
assert response.status_code == expected_status
@pytest.mark.auth
def test_login_inactive_user(self, client, inactive_user):
"""Test login with inactive user"""
response = client.post('/api/auth/login', json={
'email': inactive_user.email,
'password': 'password123'
})
assert response.status_code == 401
data = response.get_json()
assert 'inactive' in data['error'].lower()
@pytest.mark.auth
def test_get_current_user(self, client, auth_headers, regular_user):
"""Test getting current user"""
response = client.get('/api/users/me', headers=auth_headers)
assert response.status_code == 200
data = response.get_json()
assert data['email'] == regular_user.email
@pytest.mark.auth
def test_get_current_user_unauthorized(self, client):
"""Test getting current user without authentication"""
response = client.get('/api/users/me')
assert response.status_code == 401
class TestProductRoutes:
"""Test product routes"""
@pytest.mark.product
def test_get_products(self, client, products):
"""Test getting all products"""
response = client.get('/api/products')
assert response.status_code == 200
data = response.get_json()
assert len(data) == 5
@pytest.mark.product
def test_get_products_empty(self, client):
"""Test getting products when none exist"""
response = client.get('/api/products')
assert response.status_code == 200
data = response.get_json()
assert len(data) == 0
@pytest.mark.product
def test_get_single_product(self, client, product):
"""Test getting a single product"""
response = client.get(f'/api/products/{product.id}')
assert response.status_code == 200
data = response.get_json()
assert data['id'] == product.id
assert data['name'] == product.name
@pytest.mark.product
def test_get_product_not_found(self, client):
"""Test getting non-existent product"""
response = client.get('/api/products/999')
assert response.status_code == 404
@pytest.mark.product
def test_create_product_admin(self, client, admin_headers):
"""Test creating product as admin"""
response = client.post('/api/products', headers=admin_headers, json={
'name': 'New Product',
'description': 'A new product',
'price': 29.99,
'stock': 10
})
assert response.status_code == 201
data = response.get_json()
assert data['name'] == 'New Product'
assert data['price'] == 29.99
@pytest.mark.product
def test_create_product_regular_user(self, client, auth_headers):
"""Test creating product as regular user (should fail)"""
response = client.post('/api/products', headers=auth_headers, json={
'name': 'New Product',
'price': 29.99
})
assert response.status_code == 403
data = response.get_json()
assert 'admin' in data['error'].lower()
@pytest.mark.product
def test_create_product_unauthorized(self, client):
"""Test creating product without authentication"""
response = client.post('/api/products', json={
'name': 'New Product',
'price': 29.99
})
assert response.status_code == 401
@pytest.mark.product
def test_create_product_validation_error(self, client, admin_headers):
"""Test creating product with invalid data"""
response = client.post('/api/products', headers=admin_headers, json={
'name': 'New Product',
'price': -10.99
})
assert response.status_code == 400
data = response.get_json()
assert 'Validation error' in data['error']
@pytest.mark.product
def test_create_product_missing_required_fields(self, client, admin_headers):
"""Test creating product with missing required fields"""
response = client.post('/api/products', headers=admin_headers, json={
'description': 'Missing name and price'
})
assert response.status_code == 400
data = response.get_json()
assert 'Validation error' in data['error']
@pytest.mark.product
def test_create_product_minimal_data(self, client, admin_headers):
"""Test creating product with minimal valid data"""
response = client.post('/api/products', headers=admin_headers, json={
'name': 'Minimal Product',
'price': 19.99
})
assert response.status_code == 201
data = response.get_json()
assert data['name'] == 'Minimal Product'
assert data['stock'] == 0 # Default value
@pytest.mark.product
def test_update_product_admin(self, client, admin_headers, product):
"""Test updating product as admin"""
response = client.put(f'/api/products/{product.id}', headers=admin_headers, json={
'name': 'Updated Product',
'price': 39.99
})
assert response.status_code == 200
data = response.get_json()
assert data['name'] == 'Updated Product'
assert data['price'] == 39.99
@pytest.mark.product
def test_delete_product_admin(self, client, admin_headers, product):
"""Test deleting product as admin"""
response = client.delete(f'/api/products/{product.id}', headers=admin_headers)
assert response.status_code == 200
# Verify product is deleted
response = client.get(f'/api/products/{product.id}')
assert response.status_code == 404
class TestOrderRoutes:
"""Test order routes"""
@pytest.mark.order
def test_get_orders(self, client, auth_headers, order):
"""Test getting orders for current user"""
response = client.get('/api/orders', headers=auth_headers)
assert response.status_code == 200
data = response.get_json()
assert len(data) >= 1
@pytest.mark.order
def test_get_orders_unauthorized(self, client):
"""Test getting orders without authentication"""
response = client.get('/api/orders')
assert response.status_code == 401
@pytest.mark.order
def test_create_order(self, client, auth_headers, products):
"""Test creating an order"""
response = client.post('/api/orders', headers=auth_headers, json={
'items': [
{'product_id': products[0].id, 'quantity': 2},
{'product_id': products[1].id, 'quantity': 1}
],
'shipping_address': '123 Test St'
})
assert response.status_code == 201
data = response.get_json()
assert 'id' in data
assert len(data['items']) == 2
@pytest.mark.order
def test_create_order_insufficient_stock(self, client, auth_headers, db_session, products):
"""Test creating order with insufficient stock"""
# Set stock to 0
products[0].stock = 0
db_session.commit()
response = client.post('/api/orders', headers=auth_headers, json={
'items': [
{'product_id': products[0].id, 'quantity': 2}
]
})
assert response.status_code == 400
data = response.get_json()
assert 'insufficient' in data['error'].lower()
@pytest.mark.order
def test_get_single_order(self, client, auth_headers, order):
"""Test getting a single order"""
response = client.get(f'/api/orders/{order.id}', headers=auth_headers)
print('test_get_single_order', response.get_json())
assert response.status_code == 200
data = response.get_json()
assert data['id'] == order.id
@pytest.mark.order
def test_get_other_users_order(self, client, admin_headers, regular_user, products):
"""Test admin accessing another user's order"""
# Create an order for regular_user
client.post('/api/auth/login', json={
'email': regular_user.email,
'password': 'password123'
})
# Admin should be able to access any order
response = client.get(f'/api/orders/1', headers=admin_headers)
# This test assumes order exists, adjust as needed
pass

View file

@ -0,0 +1,277 @@
"""Test Pydantic schemas"""
import pytest
from pydantic import ValidationError
from decimal import Decimal
from app.schemas import ProductCreateRequest, ProductResponse
class TestProductCreateRequestSchema:
"""Test ProductCreateRequest schema"""
@pytest.mark.unit
def test_valid_product_request(self):
"""Test valid product creation request"""
data = {
'name': 'Handcrafted Wooden Bowl',
'description': 'A beautiful handcrafted bowl',
'price': 45.99,
'stock': 10,
'image_url': 'https://example.com/bowl.jpg'
}
product = ProductCreateRequest(**data)
assert product.name == data['name']
assert product.description == data['description']
assert product.price == Decimal('45.99')
assert product.stock == 10
assert product.image_url == data['image_url']
@pytest.mark.unit
def test_minimal_valid_request(self):
"""Test minimal valid request (only required fields)"""
data = {
'name': 'Simple Product',
'price': 19.99
}
product = ProductCreateRequest(**data)
assert product.name == 'Simple Product'
assert product.price == Decimal('19.99')
assert product.stock == 0
assert product.description is None
assert product.image_url is None
@pytest.mark.unit
def test_missing_name(self):
"""Test request with missing name"""
data = {
'price': 19.99
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['loc'] == ('name',) for error in errors)
@pytest.mark.unit
def test_missing_price(self):
"""Test request with missing price"""
data = {
'name': 'Test Product'
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['loc'] == ('price',) for error in errors)
@pytest.mark.unit
def test_invalid_price_negative(self):
"""Test request with negative price"""
data = {
'name': 'Test Product',
'price': -10.99
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['type'] == 'greater_than' for error in errors)
@pytest.mark.unit
def test_invalid_price_zero(self):
"""Test request with zero price"""
data = {
'name': 'Test Product',
'price': 0.0
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['type'] == 'greater_than' for error in errors)
@pytest.mark.unit
def test_invalid_price_too_many_decimals(self):
"""Test request with too many decimal places"""
data = {
'name': 'Test Product',
'price': 10.999
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any('decimal places' in str(error).lower() for error in errors)
@pytest.mark.unit
def test_invalid_stock_negative(self):
"""Test request with negative stock"""
data = {
'name': 'Test Product',
'price': 19.99,
'stock': -5
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['type'] == 'greater_than_equal' for error in errors)
@pytest.mark.unit
def test_name_too_long(self):
"""Test request with name exceeding max length"""
data = {
'name': 'A' * 201, # Exceeds 200 character limit
'price': 19.99
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['loc'] == ('name',) for error in errors)
@pytest.mark.unit
def test_image_url_too_long(self):
"""Test request with image_url exceeding max length"""
data = {
'name': 'Test Product',
'price': 19.99,
'image_url': 'A' * 501 # Exceeds 500 character limit
}
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['loc'] == ('image_url',) for error in errors)
@pytest.mark.unit
def test_price_string_conversion(self):
"""Test price string to Decimal conversion"""
data = {
'name': 'Test Product',
'price': '29.99'
}
product = ProductCreateRequest(**data)
assert product.price == Decimal('29.99')
@pytest.mark.unit
def test_stock_string_conversion(self):
"""Test stock string to int conversion"""
data = {
'name': 'Test Product',
'price': 19.99,
'stock': '10'
}
product = ProductCreateRequest(**data)
assert product.stock == 10
assert isinstance(product.stock, int)
class TestProductResponseSchema:
"""Test ProductResponse schema"""
@pytest.mark.unit
def test_valid_product_response(self):
"""Test valid product response"""
data = {
'id': 1,
'name': 'Test Product',
'description': 'A test product',
'price': 45.99,
'stock': 10,
'image_url': 'https://example.com/product.jpg',
'is_active': True,
'created_at': '2024-01-15T10:30:00',
'updated_at': '2024-01-15T10:30:00'
}
product = ProductResponse(**data)
assert product.id == 1
assert product.name == 'Test Product'
assert product.price == 45.99
assert product.stock == 10
assert product.is_active is True
@pytest.mark.unit
def test_product_response_with_none_fields(self):
"""Test product response with optional None fields"""
data = {
'id': 1,
'name': 'Test Product',
'price': 19.99,
'stock': 0,
'is_active': True
}
product = ProductResponse(**data)
assert product.description is None
assert product.image_url is None
assert product.created_at is None
assert product.updated_at is None
@pytest.mark.unit
def test_model_validate_from_sqlalchemy(self, db_session):
"""Test validating SQLAlchemy model to Pydantic schema"""
from app.models import Product
db_product = Product(
name='Test Product',
description='A test product',
price=Decimal('45.99'),
stock=10
)
db_session.add(db_product)
db_session.commit()
# Validate using model_validate (for SQLAlchemy models)
response = ProductResponse.model_validate(db_product)
assert response.name == 'Test Product'
assert response.price == 45.99
assert response.stock == 10
@pytest.mark.unit
def test_model_dump(self):
"""Test model_dump method"""
data = {
'id': 1,
'name': 'Test Product',
'price': 19.99,
'stock': 5,
'is_active': True
}
product = ProductResponse(**data)
dumped = product.model_dump()
assert isinstance(dumped, dict)
assert dumped['id'] == 1
assert dumped['name'] == 'Test Product'
assert dumped['price'] == 19.99
@pytest.mark.unit
def test_model_dump_json(self):
"""Test model_dump_json method"""
data = {
'id': 1,
'name': 'Test Product',
'price': 19.99,
'stock': 5,
'is_active': True
}
product = ProductResponse(**data)
json_str = product.model_dump_json()
assert isinstance(json_str, str)
assert 'Test Product' in json_str

491
docs/testing_guide.md Normal file
View file

@ -0,0 +1,491 @@
# Testing Guide
## Overview
This document provides a comprehensive guide to testing the Flask application using pytest. The testing infrastructure includes unit tests, integration tests, CI/CD pipelines, and pre-commit hooks.
## Table of Contents
1. [Installation](#installation)
2. [Running Tests](#running-tests)
3. [Test Structure](#test-structure)
4. [Writing Tests](#writing-tests)
5. [Fixtures](#fixtures)
6. [Coverage](#coverage)
7. [CI/CD Pipeline](#cicd-pipeline)
8. [Pre-commit Hooks](#pre-commit-hooks)
9. [Best Practices](#best-practices)
## Installation
### Install Test Dependencies
```bash
cd backend
pip install -r requirements/base.txt
```
The base requirements include:
- `pytest==7.4.3` - Testing framework
- `pytest-flask==1.3.0` - Flask integration
- `pytest-cov==4.1.0` - Coverage reporting
- `pytest-mock==3.12.0` - Mocking utilities
- `factory-boy==3.3.0` - Test data factories
- `faker==20.1.0` - Fake data generation
## Running Tests
### Run All Tests
```bash
cd backend
pytest
```
### Run with Verbose Output
```bash
pytest -v
```
### Run with Coverage Report
```bash
pytest --cov=app --cov-report=html --cov-report=term
```
### Run Specific Test Files
```bash
# Run all model tests
pytest tests/test_models.py
# Run all route tests
pytest tests/test_routes.py
# Run all schema tests
pytest tests/test_schemas.py
```
### Run by Test Name
```bash
pytest -k "test_user_creation"
pytest -k "test_login"
```
### Run by Markers
```bash
# Run only unit tests
pytest -m unit
# Run only integration tests
pytest -m integration
# Run only authentication tests
pytest -m auth
# Run only product tests
pytest -m product
# Run only order tests
pytest -m order
```
### Run Tests in Parallel (faster)
Install pytest-xdist:
```bash
pip install pytest-xdist
pytest -n auto # Use all available CPUs
```
## Test Structure
```
backend/
├── tests/
│ ├── __init__.py
│ ├── conftest.py # Global fixtures and configuration
│ ├── test_models.py # Model tests
│ ├── test_routes.py # Route/API tests
│ └── test_schemas.py # Pydantic schema tests
├── pytest.ini # Pytest configuration
├── .coveragerc # Coverage configuration
└── app/
├── __init__.py
├── models/ # Database models
├── routes/ # API routes
├── schemas/ # Pydantic schemas
└── ...
```
## Writing Tests
### Test File Structure
```python
import pytest
from app.models import User
from app import db
class TestUserModel:
"""Test User model"""
@pytest.mark.unit
def test_user_creation(self, db_session):
"""Test creating a user"""
user = User(
email='test@example.com',
username='testuser'
)
user.set_password('password123')
db_session.add(user)
db_session.commit()
assert user.id is not None
assert user.email == 'test@example.com'
```
### Test API Routes
```python
def test_get_products(client, products):
"""Test getting all products"""
response = client.get('/api/products')
assert response.status_code == 200
data = response.get_json()
assert len(data) == 5
def test_create_product(client, admin_headers):
"""Test creating a product"""
response = client.post('/api/products',
headers=admin_headers,
json={
'name': 'New Product',
'price': 29.99
})
assert response.status_code == 201
data = response.get_json()
assert data['name'] == 'New Product'
```
### Parameterized Tests
```python
@pytest.mark.parametrize("email,password,expected_status", [
("user@example.com", "correct123", 200),
("wrong@email.com", "correct123", 401),
("user@example.com", "wrongpass", 401),
])
def test_login_validation(client, email, password, expected_status):
"""Test login with various inputs"""
response = client.post('/api/auth/login', json={
'email': email,
'password': password
})
assert response.status_code == expected_status
```
## Fixtures
### Available Fixtures
#### Application Fixtures
- **`app`**: Flask application instance with test configuration
- **`client`**: Test client for making HTTP requests
- **`runner`**: CLI runner for testing Flask CLI commands
- **`db_session`**: Database session for database operations
#### User Fixtures
- **`admin_user`**: Creates an admin user
- **`regular_user`**: Creates a regular user
- **`inactive_user`**: Creates an inactive user
#### Product Fixtures
- **`product`**: Creates a single product
- **`products`**: Creates 5 products
#### Authentication Fixtures
- **`auth_headers`**: JWT headers for regular user
- **`admin_headers`**: JWT headers for admin user
#### Order Fixtures
- **`order`**: Creates an order with items
### Creating Custom Fixtures
```python
# In conftest.py or test file
@pytest.fixture
def custom_product(db_session):
"""Create a custom product"""
product = Product(
name='Custom Product',
price=99.99,
stock=50
)
db_session.add(product)
db_session.commit()
return product
# Use in tests
def test_custom_fixture(custom_product):
assert custom_product.name == 'Custom Product'
```
## Coverage
### Coverage Configuration
Coverage is configured in `.coveragerc`:
```ini
[run]
source = app
omit =
*/tests/*
*/migrations/*
*/__pycache__/*
[report]
exclude_lines =
pragma: no cover
def __repr__
raise NotImplementedError
```
### Coverage Thresholds
The CI/CD pipeline enforces 80% minimum code coverage.
### Generate Coverage Report
```bash
# Terminal report
pytest --cov=app --cov-report=term
# HTML report
pytest --cov=app --cov-report=html
open htmlcov/index.html # Mac
xdg-open htmlcov/index.html # Linux
```
### Coverage Report Example
```
Name Stmts Miss Cover Missing
----------------------------------------------
app/__init__.py 10 2 80% 15-16
app/models/user.py 45 5 89% 23, 45
app/routes/api.py 120 20 83% 78-85
----------------------------------------------
TOTAL 175 27 85%
```
## CI/CD Pipeline
### GitHub Actions Workflow
The backend has automated testing via GitHub Actions:
**File**: `.github/workflows/backend-tests.yml`
### Pipeline Stages
1. **Test Matrix**: Runs tests on Python 3.10, 3.11, and 3.12
2. **Services**: Sets up PostgreSQL and Redis
3. **Linting**: Runs flake8 for code quality
4. **Testing**: Executes pytest with coverage
5. **Coverage Upload**: Sends coverage to Codecov
6. **Security Scan**: Runs bandit and safety
### Triggering the Pipeline
The pipeline runs automatically on:
- Push to `main` or `develop` branches
- Pull requests to `main` or `develop` branches
- Changes to `backend/**` or workflow files
### Viewing Results
1. Go to the Actions tab in your GitHub repository
2. Click on the latest workflow run
3. View test results, coverage, and artifacts
## Pre-commit Hooks
### Setup Pre-commit Hooks
```bash
# Install pre-commit
pip install pre-commit
# Install hooks
pre-commit install
# Run hooks manually
pre-commit run --all-files
```
### Available Hooks
The `.pre-commit-config.yaml` includes:
1. **Black**: Code formatting
2. **isort**: Import sorting
3. **flake8**: Linting
4. **pytest**: Run tests before committing
5. **mypy**: Type checking
6. **bandit**: Security checks
### Hook Behavior
Hooks run automatically on:
- `git commit`
- Can be skipped with `git commit --no-verify`
## Best Practices
### ✅ DO
1. **Use descriptive test names**
```python
def test_user_creation_with_valid_data(): # Good
def test_user(): # Bad
```
2. **Test both success and failure cases**
```python
def test_login_success(): ...
def test_login_invalid_credentials(): ...
def test_login_missing_fields(): ...
```
3. **Use fixtures for common setup**
```python
def test_something(client, admin_user, products): ...
```
4. **Mock external services**
```python
def test_external_api(mocker):
mock_response = {'data': 'mocked'}
mocker.patch('requests.get', return_value=mock_response)
```
5. **Keep tests independent**
- Each test should be able to run alone
- Don't rely on test execution order
6. **Use markers appropriately**
```python
@pytest.mark.slow
def test_expensive_operation(): ...
```
### ❌ DON'T
1. **Don't share state between tests**
```python
# Bad - shared state
global_user = User(...)
# Good - use fixtures
@pytest.fixture
def user(): return User(...)
```
2. **Don't hardcode sensitive data**
```python
# Bad
password = 'real_password_123'
# Good
password = fake.password()
```
3. **Don't use production database**
- Always use test database (SQLite)
- Fixtures automatically create isolated databases
4. **Don't skip error cases**
```python
# Bad - only tests success
def test_create_product(): ...
# Good - tests both
def test_create_product_success(): ...
def test_create_product_validation_error(): ...
```
5. **Don't ignore slow tests in CI**
- Mark slow tests with `@pytest.mark.slow`
- Run them separately if needed
## Test Coverage Requirements
| Module | Line Coverage | Branch Coverage |
|--------|--------------|-----------------|
| routes.py | >90% | >85% |
| models.py | >85% | >80% |
| schemas.py | >90% | >85% |
| services/ | >80% | >75% |
| utils/ | >70% | >65% |
## Troubleshooting
### Tests Fail with Database Errors
```bash
# Clean up test databases
rm -f backend/*.db
```
### Coverage Not Showing
```bash
# Install coverage separately
pip install coverage
# Clean previous coverage data
coverage erase
# Run tests again
pytest --cov=app
```
### Import Errors
```bash
# Ensure you're in the backend directory
cd backend
# Install in development mode
pip install -e .
```
### Slow Tests
```bash
# Run only specific tests
pytest tests/test_routes.py::TestProductRoutes::test_get_products
# Run in parallel
pytest -n auto
```
## Additional Resources
- [Pytest Documentation](https://docs.pytest.org/)
- [Pytest-Flask Documentation](https://pytest-flask.readthedocs.io/)
- [Pydantic Documentation](https://docs.pydantic.dev/)
- [Flask Testing Documentation](https://flask.palletsprojects.com/en/latest/testing/)

View file

@ -119,6 +119,18 @@ def create_product():
- **ALWAYS** import `db` from `app` - **ALWAYS** import `db` from `app`
- Use `db.session.add()` and `db.session.commit()` for transactions - Use `db.session.add()` and `db.session.commit()` for transactions
- Use `db.session.flush()` when you need the ID before commit - Use `db.session.flush()` when you need the ID before commit
- **ALWAYS** use `db.session.get(Model, id)` instead of `Model.query.get(id)` (SQLAlchemy 2.0)
- Use `Model.query.get_or_404(id)` for 404 handling when appropriate
```python
# ✅ CORRECT - SQLAlchemy 2.0 syntax
from app import db
from app.models import User
user = db.session.get(User, user_id)
# ❌ WRONG - Legacy syntax (deprecated)
user = User.query.get(user_id)
```
### Error Handling ### Error Handling
- Handle common errors (404, 400, 401, 403, 500) - Handle common errors (404, 400, 401, 403, 500)
@ -189,11 +201,308 @@ def create_product_service(data):
- Use pytest framework - Use pytest framework
- Place tests in `backend/tests/` - Place tests in `backend/tests/`
- Use fixtures for common setup - Use fixtures for common setup
- Organize tests by functionality: `test_models.py`, `test_routes.py`, `test_schemas.py`
### Test Naming Conventions
- Test files must start with `test_`: `test_products.py`, `test_users.py`
- Test classes must start with `Test`: `TestProductModel`, `TestAuthRoutes`
- Test functions must start with `test_`: `test_create_product`, `test_login_success`
- Use descriptive names: `test_create_product_with_valid_data` (not `test_product`)
### Writing Tests
#### Basic Test Structure
```python
# ✅ CORRECT
import pytest
from app import db
from app.models import Product
class TestProductModel:
"""Test Product model"""
@pytest.mark.unit
def test_product_creation(self, db_session):
"""Test creating a product with valid data"""
product = Product(
name='Test Product',
price=99.99
)
db_session.add(product)
db_session.commit()
assert product.id is not None
assert product.name == 'Test Product'
```
#### Testing API Routes
```python
# ✅ CORRECT
def test_create_product(client, admin_headers):
"""Test creating a product as admin"""
response = client.post('/api/products',
headers=admin_headers,
json={
'name': 'New Product',
'price': 29.99
})
assert response.status_code == 201
data = response.get_json()
assert data['name'] == 'New Product'
assert 'password' not in data
```
#### Using Fixtures
```python
# ✅ CORRECT - Use available fixtures
def test_get_products(client, products):
"""Test getting all products"""
response = client.get('/api/products')
assert response.status_code == 200
data = response.get_json()
assert len(data) == 5
# ❌ WRONG - Don't create fixtures manually in tests
def test_get_products_wrong(client, db_session):
products = []
for _ in range(5):
p = Product(name='Test', price=10)
db_session.add(p)
products.append(p)
db_session.commit()
# ... use fixtures instead!
```
#### Testing Both Success and Failure Cases
```python
# ✅ CORRECT - Test both scenarios
def test_create_product_success(client, admin_headers):
"""Test creating product successfully"""
response = client.post('/api/products',
headers=admin_headers,
json={'name': 'Test', 'price': 10})
assert response.status_code == 201
def test_create_product_unauthorized(client):
"""Test creating product without authentication"""
response = client.post('/api/products',
json={'name': 'Test', 'price': 10})
assert response.status_code == 401
def test_create_product_validation_error(client, admin_headers):
"""Test creating product with invalid data"""
response = client.post('/api/products',
headers=admin_headers,
json={'name': 'Test', 'price': -10})
assert response.status_code == 400
```
#### Parameterized Tests
```python
# ✅ CORRECT - Use parameterization for similar tests
@pytest.mark.parametrize("email,password,expected_status", [
("user@example.com", "correct123", 200),
("wrong@email.com", "correct123", 401),
("user@example.com", "wrongpass", 401),
])
def test_login_validation(client, email, password, expected_status):
"""Test login with various invalid inputs"""
response = client.post('/api/auth/login', json={
'email': email,
'password': password
})
assert response.status_code == expected_status
```
### Test Markers
Use appropriate markers for categorizing tests:
```python
# ✅ CORRECT
@pytest.mark.unit
def test_user_creation(self, db_session):
"""Unit test - no HTTP, no external services"""
pass
@pytest.mark.integration
def test_user_workflow(self, client):
"""Integration test - full request/response cycle"""
pass
@pytest.mark.auth
def test_login(self, client):
"""Authentication-related test"""
pass
@pytest.mark.product
def test_get_products(self, client):
"""Product-related test"""
pass
```
### Database in Tests ### Database in Tests
- Use in-memory SQLite for tests - Use in-memory SQLite for tests
- Clean up database between tests - Clean up database between tests
- Use `pytest.fixture` for database setup - Use `pytest.fixture` for database setup
- **NEVER** use production database in tests
- **NEVER** share state between tests
```python
# ✅ CORRECT - Use db_session fixture
def test_something(db_session):
user = User(email='test@example.com')
db_session.add(user)
db_session.commit()
# ❌ WRONG - Don't access db directly
def test_something_wrong():
from app import db
user = User(email='test@example.com')
db.session.add(user)
db.session.commit()
```
### Available Fixtures
Use these fixtures from `tests/conftest.py`:
- **`app`**: Flask application instance with test configuration
- **`client`**: Test client for making HTTP requests
- **`runner`**: CLI runner for Flask commands
- **`db_session`**: Database session for database operations
- **`admin_user`**: Pre-created admin user
- **`regular_user`**: Pre-created regular user
- **`inactive_user`**: Pre-created inactive user
- **`product`**: Single product
- **`products`**: Multiple products (5 items)
- **`auth_headers`**: JWT headers for regular user
- **`admin_headers`**: JWT headers for admin user
- **`order`**: Pre-created order with items
### Creating Custom Fixtures
```python
# In tests/conftest.py or test file
@pytest.fixture
def custom_resource(db_session):
"""Create a custom test resource"""
resource = CustomModel(
name='Test Resource',
value=100
)
db_session.add(resource)
db_session.commit()
return resource
# Use in tests
def test_custom_fixture(custom_resource):
assert custom_resource.name == 'Test Resource'
```
### Running Tests
```bash
# Run all tests
make test-backend
# Run with coverage
make test-backend-cov
# Run with verbose output
make test-backend-verbose
# Run specific test file
make test-backend-specific TEST=test_models.py
# Run by marker
make test-backend-marker MARKER=auth
# Run only failed tests
make test-backend-failed
# Run in parallel (faster)
make test-backend-parallel
```
### Test Coverage Requirements
- **Minimum 80%** code coverage required
- **Critical paths** (auth, payments, data modification) must have >90% coverage
- All new features must include tests
```python
# ✅ CORRECT - Comprehensive test coverage
def test_product_crud(self, client, admin_headers):
"""Test complete CRUD operations"""
# Create
response = client.post('/api/products',
headers=admin_headers,
json={'name': 'Test', 'price': 10})
assert response.status_code == 201
product_id = response.get_json()['id']
# Read
response = client.get(f'/api/products/{product_id}')
assert response.status_code == 200
# Update
response = client.put(f'/api/products/{product_id}',
headers=admin_headers,
json={'name': 'Updated', 'price': 20})
assert response.status_code == 200
# Delete
response = client.delete(f'/api/products/{product_id}',
headers=admin_headers)
assert response.status_code == 200
```
### Mocking External Services
```python
# ✅ CORRECT - Mock external dependencies
def test_external_api_call(client, mocker):
"""Test endpoint that calls external API"""
mock_response = {'data': 'mocked data'}
# Mock requests.get
mock_get = mocker.patch('requests.get')
mock_get.return_value.json.return_value = mock_response
mock_get.return_value.status_code = 200
response = client.get('/api/external-data')
assert response.status_code == 200
assert response.get_json() == mock_response
mock_get.assert_called_once()
```
### Test DOs and DON'Ts
✅ **DO:**
- Use descriptive test names
- Test both success and failure cases
- Use fixtures for common setup
- Mock external services
- Keep tests independent
- Use markers appropriately
- Test edge cases and boundary conditions
❌ **DON'T:**
- Share state between tests
- Hardcode sensitive data (use faker)
- Use production database
- Skip error case testing
- Write tests after deployment
- Ignore slow tests in CI
- Use complex setup in test methods (use fixtures instead)
## Security Rules ## Security Rules