david/kanban-app
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

update actions config, fix lint errors

Browse source
This commit is contained in:
david 2026-02-24 19:19:15 +03:00
parent a9b0fd34a2
commit 1c35df931a
26 changed files with 732 additions and 691 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.github/workflows/backend-tests.yml vendored
View file

@ -14,7 +14,7 @@ on:
jobs:
test:
runs-on: ubuntu-latest
runs-on: [docker]
strategy:
matrix:
@ -32,8 +32,6 @@ jobs:
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
redis:
image: redis:7-alpine
@ -106,7 +104,7 @@ jobs:
coverage report --fail-under=80
security-scan:
runs-on: ubuntu-latest
runs-on: [docker]
steps:
- name: Checkout code
uses: actions/checkout@v4

96
.github/workflows/cd.yml vendored
View file

@ -1,56 +1,56 @@
name: CD
# name: CD
on:
push:
branches: [ main ]
workflow_dispatch:
# on:
# push:
# branches: [ main ]
# workflow_dispatch:
jobs:
deploy:
runs-on: [docker]
# jobs:
# deploy:
# runs-on: [docker]
steps:
- uses: actions/checkout@v3
# steps:
# - uses: actions/checkout@v3
- name: Configure AWS credentials
uses: aws-actions/configure-aws-credentials@v2
with:
aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
aws-region: ${{ secrets.AWS_REGION }}
# - name: Configure AWS credentials
# uses: aws-actions/configure-aws-credentials@v2
# with:
# aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
# aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
# aws-region: ${{ secrets.AWS_REGION }}
- name: Login to Amazon ECR
id: login-ecr
uses: aws-actions/amazon-ecr-login@v1
# - name: Login to Amazon ECR
# id: login-ecr
# uses: aws-actions/amazon-ecr-login@v1
- name: Build and push backend
env:
ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
ECR_REPOSITORY: crafting-shop-backend
IMAGE_TAG: ${{ github.sha }}
run: |
cd backend
docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
docker tag $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG $ECR_REGISTRY/$ECR_REPOSITORY:latest
docker push $ECR_REGISTRY/$ECR_REPOSITORY:latest
# - name: Build and push backend
# env:
# ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
# ECR_REPOSITORY: crafting-shop-backend
# IMAGE_TAG: ${{ github.sha }}
# run: |
# cd backend
# docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
# docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
# docker tag $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG $ECR_REGISTRY/$ECR_REPOSITORY:latest
# docker push $ECR_REGISTRY/$ECR_REPOSITORY:latest
- name: Build and push frontend
env:
ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
ECR_REPOSITORY: crafting-shop-frontend
IMAGE_TAG: ${{ github.sha }}
run: |
cd frontend
docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
docker tag $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG $ECR_REGISTRY/$ECR_REPOSITORY:latest
docker push $ECR_REGISTRY/$ECR_REPOSITORY:latest
# - name: Build and push frontend
# env:
# ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
# ECR_REPOSITORY: crafting-shop-frontend
# IMAGE_TAG: ${{ github.sha }}
# run: |
# cd frontend
# docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
# docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
# docker tag $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG $ECR_REGISTRY/$ECR_REPOSITORY:latest
# docker push $ECR_REGISTRY/$ECR_REPOSITORY:latest
- name: Deploy to ECS
uses: aws-actions/amazon-ecs-deploy-task-definition@v1
with:
task-definition: crafting-shop-task
service: crafting-shop-service
cluster: crafting-shop-cluster
wait-for-service-stability: true
# - name: Deploy to ECS
# uses: aws-actions/amazon-ecs-deploy-task-definition@v1
# with:
# task-definition: crafting-shop-task
# service: crafting-shop-service
# cluster: crafting-shop-cluster
# wait-for-service-stability: true

33
.github/workflows/ci.yml vendored
View file

@ -22,25 +22,26 @@ jobs:
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
container:
image: nikolaik/python-nodejs:python3.12-nodejs24-alpine
steps:
- uses: actions/checkout@v3
- uses: actions/checkout@v6
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
run: |
python --version
- name: Cache pip packages
uses: actions/cache@v3
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
# This path is on the runner, but we mounted it to the container
path: /tmp/pip-cache
key: ${{ runner.os }}-pip-${{ hashFiles('backend/requirements/dev.txt') }}
restore-keys: |
${{ runner.os }}-pip-
- name: Install dependencies
run: |
cd backend
@ -50,12 +51,11 @@ jobs:
- name: Lint with flake8
run: |
cd backend
flake8 app tests --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 app tests --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
flake8 app tests --count --max-complexity=10 --max-line-length=127 --statistics --show-source
- name: Run tests
env:
DATABASE_URL: postgresql://test:test@localhost:5432/test_db
DATABASE_URL: postgresql://test:test@postgres:5432/test_db
SECRET_KEY: test-secret-key
JWT_SECRET_KEY: test-jwt-secret
FLASK_ENV: testing
@ -63,13 +63,6 @@ jobs:
cd backend
pytest --cov=app --cov-report=xml --cov-report=term
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
files: ./backend/coverage.xml
flags: backend
name: backend-coverage
frontend-test:
runs-on: [docker]

18
Makefile
View file

@ -126,7 +126,23 @@ lint-frontend: ## Lint frontend only
format: ## Format code
@echo "Formatting backend..."
cd backend && . venv/bin/activate && black app tests && isort app tests
cd backend && . venv/bin/activate && black app tests
cd backend && . venv/bin/activate && isort app tests
@echo "Formatting frontend..."
cd frontend && npx prettier --write "src/**/*.{js,jsx,ts,tsx,css}"
format-backend: ## Format backend code only
@echo "Formatting backend with black..."
cd backend && . venv/bin/activate && black app tests
@echo "Sorting imports with isort..."
cd backend && . venv/bin/activate && isort app tests
format-backend-check: ## Check if backend code needs formatting
@echo "Checking backend formatting..."
cd backend && . venv/bin/activate && black --check app tests
cd backend && . venv/bin/activate && isort --check-only app tests
format-frontend: ## Format frontend code only
@echo "Formatting frontend..."
cd frontend && npx prettier --write "src/**/*.{js,jsx,ts,tsx,css}"

