474 lines
12 KiB
Python
474 lines
12 KiB
Python
"""
|
|
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.shared.config.database import Base, get_db
|
|
from src.shared.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.auth.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_accountant_role(db_session):
|
|
"""Create an accountant role."""
|
|
role = Role(
|
|
name="accountant",
|
|
description="Accountant role"
|
|
)
|
|
db_session.add(role)
|
|
db_session.commit()
|
|
db_session.refresh(role)
|
|
return role
|
|
|
|
|
|
@pytest.fixture
|
|
def test_housekeeping_role(db_session):
|
|
"""Create a housekeeping role."""
|
|
role = Role(
|
|
name="housekeeping",
|
|
description="Housekeeping 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 test_accountant_user(db_session, test_accountant_role):
|
|
"""Create a test accountant user."""
|
|
hashed_password = bcrypt.hashpw("accountantpassword123".encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
user = User(
|
|
email="accountant@example.com",
|
|
password=hashed_password,
|
|
full_name="Accountant User",
|
|
phone="1234567890",
|
|
role_id=test_accountant_role.id,
|
|
is_active=True
|
|
)
|
|
db_session.add(user)
|
|
db_session.commit()
|
|
db_session.refresh(user)
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def test_housekeeping_user(db_session, test_housekeeping_role):
|
|
"""Create a test housekeeping user."""
|
|
hashed_password = bcrypt.hashpw("housekeepingpassword123".encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
|
user = User(
|
|
email="housekeeping@example.com",
|
|
password=hashed_password,
|
|
full_name="Housekeeping User",
|
|
phone="1234567890",
|
|
role_id=test_housekeeping_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 (from cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "test@example.com",
|
|
"password": "testpassword123"
|
|
}
|
|
)
|
|
if response.status_code == 200:
|
|
# Token is now in httpOnly cookie, return cookie value for testing
|
|
# In real usage, cookies are sent automatically
|
|
cookie_token = response.cookies.get("accessToken")
|
|
return cookie_token
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_token(client, test_admin_user):
|
|
"""Get authentication token for admin user (from cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "admin@example.com",
|
|
"password": "adminpassword123"
|
|
}
|
|
)
|
|
if response.status_code == 200:
|
|
# Token is now in httpOnly cookie
|
|
cookie_token = response.cookies.get("accessToken")
|
|
return cookie_token
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def staff_token(client, test_staff_user):
|
|
"""Get authentication token for staff user (from cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "staff@example.com",
|
|
"password": "staffpassword123"
|
|
}
|
|
)
|
|
if response.status_code == 200:
|
|
# Token is now in httpOnly cookie
|
|
cookie_token = response.cookies.get("accessToken")
|
|
return cookie_token
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def accountant_token(client, test_accountant_user):
|
|
"""Get authentication token for accountant user (from cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "accountant@example.com",
|
|
"password": "accountantpassword123"
|
|
}
|
|
)
|
|
if response.status_code == 200:
|
|
cookie_token = response.cookies.get("accessToken")
|
|
return cookie_token
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def housekeeping_token(client, test_housekeeping_user):
|
|
"""Get authentication token for housekeeping user (from cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "housekeeping@example.com",
|
|
"password": "housekeepingpassword123"
|
|
}
|
|
)
|
|
if response.status_code == 200:
|
|
cookie_token = response.cookies.get("accessToken")
|
|
return cookie_token
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def authenticated_client(client, test_user):
|
|
"""Create an authenticated test client (uses cookies)."""
|
|
# Login to set cookies
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "test@example.com",
|
|
"password": "testpassword123"
|
|
}
|
|
)
|
|
# Cookies are automatically sent with subsequent requests
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_client(client, test_admin_user):
|
|
"""Create an authenticated admin test client (uses cookies)."""
|
|
# Login to set cookies
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "admin@example.com",
|
|
"password": "adminpassword123"
|
|
}
|
|
)
|
|
# Cookies are automatically sent with subsequent requests
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def accountant_client(client, test_accountant_user):
|
|
"""Create an authenticated accountant test client (uses cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "accountant@example.com",
|
|
"password": "accountantpassword123"
|
|
}
|
|
)
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def housekeeping_client(client, test_housekeeping_user):
|
|
"""Create an authenticated housekeeping test client (uses cookies)."""
|
|
response = client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"email": "housekeeping@example.com",
|
|
"password": "housekeepingpassword123"
|
|
}
|
|
)
|
|
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
|
|
|