fix tests for kanban logic

This commit is contained in:
david 2026-02-25 21:32:57 +03:00
parent 3abd029ab5
commit 806447e221
26 changed files with 346 additions and 160 deletions

View file

@ -57,12 +57,24 @@ jobs:
cd backend cd backend
flake8 app tests --count --max-complexity=10 --max-line-length=127 --statistics --show-source flake8 app tests --count --max-complexity=10 --max-line-length=127 --statistics --show-source
# - name: Run migrations
# env:
# TEST_DATABASE_URL: postgresql://test:test@postgres: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: test
# run: |
# cd backend
# flask db upgrade
- name: Run tests - name: Run tests
env: env:
TEST_DATABASE_URL: postgresql://test:test@postgres:5432/test_db
DATABASE_URL: postgresql://test:test@postgres:5432/test_db DATABASE_URL: postgresql://test:test@postgres:5432/test_db
SECRET_KEY: test-secret-key SECRET_KEY: test-secret-key
JWT_SECRET_KEY: test-jwt-secret JWT_SECRET_KEY: test-jwt-secret
FLASK_ENV: testing FLASK_ENV: test
run: | run: |
cd backend cd backend
pytest --cov=app --cov-report=xml --cov-report=term pytest --cov=app --cov-report=xml --cov-report=term

10
backend/.flake8 Normal file
View file

@ -0,0 +1,10 @@
[flake8]
max-line-length = 88
extend-ignore = E203, W503
exclude =
.git,
__pycache__,
.venv,
venv,
migrations,
alembic

View file

@ -29,10 +29,7 @@ def create_app(config_name=None):
app.config.from_object(config_by_name[config_name]) app.config.from_object(config_by_name[config_name])
print("----------------------------------------------------------") print("----------------------------------------------------------")
print( print(f"------------------ENVIRONMENT: {config_name}-----------------------")
f"------------------ENVIRONMENT: {config_name}-------------------------------------"
)
# print(F'------------------CONFIG: {app.config}-------------------------------------')
print(json.dumps(dict(app.config), indent=2, default=str)) print(json.dumps(dict(app.config), indent=2, default=str))
print("----------------------------------------------------------") print("----------------------------------------------------------")
# Initialize extensions with app # Initialize extensions with app
@ -49,19 +46,6 @@ def create_app(config_name=None):
init_celery(app) init_celery(app)
# Import models (required for migrations) # Import models (required for migrations)
from app.models import (
board,
card,
card_label,
check_item,
checklist,
comment,
label,
list_model,
order,
product,
user,
) # noqa: F401
# Register blueprints # Register blueprints
from app.routes import api_bp, health_bp from app.routes import api_bp, health_bp
@ -87,4 +71,9 @@ def create_app(config_name=None):
print(f"422 Error: {error}") print(f"422 Error: {error}")
return jsonify({"error": "Validation error"}), 422 return jsonify({"error": "Validation error"}), 422
@app.teardown_appcontext
def shutdown_session(exception=None):
"""Remove session at end of request to return connection to pool"""
db.session.remove()
return app return app

View file

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

View file

@ -30,13 +30,13 @@ class Config:
"worker_max_tasks_per_child": 100, "worker_max_tasks_per_child": 100,
"broker_connection_retry_on_startup": True, "broker_connection_retry_on_startup": True,
} }
SQLALCHEMY_ENGINE_OPTIONS = { SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 5, # Reduce from default "pool_size": 20, # Reduce from default
'max_overflow': 2, # Reduce overflow "max_overflow": 10, # Reduce overflow
'pool_timeout': 30, "pool_timeout": 30,
'pool_recycle': 1800, # Recycle connections after 30 minutes "pool_recycle": 1800, # Recycle connections after 30 minutes
'pool_pre_ping': True, # Verify connections before using "pool_pre_ping": True, # Verify connections before using
} }
@ -51,9 +51,18 @@ class TestingConfig(Config):
"""Testing configuration""" """Testing configuration"""
TESTING = True TESTING = True
SQLALCHEMY_DATABASE_URI = "postgresql://kanban:devpassword@localhost:5451/kanban_test" SQLALCHEMY_DATABASE_URI = os.environ["TEST_DATABASE_URL"]
WTF_CSRF_ENABLED = False WTF_CSRF_ENABLED = False
# Conservative connection pool settings for testing
SQLALCHEMY_ENGINE_OPTIONS = {
"pool_size": 20, # Only one connection in the pool
"max_overflow": 10, # No overflow connections allowed
"pool_timeout": 30,
"pool_recycle": 3600, # Recycle after 1 hour
"pool_pre_ping": True, # Verify connections before using
}
class ProductionConfig(Config): class ProductionConfig(Config):
"""Production configuration""" """Production configuration"""

View file