47
backend/app/__init__.py
View file

@ -1,11 +1,12 @@
import json
import os
from dotenv import load_dotenv
from flask import Flask, jsonify
from flask_cors import CORS
from flask_jwt_extended import JWTManager
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
import os
from dotenv import load_dotenv
from flask_sqlalchemy import SQLAlchemy
# Create extensions but don't initialize them yet
db = SQLAlchemy()
migrate = Migrate()
@ -13,54 +14,62 @@ jwt = JWTManager()
cors = CORS()
load_dotenv(override=True)
def create_app(config_name=None):
"""Application factory pattern"""
app = Flask(__name__)
# Load configuration
if config_name is None:
config_name = os.environ.get("FLASK_ENV", "development")
from app.config import config_by_name
app.config.from_object(config_by_name[config_name])
print('----------------------------------------------------------')
print(F'------------------ENVIRONMENT: {config_name}-------------------------------------')
print("----------------------------------------------------------")
print(
f"------------------ENVIRONMENT: {config_name}-------------------------------------"
)
# print(F'------------------CONFIG: {app.config}-------------------------------------')
# print(json.dumps(dict(app.config), indent=2, default=str))
print('----------------------------------------------------------')
print("----------------------------------------------------------")
# Initialize extensions with app
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
cors.init_app(app, resources={r"/api/*": {"origins": app.config.get("CORS_ORIGINS", "*")}})
cors.init_app(
app, resources={r"/api/*": {"origins": app.config.get("CORS_ORIGINS", "*")}}
)
# Initialize Celery
from app.celery import init_celery
init_celery(app)
# Import models (required for migrations)
from app.models import user, product, order
from app.models import order, product, user # noqa: F401
# Register blueprints
from app.routes import api_bp, health_bp
app.register_blueprint(api_bp, url_prefix="/api")
app.register_blueprint(health_bp)
# Global error handlers
@app.errorhandler(404)
def not_found(error):
print(f"404 Error: {error}")
return jsonify({"error": "Not found"}), 404
@app.errorhandler(500)
def internal_error(error):
print(f"500 Error: {error}")
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

32
backend/app/celery/__init__.py
View file

@ -9,10 +9,10 @@ from flask import Flask
def make_celery(app: Flask) -> Celery:
"""
Create and configure a Celery application with Flask context.
Args:
app: Flask application instance
Returns:
Configured Celery application instance.
"""
@ -20,36 +20,38 @@ def make_celery(app: Flask) -> Celery:
celery_app = Celery(
app.import_name,
broker=app.config["CELERY"]["broker_url"],
backend=app.config["CELERY"]["result_backend"]
backend=app.config["CELERY"]["result_backend"],
)
# Update configuration from Flask config
celery_app.conf.update(app.config["CELERY"])
# Set up Flask application context for tasks
# This ensures tasks have access to Flask extensions (db, etc.)
class ContextTask(celery_app.Task):
"""Celery task that runs within Flask application context."""
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery_app.Task = ContextTask
# Auto-discover tasks in the tasks module
celery_app.autodiscover_tasks(['app.celery.tasks'])
celery_app.autodiscover_tasks(["app.celery.tasks"])
# Configure Beat schedule
from .beat_schedule import configure_beat_schedule
configure_beat_schedule(celery_app)
# Import tasks to ensure they're registered
from .tasks import example_tasks
from .tasks import example_tasks # noqa: F401
print(f"✅ Celery configured with broker: {celery_app.conf.broker_url}")
print(f"✅ Celery configured with backend: {celery_app.conf.result_backend}")
print(f"✅ Beat schedule configured with {len(celery_app.conf.beat_schedule)} tasks")
return celery_app
@ -61,10 +63,10 @@ def init_celery(app: Flask) -> Celery:
"""
Initialize the global celery instance with Flask app.
This should be called in create_app() after Flask app is created.
Args:
app: Flask application instance
Returns:
Configured Celery application instance
"""

15
backend/app/celery/beat_schedule.py
View file

@ -4,7 +4,6 @@ This defines when scheduled tasks should run.
"""
from celery.schedules import crontab
# Celery Beat schedule configuration
beat_schedule = {
# Run every minute (for testing/demo)
@ -14,14 +13,12 @@ beat_schedule = {
"args": ("Celery Beat",),
"options": {"queue": "default"},
},
# Run daily at 9:00 AM
"send-daily-report": {
"task": "tasks.send_daily_report",
"schedule": crontab(hour=9, minute=0), # 9:00 AM daily
"options": {"queue": "reports"},
},
# Run every hour at minute 0
"update-product-stats-hourly": {
"task": "tasks.update_product_statistics",
@ -29,7 +26,6 @@ beat_schedule = {
"args": (None,), # Update all products
"options": {"queue": "stats"},
},
# Run every Monday at 8:00 AM
"weekly-maintenance": {
"task": "tasks.long_running_task",
@ -37,7 +33,6 @@ beat_schedule = {
"args": (5,), # 5 iterations
"options": {"queue": "maintenance"},
},
# Run every 5 minutes (for monitoring/heartbeat)
"heartbeat-check": {
"task": "tasks.print_hello",
@ -51,16 +46,16 @@ beat_schedule = {
def configure_beat_schedule(celery_app):
"""
Configure Celery Beat schedule on the Celery app.
Args:
celery_app: Celery application instance
"""
celery_app.conf.beat_schedule = beat_schedule
# Configure timezone
celery_app.conf.timezone = "UTC"
celery_app.conf.enable_utc = True
# Configure task routes for scheduled tasks
celery_app.conf.task_routes = {
"tasks.print_hello": {"queue": "default"},
@ -68,7 +63,7 @@ def configure_beat_schedule(celery_app):
"tasks.update_product_statistics": {"queue": "stats"},
"tasks.long_running_task": {"queue": "maintenance"},
}
# Configure queues
celery_app.conf.task_queues = {
"default": {
@ -96,4 +91,4 @@ def configure_beat_schedule(celery_app):
"exchange_type": "direct",
"routing_key": "monitoring",
},
}
}

8
backend/app/celery/tasks/__init__.py
View file

@ -4,12 +4,12 @@ Tasks are organized by domain/functionality.
"""
# Import all task modules here to ensure they're registered with Celery
from . import example_tasks
from . import example_tasks # noqa: F401
# Re-export tasks for easier imports
from .example_tasks import (
print_hello,
from .example_tasks import ( # noqa: F401
divide_numbers,
print_hello,
send_daily_report,
update_product_statistics,
)
)

