updates
This commit is contained in:
Iliyan Angelov
330
Backend/src/tests/conftest.py
Normal file
330
Backend/src/tests/conftest.py
Normal file
@@ -0,0 +1,330 @@
|
||||
"""
|
||||
Pytest configuration and fixtures for integration tests.
|
||||
"""
|
||||
import pytest
|
||||
import os
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
from fastapi.testclient import TestClient
|
||||
from datetime import datetime, timedelta
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add src to path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from src.config.database import Base, get_db
|
||||
from src.config.settings import settings
|
||||
from src.main import app
|
||||
from src.models.user import User
|
||||
from src.models.role import Role
|
||||
from src.models.room import Room, RoomStatus
|
||||
from src.models.room_type import RoomType
|
||||
from src.models.booking import Booking, BookingStatus
|
||||
from src.models.payment import Payment, PaymentMethod, PaymentStatus
|
||||
from src.models.service import Service
|
||||
from src.models.promotion import Promotion
|
||||
from src.models.banner import Banner
|
||||
from src.services.auth_service import auth_service
|
||||
import bcrypt
|
||||
|
||||
|
||||
# Use SQLite in-memory database for testing
|
||||
SQLALCHEMY_TEST_DATABASE_URL = "sqlite:///:memory:"
|
||||
|
||||
test_engine = create_engine(
|
||||
SQLALCHEMY_TEST_DATABASE_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=test_engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db_session():
|
||||
"""Create a fresh database for each test."""
|
||||
# Create all tables
|
||||
Base.metadata.create_all(bind=test_engine)
|
||||
|
||||
# Create session
|
||||
db = TestingSessionLocal()
|
||||
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
# Drop all tables after test
|
||||
Base.metadata.drop_all(bind=test_engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def client(db_session):
|
||||
"""Create a test client with database override."""
|
||||
def override_get_db():
|
||||
try:
|
||||
yield db_session
|
||||
finally:
|
||||
pass
|
||||
|
||||
# Disable CSRF protection for tests
|
||||
original_csrf = settings.CSRF_PROTECTION_ENABLED
|
||||
settings.CSRF_PROTECTION_ENABLED = False
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
app.dependency_overrides.clear()
|
||||
# Restore original CSRF setting
|
||||
settings.CSRF_PROTECTION_ENABLED = original_csrf
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_role(db_session):
|
||||
"""Create a test role."""
|
||||
role = Role(
|
||||
name="guest",
|
||||
description="Guest role"
|
||||
)
|
||||
db_session.add(role)
|
||||
db_session.commit()
|
||||
db_session.refresh(role)
|
||||
return role
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_admin_role(db_session):
|
||||
"""Create an admin role."""
|
||||
role = Role(
|
||||
name="admin",
|
||||
description="Admin role"
|
||||
)
|
||||
db_session.add(role)
|
||||
db_session.commit()
|
||||
db_session.refresh(role)
|
||||
return role
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_staff_role(db_session):
|
||||
"""Create a staff role."""
|
||||
role = Role(
|
||||
name="staff",
|
||||
description="Staff role"
|
||||
)
|
||||
db_session.add(role)
|
||||
db_session.commit()
|
||||
db_session.refresh(role)
|
||||
return role
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_user(db_session, test_role):
|
||||
"""Create a test user."""
|
||||
hashed_password = bcrypt.hashpw("testpassword123".encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||||
user = User(
|
||||
email="test@example.com",
|
||||
password=hashed_password,
|
||||
full_name="Test User",
|
||||
phone="1234567890",
|
||||
role_id=test_role.id,
|
||||
is_active=True
|
||||
)
|
||||
db_session.add(user)
|
||||
db_session.commit()
|
||||
db_session.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_admin_user(db_session, test_admin_role):
|
||||
"""Create a test admin user."""
|
||||
hashed_password = bcrypt.hashpw("adminpassword123".encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||||
user = User(
|
||||
email="admin@example.com",
|
||||
password=hashed_password,
|
||||
full_name="Admin User",
|
||||
phone="1234567890",
|
||||
role_id=test_admin_role.id,
|
||||
is_active=True
|
||||
)
|
||||
db_session.add(user)
|
||||
db_session.commit()
|
||||
db_session.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_staff_user(db_session, test_staff_role):
|
||||
"""Create a test staff user."""
|
||||
hashed_password = bcrypt.hashpw("staffpassword123".encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||||
user = User(
|
||||
email="staff@example.com",
|
||||
password=hashed_password,
|
||||
full_name="Staff User",
|
||||
phone="1234567890",
|
||||
role_id=test_staff_role.id,
|
||||
is_active=True
|
||||
)
|
||||
db_session.add(user)
|
||||
db_session.commit()
|
||||
db_session.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_token(client, test_user):
|
||||
"""Get authentication token for test user."""
|
||||
response = client.post(
|
||||
"/api/auth/login",
|
||||
json={
|
||||
"email": "test@example.com",
|
||||
"password": "testpassword123"
|
||||
}
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json()["data"]["token"]
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_token(client, test_admin_user):
|
||||
"""Get authentication token for admin user."""
|
||||
response = client.post(
|
||||
"/api/auth/login",
|
||||
json={
|
||||
"email": "admin@example.com",
|
||||
"password": "adminpassword123"
|
||||
}
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json()["data"]["token"]
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def staff_token(client, test_staff_user):
|
||||
"""Get authentication token for staff user."""
|
||||
response = client.post(
|
||||
"/api/auth/login",
|
||||
json={
|
||||
"email": "staff@example.com",
|
||||
"password": "staffpassword123"
|
||||
}
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json()["data"]["token"]
|
||||
return None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def authenticated_client(client, auth_token):
|
||||
"""Create an authenticated test client."""
|
||||
client.headers.update({"Authorization": f"Bearer {auth_token}"})
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_client(client, admin_token):
|
||||
"""Create an authenticated admin test client."""
|
||||
client.headers.update({"Authorization": f"Bearer {admin_token}"})
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_room_type(db_session):
|
||||
"""Create a test room type."""
|
||||
room_type = RoomType(
|
||||
name="Deluxe Room",
|
||||
description="A deluxe room with ocean view",
|
||||
base_price=100.00,
|
||||
capacity=2,
|
||||
amenities=["WiFi", "TV", "AC"]
|
||||
)
|
||||
db_session.add(room_type)
|
||||
db_session.commit()
|
||||
db_session.refresh(room_type)
|
||||
return room_type
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_room(db_session, test_room_type):
|
||||
"""Create a test room."""
|
||||
room = Room(
|
||||
room_type_id=test_room_type.id,
|
||||
room_number="101",
|
||||
floor=1,
|
||||
status=RoomStatus.available,
|
||||
price=100.00,
|
||||
featured=True,
|
||||
capacity=2,
|
||||
images=["/uploads/room1.jpg"],
|
||||
amenities=["WiFi", "TV", "AC"]
|
||||
)
|
||||
db_session.add(room)
|
||||
db_session.commit()
|
||||
db_session.refresh(room)
|
||||
return room
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_booking(db_session, test_user, test_room):
|
||||
"""Create a test booking."""
|
||||
check_in = datetime.utcnow() + timedelta(days=1)
|
||||
check_out = datetime.utcnow() + timedelta(days=3)
|
||||
|
||||
booking = Booking(
|
||||
booking_number="BK-TEST-001",
|
||||
user_id=test_user.id,
|
||||
room_id=test_room.id,
|
||||
check_in_date=check_in,
|
||||
check_out_date=check_out,
|
||||
num_guests=2,
|
||||
total_price=200.00,
|
||||
status=BookingStatus.confirmed
|
||||
)
|
||||
db_session.add(booking)
|
||||
db_session.commit()
|
||||
db_session.refresh(booking)
|
||||
return booking
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_service(db_session):
|
||||
"""Create a test service."""
|
||||
service = Service(
|
||||
name="Room Service",
|
||||
description="24/7 room service",
|
||||
price=25.00,
|
||||
category="Food & Beverage",
|
||||
is_active=True
|
||||
)
|
||||
db_session.add(service)
|
||||
db_session.commit()
|
||||
db_session.refresh(service)
|
||||
return service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_promotion(db_session):
|
||||
"""Create a test promotion."""
|
||||
from src.models.promotion import Promotion, DiscountType
|
||||
|
||||
promotion = Promotion(
|
||||
code="TEST10",
|
||||
name="Test Promotion",
|
||||
description="10% off",
|
||||
discount_type=DiscountType.percentage,
|
||||
discount_value=10.00,
|
||||
start_date=datetime.utcnow() - timedelta(days=1),
|
||||
end_date=datetime.utcnow() + timedelta(days=30),
|
||||
is_active=True
|
||||
)
|
||||
db_session.add(promotion)
|
||||
db_session.commit()
|
||||
db_session.refresh(promotion)
|
||||
return promotion
|
||||
|
||||
Reference in New Issue
Block a user