257 lines
9.6 KiB
Python
257 lines
9.6 KiB
Python
from datetime import timedelta
|
|
|
|
from django.contrib.admin.sites import AdminSite
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.models import AnonymousUser
|
|
from django.core.exceptions import ValidationError
|
|
from django.test import RequestFactory, TestCase
|
|
from django.utils import timezone
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
|
|
from accounts.models import VendorProfile
|
|
from adventrues.models import AdventureCategory, AdventureOffering
|
|
from equipment.models import EquipmentCategory, EquipmentItem
|
|
|
|
from .admin import BookingAdmin, mark_approved
|
|
from .models import AvailabilitySlot, Booking
|
|
from .services import is_bookable, quote_booking, transition_booking_status
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class BookingServiceTests(TestCase):
|
|
def setUp(self):
|
|
self.vendor_user = User.objects.create_user(
|
|
email="vendor-booking@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=True,
|
|
is_customer=False,
|
|
)
|
|
self.vendor_profile = VendorProfile.objects.create(user=self.vendor_user, business_name="Wave Rentals")
|
|
self.customer = User.objects.create_user(
|
|
email="customer-booking@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=False,
|
|
is_customer=True,
|
|
)
|
|
category = EquipmentCategory.objects.create(name="Jet Ski", slug="jetski")
|
|
self.item = EquipmentItem.objects.create(
|
|
vendor=self.vendor_profile,
|
|
category=category,
|
|
title="Jet Ski 1",
|
|
public_id="jet-001",
|
|
price_per_day="120.00",
|
|
is_active=True,
|
|
)
|
|
adv_category = AdventureCategory.objects.create(name="Tour", slug="tour")
|
|
self.offering = AdventureOffering.objects.create(
|
|
vendor=self.vendor_profile,
|
|
category=adv_category,
|
|
title="Harbor Tour",
|
|
public_id="tour-001",
|
|
duration_minutes=90,
|
|
capacity=6,
|
|
price_per_person="60.00",
|
|
is_active=True,
|
|
)
|
|
|
|
def test_availability_overlap_validation(self):
|
|
starts = timezone.now() + timedelta(days=1)
|
|
ends = starts + timedelta(days=1)
|
|
Booking.objects.create(
|
|
customer=self.customer,
|
|
vendor=self.vendor_profile,
|
|
equipment_item=self.item,
|
|
starts_at=starts,
|
|
ends_at=ends,
|
|
status=Booking.Status.APPROVED,
|
|
total_price="120.00",
|
|
)
|
|
self.assertFalse(
|
|
is_bookable(
|
|
equipment_item=self.item,
|
|
starts_at=starts + timedelta(hours=1),
|
|
ends_at=ends + timedelta(hours=1),
|
|
)
|
|
)
|
|
|
|
def test_booking_quote_and_state_transition_validation(self):
|
|
starts = timezone.now() + timedelta(days=2)
|
|
ends = starts + timedelta(days=2)
|
|
quote = quote_booking(equipment_item=self.item, starts_at=starts, ends_at=ends)
|
|
self.assertEqual(str(quote), "240.00")
|
|
|
|
booking = Booking.objects.create(
|
|
customer=self.customer,
|
|
vendor=self.vendor_profile,
|
|
equipment_item=self.item,
|
|
starts_at=starts,
|
|
ends_at=ends,
|
|
status=Booking.Status.REQUESTED,
|
|
total_price=quote,
|
|
)
|
|
transition_booking_status(booking=booking, to_status=Booking.Status.APPROVED, actor=self.vendor_user, note="ok")
|
|
booking.refresh_from_db()
|
|
self.assertEqual(booking.status, Booking.Status.APPROVED)
|
|
|
|
with self.assertRaises(ValidationError):
|
|
transition_booking_status(booking=booking, to_status=Booking.Status.DECLINED, actor=self.vendor_user)
|
|
|
|
def test_slot_rule_blocks_when_unavailable_slot_overlaps(self):
|
|
starts = timezone.now() + timedelta(days=3)
|
|
ends = starts + timedelta(hours=4)
|
|
AvailabilitySlot.objects.create(
|
|
equipment_item=self.item,
|
|
starts_at=starts - timedelta(hours=1),
|
|
ends_at=ends + timedelta(hours=1),
|
|
is_available=False,
|
|
)
|
|
self.assertFalse(is_bookable(equipment_item=self.item, starts_at=starts, ends_at=ends))
|
|
|
|
|
|
class BookingApiTests(APITestCase):
|
|
def setUp(self):
|
|
self.vendor_user = User.objects.create_user(
|
|
email="vendor-api@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=True,
|
|
is_customer=False,
|
|
)
|
|
self.vendor_profile = VendorProfile.objects.create(user=self.vendor_user, business_name="Vendor API")
|
|
self.customer = User.objects.create_user(
|
|
email="customer-api@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=False,
|
|
is_customer=True,
|
|
)
|
|
self.other_customer = User.objects.create_user(
|
|
email="other@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=False,
|
|
is_customer=True,
|
|
)
|
|
category = EquipmentCategory.objects.create(name="Board", slug="board")
|
|
self.item = EquipmentItem.objects.create(
|
|
vendor=self.vendor_profile,
|
|
category=category,
|
|
title="Board 1",
|
|
public_id="board-001",
|
|
price_per_day="80.00",
|
|
is_active=True,
|
|
)
|
|
|
|
def test_booking_request_happy_path_and_overlap_rejection(self):
|
|
starts = timezone.now() + timedelta(days=5)
|
|
ends = starts + timedelta(days=1)
|
|
self.client.force_authenticate(self.customer)
|
|
first = self.client.post(
|
|
"/api/v1/booking/bookings/request/",
|
|
{
|
|
"equipment_item_id": self.item.id,
|
|
"starts_at": starts.isoformat(),
|
|
"ends_at": ends.isoformat(),
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(first.status_code, status.HTTP_201_CREATED)
|
|
|
|
overlap = self.client.post(
|
|
"/api/v1/booking/bookings/request/",
|
|
{
|
|
"equipment_item_id": self.item.id,
|
|
"starts_at": (starts + timedelta(hours=1)).isoformat(),
|
|
"ends_at": (ends + timedelta(hours=1)).isoformat(),
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(overlap.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_booking_invalid_range_rejected(self):
|
|
starts = timezone.now() + timedelta(days=2)
|
|
self.client.force_authenticate(self.customer)
|
|
res = self.client.post(
|
|
"/api/v1/booking/bookings/request/",
|
|
{
|
|
"equipment_item_id": self.item.id,
|
|
"starts_at": (starts + timedelta(hours=2)).isoformat(),
|
|
"ends_at": starts.isoformat(),
|
|
},
|
|
format="json",
|
|
)
|
|
self.assertEqual(res.status_code, status.HTTP_400_BAD_REQUEST)
|
|
|
|
def test_vendor_approve_and_decline_permissions(self):
|
|
starts = timezone.now() + timedelta(days=6)
|
|
ends = starts + timedelta(days=1)
|
|
booking = Booking.objects.create(
|
|
customer=self.customer,
|
|
vendor=self.vendor_profile,
|
|
equipment_item=self.item,
|
|
starts_at=starts,
|
|
ends_at=ends,
|
|
status=Booking.Status.REQUESTED,
|
|
total_price="80.00",
|
|
)
|
|
|
|
self.client.force_authenticate(self.other_customer)
|
|
forbidden = self.client.post(f"/api/v1/booking/bookings/{booking.id}/approve/", {}, format="json")
|
|
self.assertEqual(forbidden.status_code, status.HTTP_404_NOT_FOUND)
|
|
|
|
self.client.force_authenticate(self.vendor_user)
|
|
approved = self.client.post(f"/api/v1/booking/bookings/{booking.id}/approve/", {}, format="json")
|
|
self.assertEqual(approved.status_code, status.HTTP_200_OK)
|
|
booking.refresh_from_db()
|
|
self.assertEqual(booking.status, Booking.Status.APPROVED)
|
|
|
|
|
|
class BookingAdminTests(TestCase):
|
|
def setUp(self):
|
|
self.site = AdminSite()
|
|
self.factory = RequestFactory()
|
|
self.admin_user = User.objects.create_superuser(email="admin@example.com", password="Pass123456!")
|
|
vendor_user = User.objects.create_user(
|
|
email="vendor-admin@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=True,
|
|
is_customer=False,
|
|
)
|
|
self.vendor_profile = VendorProfile.objects.create(user=vendor_user, business_name="Admin Vendor")
|
|
customer = User.objects.create_user(
|
|
email="customer-admin@example.com",
|
|
password="Pass123456!",
|
|
is_vendor=False,
|
|
is_customer=True,
|
|
)
|
|
category = EquipmentCategory.objects.create(name="Sail", slug="sail")
|
|
item = EquipmentItem.objects.create(
|
|
vendor=self.vendor_profile,
|
|
category=category,
|
|
title="Sail Boat",
|
|
public_id="sail-001",
|
|
price_per_day="90.00",
|
|
is_active=True,
|
|
)
|
|
starts = timezone.now() + timedelta(days=7)
|
|
ends = starts + timedelta(days=1)
|
|
self.booking = Booking.objects.create(
|
|
customer=customer,
|
|
vendor=self.vendor_profile,
|
|
equipment_item=item,
|
|
starts_at=starts,
|
|
ends_at=ends,
|
|
status=Booking.Status.REQUESTED,
|
|
total_price="90.00",
|
|
)
|
|
|
|
def test_admin_model_registration_smoke(self):
|
|
admin_obj = BookingAdmin(Booking, self.site)
|
|
self.assertIsNotNone(admin_obj)
|
|
|
|
def test_admin_action_status_update(self):
|
|
request = self.factory.post("/admin/booking/booking/")
|
|
request.user = self.admin_user
|
|
mark_approved(None, request, Booking.objects.filter(id=self.booking.id))
|
|
self.booking.refresh_from_db()
|
|
self.assertEqual(self.booking.status, Booking.Status.APPROVED)
|