89
backend/app/celery/tasks/example_tasks.py
View file

@ -2,11 +2,11 @@
Example Celery tasks for the Crafting Shop application.
These tasks demonstrate various Celery features and best practices.
"""
import time
import logging
import time
from datetime import datetime
from celery import shared_task
from celery.exceptions import MaxRetriesExceededError
# Get logger for this module
logger = logging.getLogger(__name__)
@ -16,10 +16,10 @@ logger = logging.getLogger(__name__)
def print_hello(self, name: str = "World") -> str:
"""
Simple task that prints a greeting.
Args:
name: Name to greet (default: "World")
Returns:
Greeting message
"""
@ -36,51 +36,47 @@ def print_hello(self, name: str = "World") -> str:
retry_backoff=True,
retry_backoff_max=60,
retry_jitter=True,
max_retries=3
max_retries=3,
)
def divide_numbers(self, x: float, y: float) -> float:
"""
Task that demonstrates error handling and retry logic.
Args:
x: Numerator
y: Denominator
Returns:
Result of division
Raises:
ZeroDivisionError: If y is zero (will trigger retry)
"""
logger.info(f"Dividing {x} by {y} (attempt {self.request.retries + 1})")
if y == 0:
logger.warning(f"Division by zero detected, retrying...")
logger.warning("Division by zero detected, retrying...")
raise ZeroDivisionError("Cannot divide by zero")
result = x / y
logger.info(f"Division result: {result}")
return result
@shared_task(
bind=True,
name="tasks.send_daily_report",
ignore_result=False
)
@shared_task(bind=True, name="tasks.send_daily_report", ignore_result=False)
def send_daily_report(self) -> dict:
"""
Simulates sending a daily report.
This task would typically send emails, generate reports, etc.
Returns:
Dictionary with report details
"""
logger.info("Starting daily report generation...")
# Simulate some work
time.sleep(2)
report_data = {
"date": datetime.now().isoformat(),
"task_id": self.request.id,
@ -90,46 +86,43 @@ def send_daily_report(self) -> dict:
"total_products": 150,
"total_orders": 42,
"total_users": 89,
"revenue": 12500.75
}
"revenue": 12500.75,
},
}
logger.info(f"Daily report generated: {report_data}")
print(f"📊 Daily Report Generated at {report_data['date']}")
return report_data
@shared_task(
bind=True,
name="tasks.update_product_statistics",
queue="stats",
priority=5
bind=True, name="tasks.update_product_statistics", queue="stats", priority=5
)
def update_product_statistics(self, product_id: int = None) -> dict:
"""
Simulates updating product statistics.
Demonstrates task routing to a specific queue.
Args:
product_id: Optional specific product ID to update.
If None, updates all products.
Returns:
Dictionary with update results
"""
logger.info(f"Updating product statistics for product_id={product_id}")
# Simulate database work
time.sleep(1)
if product_id is None:
# Update all products
result = {
"task": "update_all_product_stats",
"status": "completed",
"products_updated": 150,
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
else:
# Update specific product
@ -138,13 +131,9 @@ def update_product_statistics(self, product_id: int = None) -> dict:
"product_id": product_id,
"status": "completed",
"timestamp": datetime.now().isoformat(),
"new_stats": {
"views": 125,
"purchases": 15,
"rating": 4.5
}
"new_stats": {"views": 125, "purchases": 15, "rating": 4.5},
}
logger.info(f"Product statistics updated: {result}")
return result
@ -153,46 +142,46 @@ def update_product_statistics(self, product_id: int = None) -> dict:
bind=True,
name="tasks.long_running_task",
time_limit=300, # 5 minutes
soft_time_limit=240 # 4 minutes
soft_time_limit=240, # 4 minutes
)
def long_running_task(self, iterations: int = 10) -> dict:
"""
Simulates a long-running task with progress tracking.
Args:
iterations: Number of iterations to simulate
Returns:
Dictionary with results
"""
logger.info(f"Starting long-running task with {iterations} iterations")
results = []
for i in range(iterations):
# Check if task has been revoked
if self.is_aborted():
logger.warning("Task was aborted")
return {"status": "aborted", "completed_iterations": i}
# Simulate work
time.sleep(1)
# Update progress
progress = (i + 1) / iterations * 100
self.update_state(
state="PROGRESS",
meta={"current": i + 1, "total": iterations, "progress": progress}
meta={"current": i + 1, "total": iterations, "progress": progress},
)
results.append(f"iteration_{i + 1}")
logger.info(f"Completed iteration {i + 1}/{iterations}")
final_result = {
"status": "completed",
"iterations": iterations,
"results": results,
"completed_at": datetime.now().isoformat()
"completed_at": datetime.now().isoformat(),
}
logger.info(f"Long-running task completed: {final_result}")
return final_result
return final_result

20
backend/app/config.py
View file

@ -4,17 +4,20 @@ 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["JWT_SECRET_KEY"]
JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
CORS_ORIGINS = os.environ.get("CORS_ORIGINS", "*")
# Celery Configuration
CELERY = {
"broker_url": os.environ.get("CELERY_BROKER_URL", "redis://redis:6379/0"),
"result_backend": os.environ.get("CELERY_RESULT_BACKEND", "redis://redis:6379/0"),
"result_backend": os.environ.get(
"CELERY_RESULT_BACKEND", "redis://redis:6379/0"
),
"task_serializer": "json",
"result_serializer": "json",
"accept_content": ["json"],
@ -31,12 +34,14 @@ class Config:
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
@ -44,9 +49,12 @@ class TestingConfig(Config):
class ProductionConfig(Config):
"""Production configuration"""
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL") or "postgresql://user:password@localhost/proddb"
SQLALCHEMY_DATABASE_URI = (
os.environ.get("DATABASE_URL") or "postgresql://user:password@localhost/proddb"
)
# Security headers
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
@ -56,5 +64,5 @@ class ProductionConfig(Config):
config_by_name = {
"dev": DevelopmentConfig,
"test": TestingConfig,
"prod": ProductionConfig
}
"prod": ProductionConfig,
}

6
backend/app/models/__init__.py
View file

@ -1,5 +1,5 @@
from app.models.user import User
from app.models.product import Product
from app.models.order import Order, OrderItem
from app.models.product import Product
from app.models.user import User
__all__ = ["User", "Product", "Order", "OrderItem"]
__all__ = ["User", "Product", "Order", "OrderItem"]

40
backend/app/models/order.py
View file

@ -1,23 +1,34 @@
from datetime import datetime, UTC
from datetime import UTC, datetime
from app import db
class Order(db.Model):
"""Order model"""
__tablename__ = "orders"
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
status = db.Column(db.String(20), default="pending", index=True)
total_amount = db.Column(db.Numeric(10, 2), nullable=False)
shipping_address = db.Column(db.Text)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
updated_at = db.Column(
db.DateTime,
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
# Relationships
user = db.relationship("User", back_populates="orders")
items = db.relationship("OrderItem", back_populates="order", lazy="dynamic", cascade="all, delete-orphan")
items = db.relationship(
"OrderItem",
back_populates="order",
lazy="dynamic",
cascade="all, delete-orphan",
)
def to_dict(self):
"""Convert order to dictionary"""
return {
@ -28,27 +39,28 @@ class Order(db.Model):
"shipping_address": self.shipping_address,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
"items": [item.to_dict() for item in self.items]
"items": [item.to_dict() for item in self.items],
}
def __repr__(self):
return f"<Order {self.id}>"
class OrderItem(db.Model):
"""Order Item model"""
__tablename__ = "order_items"
id = db.Column(db.Integer, primary_key=True)
order_id = db.Column(db.Integer, db.ForeignKey("orders.id"), nullable=False)
product_id = db.Column(db.Integer, db.ForeignKey("products.id"), nullable=False)
quantity = db.Column(db.Integer, nullable=False)
price = db.Column(db.Numeric(10, 2), nullable=False)
# Relationships
order = db.relationship("Order", back_populates="items")
product = db.relationship("Product", back_populates="order_items")
def to_dict(self):
"""Convert order item to dictionary"""
return {
@ -56,8 +68,8 @@ class OrderItem(db.Model):
"order_id": self.order_id,
"product_id": self.product_id,
"quantity": self.quantity,
"price": float(self.price) if self.price else None
"price": float(self.price) if self.price else None,
}
def __repr__(self):
return f"<OrderItem {self.id}>"
return f"<OrderItem {self.id}>"

22
backend/app/models/product.py
View file

@ -1,11 +1,13 @@
from datetime import datetime, UTC
from datetime import UTC, datetime
from app import db
class Product(db.Model):
"""Product model"""
__tablename__ = "products"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(200), nullable=False, index=True)
description = db.Column(db.Text)
@ -14,11 +16,15 @@ class Product(db.Model):
image_url = db.Column(db.String(500))
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
updated_at = db.Column(
db.DateTime,
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
# Relationships
order_items = db.relationship("OrderItem", back_populates="product", lazy="dynamic")
def to_dict(self):
"""Convert product to dictionary"""
return {
@ -30,8 +36,8 @@ class Product(db.Model):
"image_url": self.image_url,
"is_active": self.is_active,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f"<Product {self.name}>"
return f"<Product {self.name}>"

29
backend/app/models/user.py
View file

@ -1,12 +1,15 @@
from datetime import datetime, UTC
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import UTC, datetime
from werkzeug.security import check_password_hash, generate_password_hash
from app import db
class User(db.Model):
"""User model"""
__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
username = db.Column(db.String(80), unique=True, nullable=False, index=True)
@ -16,19 +19,23 @@ class User(db.Model):
is_active = db.Column(db.Boolean, default=True)
is_admin = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
updated_at = db.Column(
db.DateTime,
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)
# Relationships
orders = db.relationship("Order", back_populates="user", lazy="dynamic")
def set_password(self, password):
"""Hash and set password"""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Check if provided password matches hash"""
return check_password_hash(self.password_hash, password)
def to_dict(self):
"""Convert user to dictionary"""
return {
@ -40,8 +47,8 @@ class User(db.Model):
"is_active": self.is_active,
"is_admin": self.is_admin,
"created_at": self.created_at.isoformat() if self.created_at else None,
"updated_at": self.updated_at.isoformat() if self.updated_at else None
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
}
def __repr__(self):
return f"<User {self.username}>"
return f"<User {self.username}>"