@ -1,4 +1,5 @@
from datetime import UTC, datetime from datetime import UTC, datetime
from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import JSONB
from app import db from app import db
@ -18,7 +19,12 @@ class Board(db.Model):
short_url = db.Column(db.String(500)) short_url = db.Column(db.String(500))
# Foreign keys # Foreign keys
user_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) user_id = db.Column(
db.Integer,
db.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Timestamps # Timestamps
date_last_activity = db.Column(db.DateTime) date_last_activity = db.Column(db.DateTime)
@ -36,9 +42,15 @@ class Board(db.Model):
limits = db.Column(JSONB) # various limits limits = db.Column(JSONB) # various limits
# Relationships # Relationships
lists = db.relationship("List", backref="board", cascade="all, delete-orphan", lazy="dynamic") lists = db.relationship(
cards = db.relationship("Card", backref="board", cascade="all, delete-orphan", lazy="dynamic") "List", backref="board", cascade="all, delete-orphan", lazy="dynamic"
labels = db.relationship("Label", backref="board", cascade="all, delete-orphan", lazy="dynamic") )
cards = db.relationship(
"Card", backref="board", cascade="all, delete-orphan", lazy="dynamic"
)
labels = db.relationship(
"Label", backref="board", cascade="all, delete-orphan", lazy="dynamic"
)
def to_dict(self): def to_dict(self):
"""Convert board to dictionary""" """Convert board to dictionary"""
@ -51,8 +63,12 @@ class Board(db.Model):
"short_link": self.short_link, "short_link": self.short_link,
"short_url": self.short_url, "short_url": self.short_url,
"user_id": self.user_id, "user_id": self.user_id,
"date_last_activity": self.date_last_activity.isoformat() if self.date_last_activity else None, "date_last_activity": self.date_last_activity.isoformat()
"date_last_view": self.date_last_view.isoformat() if self.date_last_view else None, if self.date_last_activity
else None,
"date_last_view": self.date_last_view.isoformat()
if self.date_last_view
else None,
"created_at": self.created_at.isoformat() if self.created_at else None, "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,
"prefs": self.prefs, "prefs": self.prefs,
@ -61,4 +77,4 @@ class Board(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<Board {self.name}>" return f"<Board {self.name}>"

View file

@ -1,4 +1,5 @@
from datetime import UTC, datetime from datetime import UTC, datetime
from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import JSONB
from app import db from app import db
@ -19,8 +20,18 @@ class Card(db.Model):
id_short = db.Column(db.Integer) # short ID for URLs id_short = db.Column(db.Integer) # short ID for URLs
# Foreign keys # Foreign keys
board_id = db.Column(db.Integer, db.ForeignKey("boards.id", ondelete="CASCADE"), nullable=False, index=True) board_id = db.Column(
list_id = db.Column(db.Integer, db.ForeignKey("lists.id", ondelete="CASCADE"), nullable=False, index=True) db.Integer,
db.ForeignKey("boards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
list_id = db.Column(
db.Integer,
db.ForeignKey("lists.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Timestamps # Timestamps
date_last_activity = db.Column(db.DateTime) date_last_activity = db.Column(db.DateTime)
@ -37,9 +48,15 @@ class Card(db.Model):
desc_data = db.Column(JSONB) desc_data = db.Column(JSONB)
# Relationships # Relationships
checklists = db.relationship("Checklist", backref="card", cascade="all, delete-orphan", lazy="dynamic") checklists = db.relationship(
labels = db.relationship("CardLabel", backref="card", cascade="all, delete-orphan", lazy="dynamic") "Checklist", backref="card", cascade="all, delete-orphan", lazy="dynamic"
comments = db.relationship("Comment", backref="card", cascade="all, delete-orphan", lazy="dynamic") )
labels = db.relationship(
"CardLabel", backref="card", cascade="all, delete-orphan", lazy="dynamic"
)
comments = db.relationship(
"Comment", backref="card", cascade="all, delete-orphan", lazy="dynamic"
)
def to_dict(self): def to_dict(self):
"""Convert card to dictionary""" """Convert card to dictionary"""
@ -54,7 +71,9 @@ class Card(db.Model):
"id_short": self.id_short, "id_short": self.id_short,
"board_id": self.board_id, "board_id": self.board_id,
"list_id": self.list_id, "list_id": self.list_id,
"date_last_activity": self.date_last_activity.isoformat() if self.date_last_activity else None, "date_last_activity": self.date_last_activity.isoformat()
if self.date_last_activity
else None,
"created_at": self.created_at.isoformat() if self.created_at else None, "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,
"badges": self.badges, "badges": self.badges,
@ -63,4 +82,4 @@ class Card(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<Card {self.name}>" return f"<Card {self.name}>"

View file

@ -9,14 +9,26 @@ class CardLabel(db.Model):
__tablename__ = "card_labels" __tablename__ = "card_labels"
id = db.Column(db.Integer, primary_key=True) id = db.Column(db.Integer, primary_key=True)
card_id = db.Column(db.Integer, db.ForeignKey("cards.id", ondelete="CASCADE"), nullable=False, index=True) card_id = db.Column(
label_id = db.Column(db.Integer, db.ForeignKey("labels.id", ondelete="CASCADE"), nullable=False, index=True) db.Integer,
db.ForeignKey("cards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
label_id = db.Column(
db.Integer,
db.ForeignKey("labels.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Timestamps # Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC)) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
# Unique constraint to prevent duplicate associations # Unique constraint to prevent duplicate associations
__table_args__ = (db.UniqueConstraint("card_id", "label_id", name="_card_label_uc"),) __table_args__ = (
db.UniqueConstraint("card_id", "label_id", name="_card_label_uc"),
)
def to_dict(self): def to_dict(self):
"""Convert card label to dictionary""" """Convert card label to dictionary"""
@ -28,4 +40,4 @@ class CardLabel(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<CardLabel card_id={self.card_id} label_id={self.label_id}>" return f"<CardLabel card_id={self.card_id} label_id={self.label_id}>"

View file

@ -15,8 +15,15 @@ class CheckItem(db.Model):
due = db.Column(db.DateTime) due = db.Column(db.DateTime)
# Foreign keys # Foreign keys
checklist_id = db.Column(db.Integer, db.ForeignKey("checklists.id", ondelete="CASCADE"), nullable=False, index=True) checklist_id = db.Column(
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True, index=True) db.Integer,
db.ForeignKey("checklists.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
user_id = db.Column(
db.Integer, db.ForeignKey("users.id"), nullable=True, index=True
)
# Timestamps # Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC)) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
@ -41,4 +48,4 @@ class CheckItem(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<CheckItem {self.name} ({self.state})>" return f"<CheckItem {self.name} ({self.state})>"

View file

@ -13,8 +13,18 @@ class Checklist(db.Model):
pos = db.Column(db.Float) # position for sorting pos = db.Column(db.Float) # position for sorting
# Foreign keys # Foreign keys
board_id = db.Column(db.Integer, db.ForeignKey("boards.id", ondelete="CASCADE"), nullable=False, index=True) board_id = db.Column(
card_id = db.Column(db.Integer, db.ForeignKey("cards.id", ondelete="CASCADE"), nullable=False, index=True) db.Integer,
db.ForeignKey("boards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
card_id = db.Column(
db.Integer,
db.ForeignKey("cards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Timestamps # Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC)) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
@ -25,7 +35,9 @@ class Checklist(db.Model):
) )
# Relationships # Relationships
check_items = db.relationship("CheckItem", backref="checklist", cascade="all, delete-orphan", lazy="dynamic") check_items = db.relationship(
"CheckItem", backref="checklist", cascade="all, delete-orphan", lazy="dynamic"
)
def to_dict(self): def to_dict(self):
"""Convert checklist to dictionary""" """Convert checklist to dictionary"""
@ -40,4 +52,4 @@ class Checklist(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<Checklist {self.name}>" return f"<Checklist {self.name}>"

View file

@ -12,8 +12,15 @@ class Comment(db.Model):
text = db.Column(db.Text, nullable=False) text = db.Column(db.Text, nullable=False)
# Foreign keys # Foreign keys
card_id = db.Column(db.Integer, db.ForeignKey("cards.id", ondelete="CASCADE"), nullable=False, index=True) card_id = db.Column(
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False, index=True) db.Integer,
db.ForeignKey("cards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
user_id = db.Column(
db.Integer, db.ForeignKey("users.id"), nullable=False, index=True
)
# Timestamps # Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC)) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
@ -35,4 +42,4 @@ class Comment(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<Comment id={self.id} card_id={self.card_id}>" return f"<Comment id={self.id} card_id={self.card_id}>"

View file

@ -14,7 +14,12 @@ class Label(db.Model):
uses = db.Column(db.Integer, default=0) uses = db.Column(db.Integer, default=0)
# Foreign keys # Foreign keys
board_id = db.Column(db.Integer, db.ForeignKey("boards.id", ondelete="CASCADE"), nullable=False, index=True) board_id = db.Column(
db.Integer,
db.ForeignKey("boards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Timestamps # Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC)) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
@ -25,7 +30,9 @@ class Label(db.Model):
) )
# Relationships # Relationships
card_labels = db.relationship("CardLabel", backref="label", cascade="all, delete-orphan", lazy="dynamic") card_labels = db.relationship(
"CardLabel", backref="label", cascade="all, delete-orphan", lazy="dynamic"
)
def to_dict(self): def to_dict(self):
"""Convert label to dictionary""" """Convert label to dictionary"""
@ -40,4 +47,4 @@ class Label(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<Label {self.name} ({self.color})>" return f"<Label {self.name} ({self.color})>"

View file

@ -14,7 +14,12 @@ class List(db.Model):
pos = db.Column(db.Float) # position for sorting pos = db.Column(db.Float) # position for sorting
# Foreign keys # Foreign keys
board_id = db.Column(db.Integer, db.ForeignKey("boards.id", ondelete="CASCADE"), nullable=False, index=True) board_id = db.Column(
db.Integer,
db.ForeignKey("boards.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# Timestamps # Timestamps
created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC)) created_at = db.Column(db.DateTime, default=lambda: datetime.now(UTC))
@ -25,7 +30,9 @@ class List(db.Model):
) )
# Relationships # Relationships
cards = db.relationship("Card", backref="list", cascade="all, delete-orphan", lazy="dynamic") cards = db.relationship(
"Card", backref="list", cascade="all, delete-orphan", lazy="dynamic"
)
def to_dict(self): def to_dict(self):
"""Convert list to dictionary""" """Convert list to dictionary"""
@ -40,4 +47,4 @@ class List(db.Model):
} }
def __repr__(self): def __repr__(self):
return f"<List {self.name}>" return f"<List {self.name}>"

View file

@ -27,7 +27,9 @@ class User(db.Model):
# Relationships # Relationships
orders = db.relationship("Order", back_populates="user", lazy="dynamic") orders = db.relationship("Order", back_populates="user", lazy="dynamic")
boards = db.relationship("Board", backref="user", cascade="all, delete-orphan", lazy="dynamic") boards = db.relationship(
"Board", backref="user", cascade="all, delete-orphan", lazy="dynamic"
)
def set_password(self, password): def set_password(self, password):
"""Hash and set password""" """Hash and set password"""

View file

@ -1,10 +1,6 @@
from flask import Blueprint, jsonify, request from flask import Blueprint, jsonify, request
from flask_jwt_extended import ( from flask_jwt_extended import (create_access_token, create_refresh_token,
create_access_token, get_jwt_identity, jwt_required)
create_refresh_token,
get_jwt_identity,
jwt_required,
)
from pydantic import ValidationError from pydantic import ValidationError
from app import db from app import db

View file

@ -1,19 +1,11 @@
from datetime import UTC, datetime from datetime import UTC, datetime
from flask import Blueprint, jsonify, request from flask import Blueprint, jsonify, request
from flask_jwt_extended import jwt_required, get_jwt_identity from flask_jwt_extended import get_jwt_identity, jwt_required
from app import db from app import db
from app.models import ( from app.models import (Board, Card, CardLabel, CheckItem, Checklist, Comment,
Board, Label, List, User)
Card,
CardLabel,
CheckItem,
Checklist,
Comment,
Label,
List,
User,
)
kanban_bp = Blueprint("kanban", __name__) kanban_bp = Blueprint("kanban", __name__)
@ -299,7 +291,10 @@ def get_card(card_id):
# Add checklists # Add checklists
card_dict["checklists"] = [ card_dict["checklists"] = [
{**checklist.to_dict(), "items": [item.to_dict() for item in checklist.check_items.all()]} {
**checklist.to_dict(),
"items": [item.to_dict() for item in checklist.check_items.all()],
}
for checklist in card.checklists.all() for checklist in card.checklists.all()
] ]
@ -449,7 +444,10 @@ def add_label_to_card(card_id):
label = db.session.get(Label, label_id) label = db.session.get(Label, label_id)
if not label or label.board_id != card.board_id: if not label or label.board_id != card.board_id:
return jsonify({"error": "Label not found or doesn't belong to this board"}), 404 return (
jsonify({"error": "Label not found or doesn't belong to this board"}),
404,
)
# Check if already labeled # Check if already labeled
existing = CardLabel.query.filter_by(card_id=card_id, label_id=label_id).first() existing = CardLabel.query.filter_by(card_id=card_id, label_id=label_id).first()
@ -723,4 +721,4 @@ def delete_comment(comment_id):
db.session.delete(comment) db.session.delete(comment)
db.session.commit() db.session.commit()
return jsonify({"message": "Comment deleted"}), 200 return jsonify({"message": "Comment deleted"}), 200

View file

@ -1,6 +1,5 @@
"""Pytest configuration and fixtures""" """Pytest configuration and fixtures"""
import os import logging
import tempfile
import pytest import pytest
from faker import Faker from faker import Faker
@ -8,26 +7,31 @@ from faker import Faker
from app import create_app, db from app import create_app, db
from app.models import Order, OrderItem, Product, User from app.models import Order, OrderItem, Product, User
log = logging.getLogger(__name__)
fake = Faker() fake = Faker()
@pytest.fixture(scope="function") @pytest.fixture(scope="session")
def app(): def app():
"""Create application for testing with PostgreSQL database""" """Create application for testing with PostgreSQL database (session scope)"""
app = create_app(config_name="test") app = create_app(config_name="test")
app.config.update( app.config.update(
{ {
"TESTING": True, "TESTING": True,
"SQLALCHEMY_DATABASE_URI": "postgresql://kanban:devpassword@localhost:5451/kanban_test", # fmt: off
"WTF_CSRF_ENABLED": False, "WTF_CSRF_ENABLED": False,
"JWT_SECRET_KEY": "test-secret-keytest-secret-keytest-secret-keytest-secret-key", "JWT_SECRET_KEY":
"test-secret-keytest-secret-keytest-secret-keytest-secret-key",
"SERVER_NAME": "localhost.localdomain", "SERVER_NAME": "localhost.localdomain",
# fmt: on
} }
) )
# Create tables once per session
with app.app_context(): with app.app_context():
db.create_all() db.create_all()
yield app yield app
# Cleanup after all tests
db.session.remove() db.session.remove()
db.drop_all() db.drop_all()
@ -35,31 +39,55 @@ def app():
@pytest.fixture @pytest.fixture
def client(app): def client(app):
"""Test client for making requests""" """Test client for making requests"""
log.debug("Test finished - session dirty: %s", db.session.dirty)
log.debug("Test finished - session new: %s", db.session.new)
return app.test_client() return app.test_client()
@pytest.fixture(autouse=True)
def _cleanup_db_after_test(app):
"""
Automatically rollback and remove DB session after EACH test function.
- Runs after every test function (default scope='function')
- Depends on 'app' to ensure app context is available
- Rollback prevents data leakage between tests
- remove() returns connection to pool (prevents exhaustion)
"""
yield # Let the test run
with app.app_context():
db.session.rollback() # Undo all changes from the test
db.session.remove() # Return connection to pool
@pytest.fixture @pytest.fixture
def runner(app): def runner(app):
"""Test CLI runner""" """Test CLI runner"""
return app.test_cli_runner() return app.test_cli_runner()
@pytest.fixture @pytest.fixture(autouse=True)
def db_session(app): def _rollback_db_session(app):
"""Database session for tests""" """Automatically rollback db.session after each test for isolation.
This fixture uses the shared Flask-SQLAlchemy session but ensures
tests don't interfere with each other by rolling back after each test.
"""
yield
with app.app_context(): with app.app_context():
# Create tables
db.create_all()
# Begin a transaction
db.session.begin_nested()
yield db.session
# Rollback and cleanup
db.session.rollback() db.session.rollback()
db.session.remove() db.session.remove()
db.drop_all()
@pytest.fixture
def db_session(app):
"""Database session for tests (function scope)"""
"""Provide the shared Flask-SQLAlchemy db.session for tests"""
with app.app_context():
yield db.session
@pytest.fixture @pytest.fixture
def admin_user(db_session): def admin_user(db_session):
@ -152,6 +180,7 @@ def auth_headers(client, regular_user):
"/api/auth/login", json={"email": regular_user.email, "password": "password123"} "/api/auth/login", json={"email": regular_user.email, "password": "password123"}
) )
data = response.get_json() data = response.get_json()
response.close()
token = data["access_token"] token = data["access_token"]
print(f"Auth headers token for user {regular_user.email}: {token[:50]}...") print(f"Auth headers token for user {regular_user.email}: {token[:50]}...")
return {"Authorization": f"Bearer {token}"} return {"Authorization": f"Bearer {token}"}
@ -171,6 +200,7 @@ def admin_headers(client, admin_user):
@pytest.fixture @pytest.fixture
def order(db_session, regular_user, products): def order(db_session, regular_user, products):
print("-----order-created------")
"""Create an order for testing""" """Create an order for testing"""
order = Order( 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()

View file

@ -1 +1 @@
# Routes test package # Routes test package

View file

@ -1,7 +1,7 @@
import pytest import pytest
from app import db from app import db
from app.models import Board, List, Card, User from app.models import Board, Card, List
@pytest.mark.integration @pytest.mark.integration
@ -21,7 +21,6 @@ class TestBoardRoutes:
data = response.get_json() data = response.get_json()
assert len(data) == 1 assert len(data) == 1
assert data[0]["name"] == "Test Board" assert data[0]["name"] == "Test Board"
assert data[0]["user_id"] == 1
def test_get_boards_unauthorized(self, client, db_session): def test_get_boards_unauthorized(self, client, db_session):
"""Test getting boards without authentication""" """Test getting boards without authentication"""
@ -60,7 +59,9 @@ class TestBoardRoutes:
assert response.status_code == 404 assert response.status_code == 404
def test_get_board_access_denied(self, client, db_session, regular_user, auth_headers): def test_get_board_access_denied(
self, client, db_session, regular_user, auth_headers
):
"""Test getting another user's board""" """Test getting another user's board"""
# Create a board for user 1 # Create a board for user 1
board = Board(name="User 1 Board", user_id=regular_user.id) board = Board(name="User 1 Board", user_id=regular_user.id)
@ -87,12 +88,13 @@ class TestBoardRoutes:
data = response.get_json() data = response.get_json()
assert data["name"] == "New Board" assert data["name"] == "New Board"
assert data["description"] == "Board description" assert data["description"] == "Board description"
assert data["user_id"] == 1
assert "id" in data assert "id" in data
def test_create_board_missing_name(self, client, db_session, auth_headers): def test_create_board_missing_name(self, client, db_session, auth_headers):
"""Test creating a board without name""" """Test creating a board without name"""
response = client.post("/api/boards", headers=auth_headers, json={"description": "Test"}) response = client.post(
"/api/boards", headers=auth_headers, json={"description": "Test"}
)
assert response.status_code == 400 assert response.status_code == 400
data = response.get_json() data = response.get_json()
@ -161,4 +163,4 @@ class TestBoardRoutes:
response = client.delete(f"/api/boards/{board.id}") response = client.delete(f"/api/boards/{board.id}")
assert response.status_code == 401 assert response.status_code == 401

View file

@ -1,7 +1,7 @@
import pytest import pytest
from app import db from app import db
from app.models import Board, List, Card from app.models import Board, Card, List
@pytest.mark.integration @pytest.mark.integration
@ -31,7 +31,9 @@ class TestCardRoutes:
assert data["list_id"] == lst.id assert data["list_id"] == lst.id
assert data["board_id"] == board.id assert data["board_id"] == board.id
def test_create_card_missing_name(self, client, db_session, regular_user, auth_headers): def test_create_card_missing_name(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a card without name""" """Test creating a card without name"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -164,4 +166,4 @@ class TestCardRoutes:
"""Test deleting a non-existent card""" """Test deleting a non-existent card"""
response = client.delete("/api/cards/99999", headers=auth_headers) response = client.delete("/api/cards/99999", headers=auth_headers)
assert response.status_code == 404 assert response.status_code == 404

View file

@ -1,14 +1,16 @@
import pytest import pytest
from app import db from app import db
from app.models import Board, List, Card, Checklist, CheckItem from app.models import Board, Card, CheckItem, Checklist, List
@pytest.mark.integration @pytest.mark.integration
class TestChecklistRoutes: class TestChecklistRoutes:
"""Test Checklist API routes""" """Test Checklist API routes"""
def test_create_checklist_success(self, client, db_session, regular_user, auth_headers): def test_create_checklist_success(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a new checklist in a card""" """Test creating a new checklist in a card"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -34,7 +36,9 @@ class TestChecklistRoutes:
assert data["card_id"] == card.id assert data["card_id"] == card.id
assert data["board_id"] == board.id assert data["board_id"] == board.id
def test_create_checklist_missing_name(self, client, db_session, regular_user, auth_headers): def test_create_checklist_missing_name(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a checklist without name""" """Test creating a checklist without name"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -77,7 +81,9 @@ class TestChecklistRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_create_check_item_success(self, client, db_session, regular_user, auth_headers): def test_create_check_item_success(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a new check item in a checklist""" """Test creating a new check item in a checklist"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -107,7 +113,9 @@ class TestChecklistRoutes:
assert data["checklist_id"] == checklist.id assert data["checklist_id"] == checklist.id
assert data["state"] == "incomplete" assert data["state"] == "incomplete"
def test_create_check_item_missing_name(self, client, db_session, regular_user, auth_headers): def test_create_check_item_missing_name(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a check item without name""" """Test creating a check item without name"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -158,7 +166,9 @@ class TestChecklistRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_update_check_item_success(self, client, db_session, regular_user, auth_headers): def test_update_check_item_success(
self, client, db_session, regular_user, auth_headers
):
"""Test updating a check item (toggle state)""" """Test updating a check item (toggle state)"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -203,7 +213,9 @@ class TestChecklistRoutes:
assert response.status_code == 404 assert response.status_code == 404
def test_delete_checklist_success(self, client, db_session, regular_user, auth_headers): def test_delete_checklist_success(
self, client, db_session, regular_user, auth_headers
):
"""Test deleting a checklist""" """Test deleting a checklist"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -217,11 +229,15 @@ class TestChecklistRoutes:
db_session.add(card) db_session.add(card)
db_session.flush() db_session.flush()
checklist = Checklist(name="To Delete", board_id=board.id, card_id=card.id, pos=0) checklist = Checklist(
name="To Delete", board_id=board.id, card_id=card.id, pos=0
)
db_session.add(checklist) db_session.add(checklist)
db_session.commit() db_session.commit()
response = client.delete(f"/api/checklists/{checklist.id}", headers=auth_headers) response = client.delete(
f"/api/checklists/{checklist.id}", headers=auth_headers
)
assert response.status_code == 200 assert response.status_code == 200
data = response.get_json() data = response.get_json()
@ -237,7 +253,9 @@ class TestChecklistRoutes:
assert response.status_code == 404 assert response.status_code == 404
def test_delete_check_item_success(self, client, db_session, regular_user, auth_headers): def test_delete_check_item_success(
self, client, db_session, regular_user, auth_headers
):
"""Test deleting a check item""" """Test deleting a check item"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -255,7 +273,9 @@ class TestChecklistRoutes:
db_session.add(checklist) db_session.add(checklist)
db_session.flush() db_session.flush()
item = CheckItem(name="Task", checklist_id=checklist.id, pos=0, state="incomplete") item = CheckItem(
name="Task", checklist_id=checklist.id, pos=0, state="incomplete"
)
db_session.add(item) db_session.add(item)
db_session.commit() db_session.commit()
@ -273,4 +293,4 @@ class TestChecklistRoutes:
"""Test deleting a non-existent check item""" """Test deleting a non-existent check item"""
response = client.delete("/api/check-items/99999", headers=auth_headers) response = client.delete("/api/check-items/99999", headers=auth_headers)
assert response.status_code == 404 assert response.status_code == 404

View file

@ -1,14 +1,16 @@
import pytest import pytest
from app import db from app import db
from app.models import Board, List, Card, Comment from app.models import Board, Card, Comment, List
@pytest.mark.integration @pytest.mark.integration
class TestCommentRoutes: class TestCommentRoutes:
"""Test Comment API routes""" """Test Comment API routes"""
def test_get_card_comments_success(self, client, db_session, regular_user, auth_headers): def test_get_card_comments_success(
self, client, db_session, regular_user, auth_headers
):
"""Test getting all comments for a card""" """Test getting all comments for a card"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -53,7 +55,9 @@ class TestCommentRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_create_comment_success(self, client, db_session, regular_user, auth_headers): def test_create_comment_success(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a new comment on a card""" """Test creating a new comment on a card"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -77,9 +81,11 @@ class TestCommentRoutes:
data = response.get_json() data = response.get_json()
assert data["text"] == "This is a comment" assert data["text"] == "This is a comment"
assert data["card_id"] == card.id assert data["card_id"] == card.id
assert data["user_id"] == 1 assert data["user_id"] == regular_user.id
def test_create_comment_missing_text(self, client, db_session, regular_user, auth_headers): def test_create_comment_missing_text(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a comment without text""" """Test creating a comment without text"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -122,9 +128,11 @@ class TestCommentRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_update_comment_success(self, client, db_session, auth_headers): def test_update_comment_success(
self, client, db_session, auth_headers, regular_user
):
"""Test updating a comment (by author)""" """Test updating a comment (by author)"""
board = Board(name="Test Board", user_id=1) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
db_session.flush() db_session.flush()
@ -136,7 +144,7 @@ class TestCommentRoutes:
db_session.add(card) db_session.add(card)
db_session.flush() db_session.flush()
comment = Comment(text="Original", card_id=card.id, user_id=1) comment = Comment(text="Original", card_id=card.id, user_id=regular_user.id)
db_session.add(comment) db_session.add(comment)
db_session.commit() db_session.commit()
@ -160,7 +168,9 @@ class TestCommentRoutes:
assert response.status_code == 404 assert response.status_code == 404
def test_delete_comment_success(self, client, db_session, regular_user, auth_headers): def test_delete_comment_success(
self, client, db_session, regular_user, auth_headers
):
"""Test deleting a comment""" """Test deleting a comment"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -174,7 +184,7 @@ class TestCommentRoutes:
db_session.add(card) db_session.add(card)
db_session.flush() db_session.flush()
comment = Comment(text="To Delete", card_id=card.id, user_id=1) comment = Comment(text="To Delete", card_id=card.id, user_id=regular_user.id)
db_session.add(comment) db_session.add(comment)
db_session.commit() db_session.commit()

View file

@ -1,14 +1,16 @@
import pytest import pytest
from app import db from app import db
from app.models import Board, List, Card, Label, CardLabel from app.models import Board, Card, CardLabel, Label, List
@pytest.mark.integration @pytest.mark.integration
class TestLabelRoutes: class TestLabelRoutes:
"""Test Label API routes""" """Test Label API routes"""
def test_get_board_labels_success(self, client, db_session, regular_user, auth_headers): def test_get_board_labels_success(
self, client, db_session, regular_user, auth_headers
):
"""Test getting all labels for a board""" """Test getting all labels for a board"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -55,7 +57,9 @@ class TestLabelRoutes:
assert data["color"] == "red" assert data["color"] == "red"
assert data["board_id"] == board.id assert data["board_id"] == board.id
def test_create_label_missing_name(self, client, db_session, regular_user, auth_headers): def test_create_label_missing_name(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a label without name""" """Test creating a label without name"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -82,7 +86,9 @@ class TestLabelRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_add_label_to_card_success(self, client, db_session, regular_user, auth_headers): def test_add_label_to_card_success(
self, client, db_session, regular_user, auth_headers
):
"""Test adding a label to a card""" """Test adding a label to a card"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -136,7 +142,9 @@ class TestLabelRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_remove_label_from_card_success(self, client, db_session, regular_user, auth_headers): def test_remove_label_from_card_success(
self, client, db_session, regular_user, auth_headers
):
"""Test removing a label from a card""" """Test removing a label from a card"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -173,4 +181,4 @@ class TestLabelRoutes:
.filter_by(card_id=card.id, label_id=label.id) .filter_by(card_id=card.id, label_id=label.id)
.first() .first()
) )
assert deleted_card_label is None assert deleted_card_label is None

View file

@ -1,7 +1,7 @@
import pytest import pytest
from app import db from app import db
from app.models import Board, List, Card from app.models import Board, Card, List
@pytest.mark.integration @pytest.mark.integration
@ -26,7 +26,9 @@ class TestListRoutes:
assert data["name"] == "To Do" assert data["name"] == "To Do"
assert data["board_id"] == board.id assert data["board_id"] == board.id
def test_create_list_missing_name(self, client, db_session, regular_user, auth_headers): def test_create_list_missing_name(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a list without name""" """Test creating a list without name"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -53,7 +55,9 @@ class TestListRoutes:
assert response.status_code == 401 assert response.status_code == 401
def test_create_list_board_not_found(self, client, db_session, regular_user, auth_headers): def test_create_list_board_not_found(
self, client, db_session, regular_user, auth_headers
):
"""Test creating a list in a non-existent board""" """Test creating a list in a non-existent board"""
response = client.post( response = client.post(
"/api/boards/99999/lists", "/api/boards/99999/lists",
@ -120,7 +124,9 @@ class TestListRoutes:
assert response.status_code == 404 assert response.status_code == 404
def test_delete_list_with_cards(self, client, db_session, regular_user, auth_headers): def test_delete_list_with_cards(
self, client, db_session, regular_user, auth_headers
):
"""Test deleting a list with cards (cascading delete)""" """Test deleting a list with cards (cascading delete)"""
board = Board(name="Test Board", user_id=regular_user.id) board = Board(name="Test Board", user_id=regular_user.id)
db_session.add(board) db_session.add(board)
@ -142,4 +148,4 @@ class TestListRoutes:
deleted_list = db.session.get(List, lst.id) deleted_list = db.session.get(List, lst.id)
deleted_card = db.session.get(Card, card.id) deleted_card = db.session.get(Card, card.id)
assert deleted_list is None assert deleted_list is None
assert deleted_card is None assert deleted_card is None

View file

@ -13,8 +13,8 @@ class TestUserModel:
def test_user_creation(self, db_session): def test_user_creation(self, db_session):
"""Test creating a user""" """Test creating a user"""
user = User( user = User(
email="test@example.com", email="test2@example.com",
username="testuser", username="testuser2",
first_name="Test", first_name="Test",
last_name="User", last_name="User",
is_admin=False, is_admin=False,
@ -25,15 +25,15 @@ class TestUserModel:
db_session.commit() db_session.commit()
assert user.id is not None assert user.id is not None
assert user.email == "test@example.com" assert user.email == "test2@example.com"
assert user.username == "testuser" assert user.username == "testuser2"
assert user.first_name == "Test" assert user.first_name == "Test"
assert user.last_name == "User" assert user.last_name == "User"
@pytest.mark.unit @pytest.mark.unit
def test_user_password_hashing(self, db_session): def test_user_password_hashing(self, db_session):
"""Test password hashing and verification""" """Test password hashing and verification"""
user = User(email="test@example.com", username="testuser") user = User(email="test1@example.com", username="testuser1")
user.set_password("password123") user.set_password("password123")
db_session.add(user) db_session.add(user)
db_session.commit() db_session.commit()
@ -63,12 +63,12 @@ class TestUserModel:
@pytest.mark.unit @pytest.mark.unit
def test_user_repr(self, db_session): def test_user_repr(self, db_session):
"""Test user string representation""" """Test user string representation"""
user = User(email="test@example.com", username="testuser") user = User(email="test3@example.com", username="testuser3")
user.set_password("password123") user.set_password("password123")
db_session.add(user) db_session.add(user)
db_session.commit() db_session.commit()
assert repr(user) == "<User testuser>" assert repr(user) == "<User testuser3>"
class TestProductModel: class TestProductModel:

View file

@ -118,22 +118,28 @@ class TestProductRoutes:
"""Test product routes""" """Test product routes"""
@pytest.mark.product @pytest.mark.product
def test_get_products(self, client, products): def test_get_products(self, app, client, products):
"""Test getting all products""" """Test getting all products"""
from app.models import Product
before_count = Product.query.count()
response = client.get("/api/products") response = client.get("/api/products")
assert response.status_code == 200 assert response.status_code == 200
data = response.get_json() data = response.get_json()
assert len(data) == 5 assert len(data) == before_count
@pytest.mark.product @pytest.mark.product
def test_get_products_empty(self, client): def test_get_products_empty(self, client):
"""Test getting products when none exist""" """Test getting products when none exist"""
from app.models import Product
before_count = Product.query.count()
response = client.get("/api/products") response = client.get("/api/products")
assert response.status_code == 200 assert response.status_code == 200
data = response.get_json() data = response.get_json()
assert len(data) == 0 assert len(data) == before_count
@pytest.mark.product @pytest.mark.product
def test_get_single_product(self, client, product): def test_get_single_product(self, client, product):