2
backend/app/routes/__init__.py
View file

@ -1,4 +1,4 @@
from .api import api_bp
from .health import health_bp
__all__ = ["api_bp", "health_bp"]
__all__ = ["api_bp", "health_bp"]

254
backend/app/routes/api.py
View file

@ -1,12 +1,15 @@
import time
from decimal import Decimal
from flask import Blueprint, jsonify, request
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
get_jwt_identity,
jwt_required,
)
from pydantic import ValidationError
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity, create_access_token, create_refresh_token
from app import db
from app.models import User, Product, OrderItem, Order
from app.celery import celery
from app.models import Order, OrderItem, Product, User
from app.schemas import ProductCreateRequest, ProductResponse
api_bp = Blueprint("api", __name__)
@ -17,24 +20,24 @@ api_bp = Blueprint("api", __name__)
def register():
"""Register a new user"""
data = request.get_json()
if not data or not data.get("email") or not data.get("password"):
return jsonify({"error": "Email and password are required"}), 400
if User.query.filter_by(email=data["email"]).first():
return jsonify({"error": "Email already exists"}), 400
user = User(
email=data["email"],
username=data.get("username", data["email"].split("@")[0]),
first_name=data.get("first_name"),
last_name=data.get("last_name")
last_name=data.get("last_name"),
)
user.set_password(data["password"])
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@ -42,26 +45,31 @@ def register():
def login():
"""Login user"""
data = request.get_json()
if not data or not data.get("email") or not data.get("password"):
return jsonify({"error": "Email and password are required"}), 400
user = User.query.filter_by(email=data["email"]).first()
if not user or not user.check_password(data["password"]):
return jsonify({"error": "Invalid credentials"}), 401
if not user.is_active:
return jsonify({"error": "Account is inactive"}), 401
access_token = create_access_token(identity=str(user.id))
refresh_token = create_refresh_token(identity=str(user.id))
return jsonify({
"user": user.to_dict(),
"access_token": access_token,
"refresh_token": refresh_token
}), 200
return (
jsonify(
{
"user": user.to_dict(),
"access_token": access_token,
"refresh_token": refresh_token,
}
),
200,
)
@api_bp.route("/users/me", methods=["GET"])
@ -70,10 +78,10 @@ def get_current_user():
"""Get current user"""
user_id = int(get_jwt_identity())
user = db.session.get(User, user_id)
if not user:
return jsonify({"error": "User not found"}), 404
return jsonify(user.to_dict()), 200
@ -81,12 +89,11 @@ def get_current_user():
@api_bp.route("/products", methods=["GET"])
def get_products():
"""Get all products"""
# time.sleep(5) # This adds a 5 second delay
products = Product.query.filter_by(is_active=True).all()
return jsonify([product.to_dict() for product in products]), 200
@ -105,29 +112,29 @@ def create_product():
"""Create a new product (admin only)"""
user_id = int(get_jwt_identity())
user = db.session.get(User, user_id)
if not user or not user.is_admin:
return jsonify({"error": "Admin access required"}), 403
try:
# Validate request data using Pydantic schema
product_data = ProductCreateRequest(**request.get_json())
product = Product(
name=product_data.name,
description=product_data.description,
price=product_data.price,
stock=product_data.stock,
image_url=product_data.image_url
image_url=product_data.image_url,
)
db.session.add(product)
db.session.commit()
# Use Pydantic schema for response
response = ProductResponse.model_validate(product)
return jsonify(response.model_dump()), 201
except ValidationError as e:
print(f"Pydantic Validation Error: {e.errors()}")
return jsonify({"error": "Validation error", "details": e.errors()}), 400
@ -139,25 +146,25 @@ def update_product(product_id):
"""Update a product (admin only)"""
user_id = int(get_jwt_identity())
user = db.session.get(User, user_id)
if not user or not user.is_admin:
return jsonify({"error": "Admin access required"}), 403
product = db.session.get(Product, product_id)
if not product:
return jsonify({"error": "Product not found"}), 404
data = request.get_json()
product.name = data.get("name", product.name)
product.description = data.get("description", product.description)
product.price = data.get("price", product.price)
product.stock = data.get("stock", product.stock)
product.image_url = data.get("image_url", product.image_url)
product.is_active = data.get("is_active", product.is_active)
db.session.commit()
return jsonify(product.to_dict()), 200
@ -167,17 +174,17 @@ def delete_product(product_id):
"""Delete a product (admin only)"""
user_id = int(get_jwt_identity())
user = db.session.get(User, user_id)
if not user or not user.is_admin:
return jsonify({"error": "Admin access required"}), 403
product = db.session.get(Product, product_id)
if not product:
return jsonify({"error": "Product not found"}), 404
db.session.delete(product)
db.session.commit()
return jsonify({"message": "Product deleted"}), 200
@ -197,49 +204,54 @@ def create_order():
"""Create a new order"""
user_id = int(get_jwt_identity())
data = request.get_json()
if not data or not data.get("items"):
return jsonify({"error": "Order items are required"}), 400
total_amount = 0
order_items = []
for item_data in data["items"]:
product = db.session.get(Product, item_data["product_id"])
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"]:
return jsonify({"error": f'Insufficient stock for {product.name}'}), 400
return jsonify({"error": f"Insufficient stock for {product.name}"}), 400
item_total = product.price * item_data["quantity"]
total_amount += item_total
order_items.append({
"product": product,
"quantity": item_data["quantity"],
"price": product.price
})
order_items.append(
{
"product": product,
"quantity": item_data["quantity"],
"price": product.price,
}
)
order = Order(
user_id=user_id,
total_amount=total_amount,
shipping_address=data.get("shipping_address")
shipping_address=data.get("shipping_address"),
)
db.session.add(order)
db.session.flush()
for item_data in order_items:
order_item = OrderItem(
order_id=order.id,
product_id=item_data["product"].id,
quantity=item_data["quantity"],
price=item_data["price"]
price=item_data["price"],
)
item_data["product"].stock -= item_data["quantity"]
db.session.add(order_item)
db.session.commit()
return jsonify(order.to_dict()), 201
@ -251,12 +263,12 @@ def get_order(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:
user = db.session.get(User, user_id)
if not user or not user.is_admin:
return jsonify({"error": "Access denied"}), 403
return jsonify(order.to_dict()), 200
@ -267,14 +279,15 @@ def trigger_hello_task():
"""Trigger the hello task"""
data = request.get_json() or {}
name = data.get("name", "World")
task = celery.send_task("tasks.print_hello", args=[name])
return jsonify({
"message": "Hello task triggered",
"task_id": task.id,
"status": "pending"
}), 202
return (
jsonify(
{"message": "Hello task triggered", "task_id": task.id, "status": "pending"}
),
202,
)
@api_bp.route("/tasks/divide", methods=["POST"])
@ -284,15 +297,20 @@ def trigger_divide_task():
data = request.get_json() or {}
x = data.get("x", 10)
y = data.get("y", 2)
task = celery.send_task("tasks.divide_numbers", args=[x, y])
return jsonify({
"message": "Divide task triggered",
"task_id": task.id,
"operation": f"{x} / {y}",
"status": "pending"
}), 202
return (
jsonify(
{
"message": "Divide task triggered",
"task_id": task.id,
"operation": f"{x} / {y}",
"status": "pending",
}
),
202,
)
@api_bp.route("/tasks/report", methods=["POST"])
@ -300,12 +318,17 @@ def trigger_divide_task():
def trigger_report_task():
"""Trigger the daily report task"""
task = celery.send_task("tasks.send_daily_report")
return jsonify({
"message": "Daily report task triggered",
"task_id": task.id,
"status": "pending"
}), 202
return (
jsonify(
{
"message": "Daily report task triggered",
"task_id": task.id,
"status": "pending",
}
),
202,
)
@api_bp.route("/tasks/stats", methods=["POST"])
@ -314,19 +337,15 @@ def trigger_stats_task():
"""Trigger product statistics update task"""
data = request.get_json() or {}
product_id = data.get("product_id")
if product_id:
task = celery.send_task("tasks.update_product_statistics", args=[product_id])
message = f"Product statistics update triggered for product {product_id}"
else:
task = celery.send_task("tasks.update_product_statistics", args=[None])
message = "Product statistics update triggered for all products"
return jsonify({
"message": message,
"task_id": task.id,
"status": "pending"
}), 202
return jsonify({"message": message, "task_id": task.id, "status": "pending"}), 202
@api_bp.route("/tasks/long-running", methods=["POST"])
@ -335,14 +354,19 @@ def trigger_long_running_task():
"""Trigger a long-running task"""
data = request.get_json() or {}
iterations = data.get("iterations", 10)
task = celery.send_task("tasks.long_running_task", args=[iterations])
return jsonify({
"message": f"Long-running task triggered with {iterations} iterations",
"task_id": task.id,
"status": "pending"
}), 202
return (
jsonify(
{
"message": f"Long-running task triggered with {iterations} iterations",
"task_id": task.id,
"status": "pending",
}
),
202,
)
@api_bp.route("/tasks/<task_id>", methods=["GET"])
@ -350,20 +374,20 @@ def trigger_long_running_task():
def get_task_status(task_id):
"""Get the status of a Celery task"""
task_result = celery.AsyncResult(task_id)
response = {
"task_id": task_id,
"status": task_result.status,
"ready": task_result.ready()
"ready": task_result.ready(),
}
if task_result.ready():
if task_result.successful():
response["result"] = task_result.result
else:
response["error"] = str(task_result.result)
response["traceback"] = task_result.traceback
return jsonify(response), 200
@ -374,20 +398,18 @@ def celery_health():
# Try to ping the worker
inspector = celery.control.inspect()
stats = inspector.stats()
if stats:
return jsonify({
"status": "healthy",
"workers": len(stats),
"workers_info": stats
}), 200
return (
jsonify(
{"status": "healthy", "workers": len(stats), "workers_info": stats}
),
200,
)
else:
return jsonify({
"status": "unhealthy",
"message": "No workers available"
}), 503
return (
jsonify({"status": "unhealthy", "message": "No workers available"}),
503,
)
except Exception as e:
return jsonify({
"status": "error",
"message": str(e)
}), 500
return jsonify({"status": "error", "message": str(e)}), 500

16
backend/app/routes/health.py
View file

@ -1,22 +1,16 @@
from flask import Blueprint, jsonify
health_bp = Blueprint('health', __name__)
health_bp = Blueprint("health", __name__)
@health_bp.route('/', methods=['GET'])
@health_bp.route("/", methods=["GET"])
def health_check():
"""Health check endpoint"""
return jsonify({
'status': 'healthy',
'service': 'crafting-shop-backend'
}), 200
return jsonify({"status": "healthy", "service": "crafting-shop-backend"}), 200
@health_bp.route('/readiness', methods=['GET'])
@health_bp.route("/readiness", methods=["GET"])
def readiness_check():
"""Readiness check endpoint"""
# Add database check here if needed
return jsonify({
'status': 'ready',
'service': 'crafting-shop-backend'
}), 200
return jsonify({"status": "ready", "service": "crafting-shop-backend"}), 200

2
backend/app/schemas/__init__.py
View file

@ -1,4 +1,4 @@
"""Pydantic schemas for request/response validation"""
from app.schemas.product import ProductCreateRequest, ProductResponse
__all__ = ["ProductCreateRequest", "ProductResponse"]
__all__ = ["ProductCreateRequest", "ProductResponse"]

27
backend/app/schemas/product.py
View file

@ -1,12 +1,14 @@
"""Pydantic schemas for Product model"""
from pydantic import BaseModel, Field, field_validator, ConfigDict
from decimal import Decimal
from datetime import datetime
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field, field_validator
class ProductCreateRequest(BaseModel):
"""Schema for creating a new product"""
model_config = ConfigDict(
json_schema_extra={
"example": {
@ -14,17 +16,21 @@ class ProductCreateRequest(BaseModel):
"description": "A beautiful handcrafted bowl made from oak",
"price": 45.99,
"stock": 10,
"image_url": "https://example.com/bowl.jpg"
"image_url": "https://example.com/bowl.jpg",
}
}
)
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)")
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")
image_url: Optional[str] = Field(
None, max_length=500, description="Product image URL"
)
@field_validator("price")
@classmethod
def validate_price(cls, v: Decimal) -> Decimal:
@ -36,6 +42,7 @@ class ProductCreateRequest(BaseModel):
class ProductResponse(BaseModel):
"""Schema for product response"""
model_config = ConfigDict(
from_attributes=True,
json_schema_extra={
@ -48,11 +55,11 @@ class ProductResponse(BaseModel):
"image_url": "https://example.com/bowl.jpg",
"is_active": True,
"created_at": "2024-01-15T10:30:00",
"updated_at": "2024-01-15T10:30:00"
"updated_at": "2024-01-15T10:30:00",
}
}
},
)
id: int
name: str
description: Optional[str] = None

2
backend/app/services/__init__.py
View file

@ -1 +1 @@
"""Business logic services"""
"""Business logic services"""

2
backend/app/utils/__init__.py
View file

@ -1 +1 @@
"""Utility functions and helpers"""
"""Utility functions and helpers"""

2
backend/tests/__init__.py
View file

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

77
backend/tests/conftest.py
View file

@ -1,28 +1,31 @@
"""Pytest configuration and fixtures"""
import pytest
import tempfile
import os
import tempfile
import pytest
from faker import Faker
from app import create_app, db
from app.models import User, Product, Order, OrderItem
from app.models import Order, OrderItem, Product, User
fake = Faker()
@pytest.fixture(scope='function')
@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'
})
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()
@ -62,9 +65,9 @@ def admin_user(db_session):
first_name=fake.first_name(),
last_name=fake.last_name(),
is_admin=True,
is_active=True
is_active=True,
)
user.set_password('password123')
user.set_password("password123")
db_session.add(user)
db_session.commit()
return user
@ -79,9 +82,9 @@ def regular_user(db_session):
first_name=fake.first_name(),
last_name=fake.last_name(),
is_admin=False,
is_active=True
is_active=True,
)
user.set_password('password123')
user.set_password("password123")
db_session.add(user)
db_session.commit()
return user
@ -96,9 +99,9 @@ def inactive_user(db_session):
first_name=fake.first_name(),
last_name=fake.last_name(),
is_admin=False,
is_active=False
is_active=False,
)
user.set_password('password123')
user.set_password("password123")
db_session.add(user)
db_session.commit()
return user
@ -112,7 +115,7 @@ def product(db_session):
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()
image_url=fake.url(),
)
db_session.add(product)
db_session.commit()
@ -129,7 +132,7 @@ def products(db_session):
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()
image_url=fake.url(),
)
db_session.add(product)
products.append(product)
@ -140,40 +143,36 @@ def products(db_session):
@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'
})
response = client.post(
"/api/auth/login", json={"email": regular_user.email, "password": "password123"}
)
data = response.get_json()
token = data['access_token']
token = data["access_token"]
print(f"Auth headers token for user {regular_user.email}: {token[:50]}...")
return {'Authorization': f'Bearer {token}'}
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'
})
response = client.post(
"/api/auth/login", json={"email": admin_user.email, "password": "password123"}
)
data = response.get_json()
token = data['access_token']
token = data["access_token"]
print(f"Admin headers token for user {admin_user.email}: {token[:50]}...")
return {'Authorization': f'Bearer {token}'}
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()
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)
@ -181,11 +180,11 @@ def order(db_session, regular_user, products):
order_id=order.id,
product_id=product.id,
quantity=quantity,
price=product.price
price=product.price,
)
total_amount += float(product.price) * quantity
db_session.add(order_item)
order.total_amount = total_amount
db_session.commit()
return order

128
backend/tests/test_models.py
View file

@ -1,8 +1,9 @@
"""Test models"""
import pytest
from decimal import Decimal
from datetime import datetime
from app.models import User, Product, Order, OrderItem
import pytest
from app.models import Order, OrderItem, Product, User
class TestUserModel:
@ -12,62 +13,62 @@ class TestUserModel:
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',
email="test@example.com",
username="testuser",
first_name="Test",
last_name="User",
is_admin=False,
is_active=True
is_active=True,
)
user.set_password('password123')
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'
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')
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
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'
email="test@example.com",
username="testuser",
first_name="Test",
last_name="User",
)
user.set_password('password123')
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
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')
user = User(email="test@example.com", username="testuser")
user.set_password("password123")
db_session.add(user)
db_session.commit()
assert repr(user) == '<User testuser>'
assert repr(user) == "<User testuser>"
class TestProductModel:
@ -77,18 +78,18 @@ class TestProductModel:
def test_product_creation(self, db_session):
"""Test creating a product"""
product = Product(
name='Test Product',
description='A test product',
price=Decimal('99.99'),
name="Test Product",
description="A test product",
price=Decimal("99.99"),
stock=10,
image_url='https://example.com/product.jpg'
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.name == "Test Product"
assert product.price == Decimal("99.99")
assert product.stock == 10
assert product.is_active is True
@ -96,27 +97,24 @@ class TestProductModel:
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
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)
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')
)
product = Product(name="Test Product", price=Decimal("9.99"))
db_session.add(product)
db_session.commit()
@ -128,11 +126,11 @@ class TestProductModel:
@pytest.mark.unit
def test_product_repr(self, db_session):
"""Test product string representation"""
product = Product(name='Test Product', price=Decimal('9.99'))
product = Product(name="Test Product", price=Decimal("9.99"))
db_session.add(product)
db_session.commit()
assert repr(product) == '<Product Test Product>'
assert repr(product) == "<Product Test Product>"
class TestOrderModel:
@ -143,31 +141,31 @@ class TestOrderModel:
"""Test creating an order"""
order = Order(
user_id=regular_user.id,
total_amount=Decimal('199.99'),
shipping_address='123 Test St'
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')
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'
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)
assert order_dict["user_id"] == regular_user.id
assert order_dict["total_amount"] == 199.99
assert isinstance(order_dict["created_at"], str)
class TestOrderItemModel:
@ -177,10 +175,7 @@ class TestOrderItemModel:
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
order_id=order.id, product_id=product.id, quantity=2, price=product.price
)
db_session.add(order_item)
db_session.commit()
@ -194,15 +189,12 @@ class TestOrderItemModel:
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
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
assert item_dict["order_id"] == order.id
assert item_dict["product_id"] == product.id
assert item_dict["quantity"] == 2

260
backend/tests/test_routes.py
View file

@ -1,7 +1,5 @@
"""Test API routes"""
import pytest
import json
from decimal import Decimal
class TestAuthRoutes:
@ -10,101 +8,109 @@ class TestAuthRoutes:
@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'
})
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
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'
})
response = client.post(
"/api/auth/register", json={"email": "newuser@example.com"}
)
assert response.status_code == 400
data = response.get_json()
assert 'error' in data
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'
})
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()
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'
})
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
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):
@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
login_data["email"] = email
if password is not None:
login_data['password'] = password
login_data["password"] = password
response = client.post('/api/auth/login', json=login_data)
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'
})
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()
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)
response = client.get("/api/users/me", headers=auth_headers)
assert response.status_code == 200
data = response.get_json()
assert data['email'] == regular_user.email
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')
response = client.get("/api/users/me")
assert response.status_code == 401
@ -114,7 +120,7 @@ class TestProductRoutes:
@pytest.mark.product
def test_get_products(self, client, products):
"""Test getting all products"""
response = client.get('/api/products')
response = client.get("/api/products")
assert response.status_code == 200
data = response.get_json()
@ -123,7 +129,7 @@ class TestProductRoutes:
@pytest.mark.product
def test_get_products_empty(self, client):
"""Test getting products when none exist"""
response = client.get('/api/products')
response = client.get("/api/products")
assert response.status_code == 200
data = response.get_json()
@ -132,113 +138,122 @@ class TestProductRoutes:
@pytest.mark.product
def test_get_single_product(self, client, product):
"""Test getting a single product"""
response = client.get(f'/api/products/{product.id}')
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
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')
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
})
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
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
})
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()
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
})
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
})
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']
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'
})
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']
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
})
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
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
})
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
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)
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}')
response = client.get(f"/api/products/{product.id}")
assert response.status_code == 404
@ -248,7 +263,7 @@ class TestOrderRoutes:
@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)
response = client.get("/api/orders", headers=auth_headers)
assert response.status_code == 200
data = response.get_json()
@ -257,63 +272,68 @@ class TestOrderRoutes:
@pytest.mark.order
def test_get_orders_unauthorized(self, client):
"""Test getting orders without authentication"""
response = client.get('/api/orders')
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'
})
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
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):
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}
]
})
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()
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())
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
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'
})
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

190
backend/tests/test_schemas.py
View file

@ -1,7 +1,9 @@
"""Test Pydantic schemas"""
from decimal import Decimal
import pytest
from pydantic import ValidationError
from decimal import Decimal
from app.schemas import ProductCreateRequest, ProductResponse
@ -12,31 +14,28 @@ class TestProductCreateRequestSchema:
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'
"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.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']
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
}
data = {"name": "Simple Product", "price": 19.99}
product = ProductCreateRequest(**data)
assert product.name == 'Simple Product'
assert product.price == Decimal('19.99')
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
@ -44,134 +43,107 @@ class TestProductCreateRequestSchema:
@pytest.mark.unit
def test_missing_name(self):
"""Test request with missing name"""
data = {
'price': 19.99
}
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)
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'
}
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)
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
}
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)
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
}
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)
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
}
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)
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
}
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)
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
}
data = {"name": "A" * 201, "price": 19.99} # Exceeds 200 character limit
with pytest.raises(ValidationError) as exc_info:
ProductCreateRequest(**data)
errors = exc_info.value.errors()
assert any(error['loc'] == ('name',) for error in 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
"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)
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'
}
data = {"name": "Test Product", "price": "29.99"}
product = ProductCreateRequest(**data)
assert product.price == Decimal('29.99')
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'
}
data = {"name": "Test Product", "price": 19.99, "stock": "10"}
product = ProductCreateRequest(**data)
assert product.stock == 10
@ -185,20 +157,20 @@ class TestProductResponseSchema:
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'
"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.name == "Test Product"
assert product.price == 45.99
assert product.stock == 10
assert product.is_active is True
@ -207,11 +179,11 @@ class TestProductResponseSchema:
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
"id": 1,
"name": "Test Product",
"price": 19.99,
"stock": 0,
"is_active": True,
}
product = ProductResponse(**data)
@ -224,19 +196,19 @@ class TestProductResponseSchema:
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
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.name == "Test Product"
assert response.price == 45.99
assert response.stock == 10
@ -244,34 +216,34 @@ class TestProductResponseSchema:
def test_model_dump(self):
"""Test model_dump method"""
data = {
'id': 1,
'name': 'Test Product',
'price': 19.99,
'stock': 5,
'is_active': True
"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
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
"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
assert "Test Product" in json_str