feat: sistem de linkuri publice pentru proprietăți
Expune fluxul de rezervare anonimă printr-un link public partajabil
(/book/<slug>), cu calendar public, cod QR și proprietate demo publică.
Backend:
- slug + list_on_landing pe Property; utilitar slugify (diacritice RO,
cuvinte rezervate, ban all-digit) + migrare idempotentă raw-DDL rulată
din entrypoint; SQLite WAL + busy_timeout
- GET /public/properties/{slug_or_id}(+/spaces): rezolvare slug/ID, 404 uniform
- GET /public/spaces/{id}/busy: doar {start_time,end_time} (fără PII),
pending+approved, tz-aware/interval>60z → 422, orfan/privat → 404
- availability: scrub user_name/title (fix scurgere PII)
- POST /public/bookings: overlap pending+approved, cap 5/email/spațiu/zi,
rate-limit per-IP, suprimare notificări+email pe demo, email confirmare guest
- PATCH slug (409 duplicat / 422 format / 403 non-manager); guest_email EmailStr
- email guest la aprobare/respingere cu cod #RB-<id>
- teste noi: test_public.py + test_migrate_slug.py (33 teste)
Frontend:
- rută /book/:slug; PublicBooking rescris cu calendar grid manual
(PublicCalendar + GuestBookingForm), 30min, 08-20, mobil day-view, RO
- card „Link public" în PropertyDetail: copy, edit slug, QR pe URL-ul cu ID
- secțiune proprietăți publice + CTA pe Landing; buton copy în Properties
Gate aprobat (U1-U3, G1 grid manual, G2 list_on_landing); G3-G5 în TODOS.md.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
3
backend/.gitignore
vendored
3
backend/.gitignore
vendored
@@ -28,6 +28,9 @@ env/
|
||||
|
||||
# Database
|
||||
*.db
|
||||
*.db-wal
|
||||
*.db-shm
|
||||
*.db-journal
|
||||
*.sqlite
|
||||
*.sqlite3
|
||||
|
||||
|
||||
@@ -7,7 +7,9 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.deps import get_current_admin, get_current_manager_or_superadmin, get_current_user, get_db
|
||||
from app.core.permissions import get_manager_property_ids, verify_property_access
|
||||
from app.core.slug import DEMO_SLUG
|
||||
from app.models.booking import Booking
|
||||
from app.models.property import Property
|
||||
from app.models.property_manager import PropertyManager
|
||||
from app.models.settings import Settings
|
||||
from app.models.space import Space
|
||||
@@ -61,6 +63,18 @@ def _verify_manager_booking_access(db: Session, booking: Booking, current_user:
|
||||
)
|
||||
|
||||
|
||||
def _is_demo_property_space(db: Session, space: Space | None) -> bool:
|
||||
"""Return True if the given space belongs to the demo property.
|
||||
|
||||
Used to suppress guest notification emails for the public demo property
|
||||
(avoids spamming real inboxes from demo/test bookings).
|
||||
"""
|
||||
if space is None or not space.property_id:
|
||||
return False
|
||||
prop = db.query(Property).filter(Property.id == space.property_id).first()
|
||||
return bool(prop and prop.slug == DEMO_SLUG)
|
||||
|
||||
|
||||
def _verify_manager_space_access(db: Session, space: Space, current_user: User) -> None:
|
||||
"""Verify that a manager has access to a space's property.
|
||||
|
||||
@@ -931,15 +945,15 @@ def approve_booking(
|
||||
booking.user.full_name,
|
||||
None,
|
||||
)
|
||||
elif booking.guest_email:
|
||||
# Send email notification to anonymous guest
|
||||
elif booking.guest_email and not _is_demo_property_space(db, booking.space):
|
||||
# Send email notification to anonymous guest (skipped for demo property)
|
||||
background_tasks.add_task(
|
||||
send_booking_notification,
|
||||
booking,
|
||||
"anonymous_approved",
|
||||
booking.guest_email,
|
||||
booking.guest_name or "Guest",
|
||||
None,
|
||||
{"reference_code": f"RB-{booking.id}"},
|
||||
)
|
||||
|
||||
return booking
|
||||
@@ -1018,14 +1032,14 @@ def reject_booking(
|
||||
booking.user.full_name,
|
||||
{"rejection_reason": reject_data.reason},
|
||||
)
|
||||
elif booking.guest_email:
|
||||
elif booking.guest_email and not _is_demo_property_space(db, booking.space):
|
||||
background_tasks.add_task(
|
||||
send_booking_notification,
|
||||
booking,
|
||||
"anonymous_rejected",
|
||||
booking.guest_email,
|
||||
booking.guest_name or "Guest",
|
||||
{"rejection_reason": reject_data.reason},
|
||||
{"rejection_reason": reject_data.reason, "reference_code": f"RB-{booking.id}"},
|
||||
)
|
||||
|
||||
return booking
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.deps import (
|
||||
@@ -12,6 +13,7 @@ from app.core.deps import (
|
||||
get_optional_user,
|
||||
)
|
||||
from app.core.permissions import get_manager_property_ids, verify_property_access
|
||||
from app.core.slug import generate_unique_slug, validate_slug
|
||||
from app.models.organization import Organization
|
||||
from app.models.property import Property
|
||||
from app.models.property_access import PropertyAccess
|
||||
@@ -125,6 +127,8 @@ def list_properties(
|
||||
created_at=p.created_at,
|
||||
space_count=space_count,
|
||||
managers=_get_property_managers(db, p.id),
|
||||
slug=p.slug,
|
||||
list_on_landing=p.list_on_landing,
|
||||
))
|
||||
return result
|
||||
|
||||
@@ -141,6 +145,8 @@ def get_property(
|
||||
spaces = db.query(Space).filter(Space.property_id == property_id, Space.is_active == True).all() # noqa: E712
|
||||
space_count = len(spaces)
|
||||
|
||||
settings_row = db.query(PropertySettings).filter(PropertySettings.property_id == property_id).first()
|
||||
|
||||
return PropertyWithSpaces(
|
||||
id=prop.id,
|
||||
name=prop.name,
|
||||
@@ -152,6 +158,9 @@ def get_property(
|
||||
space_count=space_count,
|
||||
managers=_get_property_managers(db, prop.id),
|
||||
spaces=[SpaceResponse.model_validate(s) for s in spaces],
|
||||
slug=prop.slug,
|
||||
list_on_landing=prop.list_on_landing,
|
||||
require_approval=settings_row.require_approval if settings_row else None,
|
||||
)
|
||||
|
||||
|
||||
@@ -190,8 +199,11 @@ def create_property(
|
||||
description=data.description,
|
||||
address=data.address,
|
||||
is_public=data.is_public,
|
||||
list_on_landing=data.list_on_landing,
|
||||
)
|
||||
db.add(prop)
|
||||
db.flush()
|
||||
prop.slug = generate_unique_slug(db, data.name, exclude_id=prop.id)
|
||||
db.commit()
|
||||
db.refresh(prop)
|
||||
|
||||
@@ -219,6 +231,8 @@ def create_property(
|
||||
created_at=prop.created_at,
|
||||
space_count=0,
|
||||
managers=_get_property_managers(db, prop.id),
|
||||
slug=prop.slug,
|
||||
list_on_landing=prop.list_on_landing,
|
||||
)
|
||||
|
||||
|
||||
@@ -241,10 +255,37 @@ def update_property(
|
||||
prop.address = data.address
|
||||
if data.is_public is not None:
|
||||
prop.is_public = data.is_public
|
||||
if data.list_on_landing is not None:
|
||||
prop.list_on_landing = data.list_on_landing
|
||||
|
||||
db.commit()
|
||||
slug_changed = False
|
||||
old_slug = prop.slug
|
||||
if data.slug is not None and data.slug != prop.slug:
|
||||
if not validate_slug(data.slug):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail="Slug invalid. Folosește doar litere mici, cifre și cratime (3-64 caractere).",
|
||||
)
|
||||
prop.slug = data.slug
|
||||
slug_changed = True
|
||||
|
||||
try:
|
||||
db.commit()
|
||||
except IntegrityError:
|
||||
db.rollback()
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Slug deja folosit")
|
||||
db.refresh(prop)
|
||||
|
||||
if slug_changed:
|
||||
log_action(
|
||||
db=db,
|
||||
action="property_slug_changed",
|
||||
user_id=current_user.id,
|
||||
target_type="property",
|
||||
target_id=prop.id,
|
||||
details={"old": old_slug, "new": prop.slug},
|
||||
)
|
||||
|
||||
space_count = db.query(Space).filter(Space.property_id == prop.id, Space.is_active == True).count() # noqa: E712
|
||||
return PropertyResponse(
|
||||
id=prop.id,
|
||||
@@ -256,6 +297,8 @@ def update_property(
|
||||
created_at=prop.created_at,
|
||||
space_count=space_count,
|
||||
managers=_get_property_managers(db, prop.id),
|
||||
slug=prop.slug,
|
||||
list_on_landing=prop.list_on_landing,
|
||||
)
|
||||
|
||||
|
||||
@@ -284,6 +327,8 @@ def update_property_status(
|
||||
created_at=prop.created_at,
|
||||
space_count=space_count,
|
||||
managers=_get_property_managers(db, prop.id),
|
||||
slug=prop.slug,
|
||||
list_on_landing=prop.list_on_landing,
|
||||
)
|
||||
|
||||
|
||||
@@ -517,6 +562,8 @@ def admin_list_all_properties(
|
||||
created_at=p.created_at,
|
||||
space_count=space_count,
|
||||
managers=_get_property_managers(db, p.id),
|
||||
slug=p.slug,
|
||||
list_on_landing=p.list_on_landing,
|
||||
))
|
||||
return result
|
||||
|
||||
|
||||
@@ -1,27 +1,169 @@
|
||||
"""Public/anonymous endpoints (no auth required)."""
|
||||
from datetime import datetime
|
||||
import os
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, time, timedelta
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, status
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy import and_, or_
|
||||
from sqlalchemy.orm import Query as SAQuery
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.deps import get_db
|
||||
from app.core.slug import DEMO_SLUG
|
||||
from app.models.booking import Booking
|
||||
from app.models.property import Property
|
||||
from app.models.property_manager import PropertyManager
|
||||
from app.models.property_settings import PropertySettings
|
||||
from app.models.space import Space
|
||||
from app.models.user import User
|
||||
from app.schemas.booking import AnonymousBookingCreate, AvailabilityCheck, BookingResponse, ConflictingBooking
|
||||
from app.schemas.booking import (
|
||||
AnonymousBookingCreate,
|
||||
AvailabilityCheck,
|
||||
BookingResponse,
|
||||
ConflictingBooking,
|
||||
)
|
||||
from app.schemas.property import PropertyResponse
|
||||
from app.schemas.space import SpaceResponse
|
||||
from app.services.booking_service import validate_booking_rules
|
||||
from app.services.email_service import send_booking_notification
|
||||
from app.services.notification_service import create_notification
|
||||
|
||||
router = APIRouter(prefix="/public", tags=["public"])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def overlapping_bookings_query(
|
||||
db: Session,
|
||||
space_id: int,
|
||||
start: datetime,
|
||||
end: datetime,
|
||||
statuses: tuple[str, ...] = ("approved", "pending"),
|
||||
) -> SAQuery:
|
||||
"""Shared overlap predicate: any booking on `space_id` whose interval
|
||||
intersects [start, end), restricted to `statuses`."""
|
||||
return db.query(Booking).filter(
|
||||
Booking.space_id == space_id,
|
||||
Booking.status.in_(statuses),
|
||||
or_(
|
||||
and_(
|
||||
Booking.start_datetime <= start,
|
||||
Booking.end_datetime > start,
|
||||
),
|
||||
and_(
|
||||
Booking.start_datetime < end,
|
||||
Booking.end_datetime >= end,
|
||||
),
|
||||
and_(
|
||||
Booking.start_datetime >= start,
|
||||
Booking.end_datetime <= end,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_property(db: Session, slug_or_id: str) -> Property | None:
|
||||
"""Resolve a property by numeric id (all-digit segment) or by slug."""
|
||||
if slug_or_id.isdigit():
|
||||
return db.query(Property).filter(Property.id == int(slug_or_id)).first()
|
||||
return db.query(Property).filter(Property.slug == slug_or_id).first()
|
||||
|
||||
|
||||
def _is_visible(prop: Property | None) -> bool:
|
||||
"""Uniform visibility rule for public endpoints: must exist, be public,
|
||||
and be active. Callers turn a False result into a bare 404 (no 403
|
||||
oracle that would let clients distinguish private-vs-missing)."""
|
||||
return bool(prop and prop.is_public and prop.is_active)
|
||||
|
||||
|
||||
def _require_approval_for(db: Session, property_id: int) -> bool | None:
|
||||
settings_row = db.query(PropertySettings).filter(PropertySettings.property_id == property_id).first()
|
||||
return settings_row.require_approval if settings_row else None
|
||||
|
||||
|
||||
def _property_to_response(db: Session, prop: Property) -> PropertyResponse:
|
||||
space_count = db.query(Space).filter(Space.property_id == prop.id, Space.is_active == True).count() # noqa: E712
|
||||
return PropertyResponse(
|
||||
id=prop.id,
|
||||
name=prop.name,
|
||||
description=prop.description,
|
||||
address=prop.address,
|
||||
is_public=prop.is_public,
|
||||
is_active=prop.is_active,
|
||||
created_at=prop.created_at,
|
||||
space_count=space_count,
|
||||
slug=prop.slug,
|
||||
list_on_landing=prop.list_on_landing,
|
||||
require_approval=_require_approval_for(db, prop.id),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_visible_space_property(db: Session, space: Space) -> Property | None:
|
||||
"""Return the space's property iff it exists, is public, and active.
|
||||
Returns None (caller raises 404) for orphan spaces or private/inactive
|
||||
properties -- uniform 404, no 403 oracle."""
|
||||
if not space.property_id:
|
||||
return None
|
||||
prop = db.query(Property).filter(Property.id == space.property_id).first()
|
||||
if not _is_visible(prop):
|
||||
return None
|
||||
return prop
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-process per-IP rate limiter for anonymous booking creation.
|
||||
#
|
||||
# A simple sliding-window counter is used instead of pulling in slowapi:
|
||||
# it needs no app-level wiring (this module cannot touch app/main.py under
|
||||
# the current work split) and is trivially bypassed in tests via the
|
||||
# PYTEST_CURRENT_TEST env var pytest sets automatically, or DISABLE_RATE_LIMIT.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_RATE_LIMIT_WINDOW = timedelta(hours=1)
|
||||
_RATE_LIMIT_MAX = 5
|
||||
_rate_limit_buckets: dict[str, list[datetime]] = defaultdict(list)
|
||||
|
||||
|
||||
def _rate_limiting_enabled() -> bool:
|
||||
if os.getenv("PYTEST_CURRENT_TEST"):
|
||||
return False
|
||||
if os.getenv("DISABLE_RATE_LIMIT"):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _enforce_booking_rate_limit(request: Request) -> None:
|
||||
if not _rate_limiting_enabled():
|
||||
return
|
||||
ip = request.client.host if request.client else "unknown"
|
||||
now = datetime.utcnow()
|
||||
bucket = _rate_limit_buckets[ip]
|
||||
bucket[:] = [t for t in bucket if now - t < _RATE_LIMIT_WINDOW]
|
||||
if len(bucket) >= _RATE_LIMIT_MAX:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
||||
detail="Prea multe cereri de rezervare de la această adresă IP. Încearcă din nou peste o oră.",
|
||||
)
|
||||
bucket.append(now)
|
||||
|
||||
|
||||
class BusyInterval(BaseModel):
|
||||
"""Minimal busy-slot payload for the public calendar. No title, name,
|
||||
email, or status is ever exposed here."""
|
||||
|
||||
start_time: datetime
|
||||
end_time: datetime
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Routes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@router.get("/properties", response_model=list[PropertyResponse])
|
||||
def list_public_properties(
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
@@ -33,43 +175,68 @@ def list_public_properties(
|
||||
.order_by(Property.name)
|
||||
.all()
|
||||
)
|
||||
result = []
|
||||
for p in properties:
|
||||
space_count = db.query(Space).filter(Space.property_id == p.id, Space.is_active == True).count() # noqa: E712
|
||||
result.append(PropertyResponse(
|
||||
id=p.id,
|
||||
name=p.name,
|
||||
description=p.description,
|
||||
address=p.address,
|
||||
is_public=p.is_public,
|
||||
is_active=p.is_active,
|
||||
created_at=p.created_at,
|
||||
space_count=space_count,
|
||||
))
|
||||
return result
|
||||
return [_property_to_response(db, p) for p in properties]
|
||||
|
||||
|
||||
@router.get("/properties/{property_id}/spaces", response_model=list[SpaceResponse])
|
||||
@router.get("/properties/{slug_or_id}", response_model=PropertyResponse)
|
||||
def get_public_property(
|
||||
slug_or_id: str,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> PropertyResponse:
|
||||
"""Resolve a public property by slug or numeric id (no auth required)."""
|
||||
prop = _resolve_property(db, slug_or_id)
|
||||
if not _is_visible(prop):
|
||||
raise HTTPException(status_code=404, detail="Property not found")
|
||||
return _property_to_response(db, prop)
|
||||
|
||||
|
||||
@router.get("/properties/{slug_or_id}/spaces", response_model=list[SpaceResponse])
|
||||
def list_public_property_spaces(
|
||||
property_id: int,
|
||||
slug_or_id: str,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> list[SpaceResponse]:
|
||||
"""List spaces of a public property (no auth required)."""
|
||||
prop = db.query(Property).filter(Property.id == property_id).first()
|
||||
if not prop:
|
||||
prop = _resolve_property(db, slug_or_id)
|
||||
if not _is_visible(prop):
|
||||
raise HTTPException(status_code=404, detail="Property not found")
|
||||
if not prop.is_public:
|
||||
raise HTTPException(status_code=403, detail="Property is private")
|
||||
|
||||
spaces = (
|
||||
db.query(Space)
|
||||
.filter(Space.property_id == property_id, Space.is_active == True) # noqa: E712
|
||||
.filter(Space.property_id == prop.id, Space.is_active == True) # noqa: E712
|
||||
.order_by(Space.name)
|
||||
.all()
|
||||
)
|
||||
return [SpaceResponse.model_validate(s) for s in spaces]
|
||||
|
||||
|
||||
@router.get("/spaces/{space_id}/busy", response_model=list[BusyInterval])
|
||||
def get_public_space_busy(
|
||||
space_id: int,
|
||||
start: Annotated[datetime, Query()],
|
||||
end: Annotated[datetime, Query()],
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> list[BusyInterval]:
|
||||
"""Return only the busy intervals for a space (no PII, no auth)."""
|
||||
if start.tzinfo is not None or end.tzinfo is not None:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail="start/end must be naive datetimes (property local time), not timezone-aware",
|
||||
)
|
||||
if end <= start:
|
||||
raise HTTPException(status_code=422, detail="end must be after start")
|
||||
if (end - start) > timedelta(days=60):
|
||||
raise HTTPException(status_code=422, detail="Range cannot exceed 60 days")
|
||||
|
||||
space = db.query(Space).filter(Space.id == space_id).first()
|
||||
if not space:
|
||||
raise HTTPException(status_code=404, detail="Space not found")
|
||||
if not _resolve_visible_space_property(db, space):
|
||||
raise HTTPException(status_code=404, detail="Space not found")
|
||||
|
||||
bookings = overlapping_bookings_query(db, space_id, start, end, statuses=("approved", "pending")).all()
|
||||
return [BusyInterval(start_time=b.start_datetime, end_time=b.end_datetime) for b in bookings]
|
||||
|
||||
|
||||
@router.get("/spaces/{space_id}/availability", response_model=AvailabilityCheck)
|
||||
def check_public_availability(
|
||||
space_id: int,
|
||||
@@ -81,36 +248,13 @@ def check_public_availability(
|
||||
space = db.query(Space).filter(Space.id == space_id).first()
|
||||
if not space:
|
||||
raise HTTPException(status_code=404, detail="Space not found")
|
||||
|
||||
# Verify space belongs to a public property
|
||||
if space.property_id:
|
||||
prop = db.query(Property).filter(Property.id == space.property_id).first()
|
||||
if prop and not prop.is_public:
|
||||
raise HTTPException(status_code=403, detail="Property is private")
|
||||
if not _resolve_visible_space_property(db, space):
|
||||
raise HTTPException(status_code=404, detail="Space not found")
|
||||
|
||||
# Find conflicting bookings
|
||||
conflicts = (
|
||||
db.query(Booking)
|
||||
.filter(
|
||||
Booking.space_id == space_id,
|
||||
Booking.status.in_(["approved", "pending"]),
|
||||
or_(
|
||||
and_(
|
||||
Booking.start_datetime <= start_datetime,
|
||||
Booking.end_datetime > start_datetime,
|
||||
),
|
||||
and_(
|
||||
Booking.start_datetime < end_datetime,
|
||||
Booking.end_datetime >= end_datetime,
|
||||
),
|
||||
and_(
|
||||
Booking.start_datetime >= start_datetime,
|
||||
Booking.end_datetime <= end_datetime,
|
||||
),
|
||||
),
|
||||
)
|
||||
.all()
|
||||
)
|
||||
conflicts = overlapping_bookings_query(
|
||||
db, space_id, start_datetime, end_datetime, statuses=("approved", "pending")
|
||||
).all()
|
||||
|
||||
if not conflicts:
|
||||
return AvailabilityCheck(available=True, conflicts=[], message="Time slot is available")
|
||||
@@ -126,10 +270,13 @@ def check_public_availability(
|
||||
return AvailabilityCheck(
|
||||
available=approved_count == 0,
|
||||
conflicts=[
|
||||
# user_name/title are scrubbed here: this endpoint is reachable
|
||||
# by anonymous callers and must not leak PII about other
|
||||
# bookers (privacy regression fix).
|
||||
ConflictingBooking(
|
||||
id=b.id,
|
||||
user_name=b.user.full_name if b.user else (b.guest_name or "Anonymous"),
|
||||
title=b.title,
|
||||
user_name="",
|
||||
title="",
|
||||
status=b.status,
|
||||
start_datetime=b.start_datetime,
|
||||
end_datetime=b.end_datetime,
|
||||
@@ -144,9 +291,12 @@ def check_public_availability(
|
||||
def create_anonymous_booking(
|
||||
data: AnonymousBookingCreate,
|
||||
background_tasks: BackgroundTasks,
|
||||
request: Request,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
) -> BookingResponse:
|
||||
"""Create an anonymous/guest booking (no auth required)."""
|
||||
_enforce_booking_rate_limit(request)
|
||||
|
||||
# Validate space exists
|
||||
space = db.query(Space).filter(Space.id == data.space_id).first()
|
||||
if not space:
|
||||
@@ -164,18 +314,36 @@ def create_anonymous_booking(
|
||||
if data.end_datetime <= data.start_datetime:
|
||||
raise HTTPException(status_code=400, detail="End time must be after start time")
|
||||
|
||||
# Check for overlapping approved bookings
|
||||
overlapping = db.query(Booking).filter(
|
||||
Booking.space_id == data.space_id,
|
||||
Booking.status == "approved",
|
||||
and_(
|
||||
Booking.start_datetime < data.end_datetime,
|
||||
Booking.end_datetime > data.start_datetime,
|
||||
),
|
||||
# Reject overlap with any pending OR approved booking (previously
|
||||
# approved-only, which let guests double-book against pending requests).
|
||||
overlapping = overlapping_bookings_query(
|
||||
db, data.space_id, data.start_datetime, data.end_datetime, statuses=("approved", "pending")
|
||||
).first()
|
||||
if overlapping:
|
||||
raise HTTPException(status_code=400, detail="Time slot is already booked")
|
||||
|
||||
# Cap pending requests per guest/space/day to deter spam.
|
||||
day_start = datetime.combine(data.start_datetime.date(), time.min)
|
||||
day_end = datetime.combine(data.start_datetime.date(), time.max)
|
||||
pending_today = (
|
||||
db.query(Booking)
|
||||
.filter(
|
||||
Booking.space_id == data.space_id,
|
||||
Booking.guest_email == data.guest_email,
|
||||
Booking.status == "pending",
|
||||
Booking.start_datetime >= day_start,
|
||||
Booking.start_datetime <= day_end,
|
||||
)
|
||||
.count()
|
||||
)
|
||||
if pending_today >= 5:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Ai atins limita de 5 cereri în așteptare pentru acest spațiu în această zi.",
|
||||
)
|
||||
|
||||
is_demo = bool(prop and prop.slug == DEMO_SLUG)
|
||||
|
||||
# Create anonymous booking
|
||||
booking = Booking(
|
||||
user_id=None,
|
||||
@@ -195,8 +363,9 @@ def create_anonymous_booking(
|
||||
db.commit()
|
||||
db.refresh(booking)
|
||||
|
||||
# Notify property managers
|
||||
if space.property_id:
|
||||
# Notify property managers/superadmins -- suppressed entirely for the
|
||||
# demo property so the public playground doesn't spam real staff.
|
||||
if space.property_id and not is_demo:
|
||||
manager_ids = [
|
||||
pm.user_id
|
||||
for pm in db.query(PropertyManager).filter(PropertyManager.property_id == space.property_id).all()
|
||||
@@ -224,4 +393,16 @@ def create_anonymous_booking(
|
||||
{"guest_email": data.guest_email},
|
||||
)
|
||||
|
||||
# Confirmation email to the guest themselves -- also suppressed for the
|
||||
# demo property (its data is wiped every few hours; nothing to confirm).
|
||||
if not is_demo:
|
||||
background_tasks.add_task(
|
||||
send_booking_notification,
|
||||
booking,
|
||||
"anonymous_created",
|
||||
data.guest_email,
|
||||
data.guest_name,
|
||||
{"guest_email": data.guest_email},
|
||||
)
|
||||
|
||||
return BookingResponse.model_validate(booking)
|
||||
|
||||
98
backend/app/core/slug.py
Normal file
98
backend/app/core/slug.py
Normal file
@@ -0,0 +1,98 @@
|
||||
"""Slug generation and validation for public property booking links.
|
||||
|
||||
Single source of truth for slug rules shared across the API, migrations,
|
||||
and the demo seed script.
|
||||
"""
|
||||
import re
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# Paths/segments that must never be shadowed by a property slug.
|
||||
RESERVED_SLUGS = {
|
||||
"api", "admin", "book", "login", "register", "dashboard", "properties",
|
||||
"public", "spaces", "organization", "users", "verify", "settings",
|
||||
"reports", "confidentialitate", "termeni",
|
||||
}
|
||||
|
||||
# Stable, deterministic slug reserved for the public demo property.
|
||||
DEMO_SLUG = "proprietate-demo"
|
||||
|
||||
# Format accepted for user-supplied (PATCH) slugs: lowercase alnum + hyphen,
|
||||
# 3-64 chars total, no leading/trailing/double hyphen (single run of `-`
|
||||
# separating alnum groups), enforced by the regex plus extra checks below.
|
||||
SLUG_REGEX = re.compile(r"^[a-z0-9](?:[a-z0-9-]{1,62}[a-z0-9])$")
|
||||
|
||||
# RO diacritics -> ASCII transliteration map.
|
||||
_RO_CHAR_MAP = {
|
||||
"ă": "a", "â": "a", "î": "i", "ș": "s", "ş": "s", "ț": "t", "ţ": "t",
|
||||
"Ă": "a", "Â": "a", "Î": "i", "Ș": "s", "Ş": "s", "Ț": "t", "Ţ": "t",
|
||||
}
|
||||
|
||||
_NON_ALNUM_RUN = re.compile(r"[^a-z0-9]+")
|
||||
|
||||
|
||||
def slugify(name: str) -> str:
|
||||
"""Transliterate RO diacritics, lowercase, and kebab-case a name.
|
||||
|
||||
Non `[a-z0-9]` runs collapse to a single `-`; leading/trailing/double
|
||||
hyphens are stripped. Pure stdlib, no external dependency.
|
||||
"""
|
||||
if not name:
|
||||
return ""
|
||||
out = []
|
||||
for ch in name:
|
||||
out.append(_RO_CHAR_MAP.get(ch, ch))
|
||||
text = "".join(out).lower()
|
||||
text = _NON_ALNUM_RUN.sub("-", text)
|
||||
return text.strip("-")
|
||||
|
||||
|
||||
def _is_all_digit(slug: str) -> bool:
|
||||
return slug.isdigit()
|
||||
|
||||
|
||||
def validate_slug(slug: str) -> bool:
|
||||
"""Validate a user-supplied slug against the PATCH format rules.
|
||||
|
||||
Rules: `[a-z0-9-]{3,64}`, no leading/trailing/double hyphen, not
|
||||
all-digit (would shadow numeric property IDs), not reserved.
|
||||
"""
|
||||
if not slug:
|
||||
return False
|
||||
if len(slug) < 3 or len(slug) > 64:
|
||||
return False
|
||||
if not SLUG_REGEX.match(slug):
|
||||
return False
|
||||
if "--" in slug:
|
||||
return False
|
||||
if _is_all_digit(slug):
|
||||
return False
|
||||
if slug in RESERVED_SLUGS:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def generate_unique_slug(db: Session, name: str, exclude_id: int | None = None) -> str:
|
||||
"""Generate a unique, safe slug for a property from its name.
|
||||
|
||||
Idempotent: given the same name and existing DB state, always returns
|
||||
the same result. Empty/all-digit/reserved base slugs get a `p-` prefix.
|
||||
Collisions are resolved by appending `-2`, `-3`, etc.
|
||||
"""
|
||||
from app.models.property import Property
|
||||
|
||||
base = slugify(name)
|
||||
if not base or _is_all_digit(base) or base in RESERVED_SLUGS:
|
||||
base = f"p-{base}" if base else "p"
|
||||
|
||||
candidate = base
|
||||
suffix = 2
|
||||
while True:
|
||||
if candidate not in RESERVED_SLUGS:
|
||||
query = db.query(Property).filter(Property.slug == candidate)
|
||||
if exclude_id is not None:
|
||||
query = query.filter(Property.id != exclude_id)
|
||||
if query.first() is None:
|
||||
return candidate
|
||||
candidate = f"{base}-{suffix}"
|
||||
suffix += 1
|
||||
@@ -1,16 +1,28 @@
|
||||
"""Database session management."""
|
||||
from collections.abc import Generator
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy import create_engine, event
|
||||
from sqlalchemy.orm import Session, declarative_base, sessionmaker
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
_is_sqlite = "sqlite" in settings.database_url
|
||||
|
||||
engine = create_engine(
|
||||
settings.database_url,
|
||||
connect_args={"check_same_thread": False} if "sqlite" in settings.database_url else {},
|
||||
connect_args={"check_same_thread": False} if _is_sqlite else {},
|
||||
)
|
||||
|
||||
if _is_sqlite:
|
||||
@event.listens_for(engine, "connect")
|
||||
def _set_sqlite_pragma(dbapi_connection, connection_record) -> None:
|
||||
"""Enable WAL mode + a busy timeout so concurrent readers/writers
|
||||
don't immediately hit 'database is locked' errors."""
|
||||
cursor = dbapi_connection.cursor()
|
||||
cursor.execute("PRAGMA journal_mode=WAL")
|
||||
cursor.execute("PRAGMA busy_timeout=5000")
|
||||
cursor.close()
|
||||
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
@@ -18,3 +18,5 @@ class Property(Base):
|
||||
is_public = Column(Boolean, default=True, nullable=False)
|
||||
is_active = Column(Boolean, default=True, nullable=False)
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
slug = Column(String, unique=True, nullable=True, index=True)
|
||||
list_on_landing = Column(Boolean, default=False, nullable=False)
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
"""Booking schemas for request/response."""
|
||||
from datetime import datetime, date
|
||||
from datetime import date, datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator, model_validator
|
||||
from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator
|
||||
|
||||
|
||||
class BookingCalendarPublic(BaseModel):
|
||||
@@ -276,5 +276,5 @@ class AnonymousBookingCreate(BaseModel):
|
||||
title: str = Field(..., min_length=1, max_length=200)
|
||||
description: str | None = None
|
||||
guest_name: str = Field(..., min_length=1)
|
||||
guest_email: str = Field(..., min_length=1)
|
||||
guest_email: EmailStr
|
||||
guest_organization: str | None = None
|
||||
|
||||
@@ -9,6 +9,7 @@ class PropertyCreate(BaseModel):
|
||||
description: str | None = None
|
||||
address: str | None = None
|
||||
is_public: bool = True
|
||||
list_on_landing: bool = False
|
||||
|
||||
|
||||
class PropertyUpdate(BaseModel):
|
||||
@@ -16,6 +17,8 @@ class PropertyUpdate(BaseModel):
|
||||
description: str | None = None
|
||||
address: str | None = None
|
||||
is_public: bool | None = None
|
||||
slug: str | None = None
|
||||
list_on_landing: bool | None = None
|
||||
|
||||
|
||||
class PropertyManagerInfo(BaseModel):
|
||||
@@ -34,6 +37,9 @@ class PropertyResponse(BaseModel):
|
||||
created_at: datetime
|
||||
space_count: int = 0
|
||||
managers: list[PropertyManagerInfo] = []
|
||||
slug: str | None = None
|
||||
list_on_landing: bool = False
|
||||
require_approval: bool | None = None
|
||||
|
||||
model_config = {"from_attributes": True}
|
||||
|
||||
|
||||
@@ -153,11 +153,13 @@ Sistemul de Rezervări
|
||||
"""
|
||||
|
||||
elif event_type == "anonymous_approved":
|
||||
reference_code = extra_data.get("reference_code", f"RB-{booking.id}") if extra_data else f"RB-{booking.id}"
|
||||
subject = "Rezervare Aprobată"
|
||||
body = f"""Bună ziua {user_name},
|
||||
|
||||
Rezervarea dumneavoastră a fost aprobată:
|
||||
|
||||
Cod de referință: #{reference_code}
|
||||
Spațiu: {space_name}
|
||||
Data și ora: {start_str} - {end_str}
|
||||
Titlu: {booking.title}
|
||||
@@ -170,11 +172,13 @@ Sistemul de Rezervări
|
||||
|
||||
elif event_type == "anonymous_rejected":
|
||||
reason = extra_data.get("rejection_reason", "Nu a fost specificat") if extra_data else "Nu a fost specificat"
|
||||
reference_code = extra_data.get("reference_code", f"RB-{booking.id}") if extra_data else f"RB-{booking.id}"
|
||||
subject = "Rezervare Respinsă"
|
||||
body = f"""Bună ziua {user_name},
|
||||
|
||||
Rezervarea dumneavoastră a fost respinsă:
|
||||
|
||||
Cod de referință: #{reference_code}
|
||||
Spațiu: {space_name}
|
||||
Data și ora: {start_str} - {end_str}
|
||||
Titlu: {booking.title}
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
#!/bin/bash
|
||||
set -e
|
||||
|
||||
echo "[entrypoint] Running property slug migration..."
|
||||
python migrate_add_property_slug.py
|
||||
|
||||
# Database tables are created automatically on application startup
|
||||
# (app/main.py runs Base.metadata.create_all). The first user to register
|
||||
# becomes the superadmin (the instance owner), so no admin seeding is needed.
|
||||
|
||||
89
backend/migrate_add_property_slug.py
Normal file
89
backend/migrate_add_property_slug.py
Normal file
@@ -0,0 +1,89 @@
|
||||
"""Migration script to add public-booking-link slug support to properties.
|
||||
|
||||
Idempotent (safe to run multiple times):
|
||||
- Adds `slug` (VARCHAR, nullable) and `list_on_landing` (BOOLEAN NOT NULL
|
||||
DEFAULT 0) columns to `properties` if missing.
|
||||
- Creates a UNIQUE INDEX on `slug` if missing.
|
||||
- Backfills NULL slugs deterministically from each property's name.
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add backend to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from sqlalchemy import inspect, text
|
||||
|
||||
from app.core.slug import generate_unique_slug
|
||||
from app.db.session import SessionLocal, engine
|
||||
from app.models.property import Property
|
||||
|
||||
SLUG_INDEX_NAME = "ix_properties_slug_unique"
|
||||
|
||||
|
||||
def migrate() -> None:
|
||||
"""Run migration to add slug + list_on_landing support to properties."""
|
||||
inspector = inspect(engine)
|
||||
columns = [col["name"] for col in inspector.get_columns("properties")]
|
||||
|
||||
print("Starting property slug migration...")
|
||||
|
||||
with engine.connect() as conn:
|
||||
if "slug" not in columns:
|
||||
print("1. Adding slug column to properties...")
|
||||
conn.execute(text("ALTER TABLE properties ADD COLUMN slug VARCHAR"))
|
||||
conn.commit()
|
||||
print(" Column added.")
|
||||
else:
|
||||
print("1. slug column already exists.")
|
||||
|
||||
if "list_on_landing" not in columns:
|
||||
print("2. Adding list_on_landing column to properties...")
|
||||
conn.execute(
|
||||
text("ALTER TABLE properties ADD COLUMN list_on_landing BOOLEAN DEFAULT 0 NOT NULL")
|
||||
)
|
||||
conn.commit()
|
||||
print(" Column added.")
|
||||
else:
|
||||
print("2. list_on_landing column already exists.")
|
||||
|
||||
# Recompute indexes after potentially adding the slug column.
|
||||
inspector = inspect(engine)
|
||||
index_names = {ix["name"] for ix in inspector.get_indexes("properties")}
|
||||
if SLUG_INDEX_NAME not in index_names:
|
||||
print("3. Creating unique index on slug...")
|
||||
with engine.connect() as conn:
|
||||
conn.execute(
|
||||
text(f"CREATE UNIQUE INDEX {SLUG_INDEX_NAME} ON properties (slug)")
|
||||
)
|
||||
conn.commit()
|
||||
print(" Index created.")
|
||||
else:
|
||||
print("3. Unique index on slug already exists.")
|
||||
|
||||
# Backfill: deterministic unique slug for every property missing one.
|
||||
print("4. Backfilling missing slugs...")
|
||||
db = SessionLocal()
|
||||
try:
|
||||
properties_without_slug = (
|
||||
db.query(Property)
|
||||
.filter(Property.slug.is_(None))
|
||||
.order_by(Property.id)
|
||||
.all()
|
||||
)
|
||||
for prop in properties_without_slug:
|
||||
prop.slug = generate_unique_slug(db, prop.name, exclude_id=prop.id)
|
||||
db.flush()
|
||||
db.commit()
|
||||
print(f" Backfilled {len(properties_without_slug)} propert(y/ies).")
|
||||
except Exception:
|
||||
db.rollback()
|
||||
raise
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
print("Migration completed successfully!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
migrate()
|
||||
@@ -15,6 +15,7 @@ from datetime import datetime, timedelta
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.security import get_password_hash
|
||||
from app.core.slug import DEMO_SLUG
|
||||
from app.db.session import Base, SessionLocal, engine
|
||||
from app.models.attachment import Attachment
|
||||
from app.models.booking import Booking
|
||||
@@ -126,8 +127,10 @@ def _seed_demo_data(db, demo: User) -> None:
|
||||
name=DEMO_PROPERTY_NAME,
|
||||
description="Mediu de test — datele se resetează automat la fiecare 3 ore.",
|
||||
address="Str. Exemplu nr. 1",
|
||||
is_public=False, # visible only to the demo account
|
||||
is_public=True,
|
||||
is_active=True,
|
||||
list_on_landing=True,
|
||||
slug=DEMO_SLUG,
|
||||
)
|
||||
db.add(prop)
|
||||
db.flush()
|
||||
|
||||
119
backend/tests/test_migrate_slug.py
Normal file
119
backend/tests/test_migrate_slug.py
Normal file
@@ -0,0 +1,119 @@
|
||||
"""Tests for the idempotent property-slug migration script.
|
||||
|
||||
Builds a temporary SQLite DB with the OLD `properties` schema (no slug /
|
||||
list_on_landing columns) via raw DDL, points the migration module at it,
|
||||
and verifies the migration adds the columns/index and deterministically
|
||||
backfills RO-transliterated, collision-safe slugs -- twice, with no error
|
||||
and no drift the second time.
|
||||
"""
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, inspect, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
import migrate_add_property_slug as migrate_mod
|
||||
|
||||
|
||||
def _create_old_schema_db(path: str) -> None:
|
||||
"""Create a properties table matching the pre-slug schema and seed rows."""
|
||||
conn = sqlite3.connect(path)
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE properties (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name VARCHAR NOT NULL,
|
||||
description VARCHAR,
|
||||
address VARCHAR,
|
||||
is_public BOOLEAN NOT NULL DEFAULT 1,
|
||||
is_active BOOLEAN NOT NULL DEFAULT 1,
|
||||
created_at DATETIME NOT NULL
|
||||
)
|
||||
"""
|
||||
)
|
||||
rows = [
|
||||
("Clădirea Centrală", "2024-01-01T00:00:00"),
|
||||
# Same (accented) name again -> must collide and get a -2 suffix.
|
||||
("Clădirea Centrală", "2024-01-02T00:00:00"),
|
||||
("Sediul Ș.ț Test", "2024-01-03T00:00:00"),
|
||||
]
|
||||
for name, created_at in rows:
|
||||
conn.execute(
|
||||
"INSERT INTO properties (name, description, address, is_public, is_active, created_at) "
|
||||
"VALUES (?, NULL, NULL, 1, 1, ?)",
|
||||
(name, created_at),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db_path(tmp_path) -> str:
|
||||
path = tmp_path / "migrate_test.db"
|
||||
_create_old_schema_db(str(path))
|
||||
return str(path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patched_engine(monkeypatch: pytest.MonkeyPatch, temp_db_path: str):
|
||||
"""Point the migration module's module-level engine/SessionLocal at the temp DB."""
|
||||
engine = create_engine(f"sqlite:///{temp_db_path}", connect_args={"check_same_thread": False})
|
||||
session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
monkeypatch.setattr(migrate_mod, "engine", engine)
|
||||
monkeypatch.setattr(migrate_mod, "SessionLocal", session_local)
|
||||
yield engine
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _fetch_rows(engine) -> list[tuple]:
|
||||
with engine.connect() as conn:
|
||||
return conn.execute(text("SELECT id, name, slug FROM properties ORDER BY id")).fetchall()
|
||||
|
||||
|
||||
def test_migration_adds_columns_and_unique_index(patched_engine) -> None:
|
||||
migrate_mod.migrate()
|
||||
|
||||
inspector = inspect(patched_engine)
|
||||
columns = {c["name"] for c in inspector.get_columns("properties")}
|
||||
assert "slug" in columns
|
||||
assert "list_on_landing" in columns
|
||||
|
||||
indexes = inspector.get_indexes("properties")
|
||||
slug_indexes = [ix for ix in indexes if "slug" in ix["column_names"]]
|
||||
assert slug_indexes, "expected an index on slug"
|
||||
assert any(ix["unique"] for ix in slug_indexes), "slug index must be unique"
|
||||
|
||||
|
||||
def test_migration_backfills_deterministic_ro_slugs_with_collision_suffix(patched_engine) -> None:
|
||||
migrate_mod.migrate()
|
||||
|
||||
rows = _fetch_rows(patched_engine)
|
||||
assert len(rows) == 3
|
||||
slugs = [r[2] for r in rows]
|
||||
|
||||
# RO diacritic transliteration: ă->a, â->a, î->i, ș/ş->s, ț/ţ->t.
|
||||
assert slugs[0] == "cladirea-centrala"
|
||||
# Second property has the identical (transliterated) name -> collision -> -2 suffix.
|
||||
assert slugs[1] == "cladirea-centrala-2"
|
||||
assert slugs[2] == "sediul-s-t-test"
|
||||
|
||||
# All slugs non-null and unique.
|
||||
assert all(s for s in slugs)
|
||||
assert len(set(slugs)) == len(slugs)
|
||||
|
||||
|
||||
def test_migration_is_idempotent_and_slugs_stable_on_rerun(patched_engine) -> None:
|
||||
migrate_mod.migrate()
|
||||
first = _fetch_rows(patched_engine)
|
||||
|
||||
# Running a second time must not raise and must not change existing slugs
|
||||
# or column/index state.
|
||||
migrate_mod.migrate()
|
||||
second = _fetch_rows(patched_engine)
|
||||
|
||||
assert first == second
|
||||
|
||||
inspector = inspect(patched_engine)
|
||||
columns = {c["name"] for c in inspector.get_columns("properties")}
|
||||
assert "slug" in columns
|
||||
assert "list_on_landing" in columns
|
||||
492
backend/tests/test_public.py
Normal file
492
backend/tests/test_public.py
Normal file
@@ -0,0 +1,492 @@
|
||||
"""Tests for public/anonymous booking endpoints and the property-slug PATCH path."""
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
# Rate limiting is env-gated (PYTEST_CURRENT_TEST / DISABLE_RATE_LIMIT); set the
|
||||
# explicit override too in case this module is ever imported outside pytest's
|
||||
# normal collection (PYTEST_CURRENT_TEST is only set during actual test runs).
|
||||
os.environ["DISABLE_RATE_LIMIT"] = "1"
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.slug import DEMO_SLUG
|
||||
from app.models.booking import Booking
|
||||
from app.models.notification import Notification
|
||||
from app.models.property import Property
|
||||
from app.models.property_manager import PropertyManager
|
||||
from app.models.property_settings import PropertySettings
|
||||
from app.models.space import Space
|
||||
from app.models.user import User
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def public_property(db: Session) -> Property:
|
||||
prop = Property(
|
||||
name="Public Test Property",
|
||||
is_public=True,
|
||||
is_active=True,
|
||||
slug="public-test-property",
|
||||
list_on_landing=False,
|
||||
)
|
||||
db.add(prop)
|
||||
db.commit()
|
||||
db.refresh(prop)
|
||||
db.add(PropertySettings(property_id=prop.id, require_approval=True))
|
||||
db.commit()
|
||||
return prop
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def public_space(db: Session, public_property: Property) -> Space:
|
||||
space = Space(
|
||||
name="Public Room",
|
||||
type="sala",
|
||||
capacity=8,
|
||||
is_active=True,
|
||||
property_id=public_property.id,
|
||||
)
|
||||
db.add(space)
|
||||
db.commit()
|
||||
db.refresh(space)
|
||||
return space
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def private_property(db: Session) -> Property:
|
||||
prop = Property(
|
||||
name="Private Test Property",
|
||||
is_public=False,
|
||||
is_active=True,
|
||||
slug="private-test-property",
|
||||
list_on_landing=False,
|
||||
)
|
||||
db.add(prop)
|
||||
db.commit()
|
||||
db.refresh(prop)
|
||||
return prop
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def private_space(db: Session, private_property: Property) -> Space:
|
||||
space = Space(
|
||||
name="Private Room",
|
||||
type="sala",
|
||||
capacity=4,
|
||||
is_active=True,
|
||||
property_id=private_property.id,
|
||||
)
|
||||
db.add(space)
|
||||
db.commit()
|
||||
db.refresh(space)
|
||||
return space
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def orphan_space(db: Session) -> Space:
|
||||
"""A space with no property (property_id NULL)."""
|
||||
space = Space(name="Orphan Room", type="sala", capacity=2, is_active=True, property_id=None)
|
||||
db.add(space)
|
||||
db.commit()
|
||||
db.refresh(space)
|
||||
return space
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def demo_property(db: Session) -> Property:
|
||||
prop = Property(
|
||||
name="Proprietate Demo",
|
||||
is_public=True,
|
||||
is_active=True,
|
||||
slug=DEMO_SLUG,
|
||||
list_on_landing=True,
|
||||
)
|
||||
db.add(prop)
|
||||
db.commit()
|
||||
db.refresh(prop)
|
||||
db.add(PropertySettings(property_id=prop.id, require_approval=True))
|
||||
db.commit()
|
||||
return prop
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def demo_space(db: Session, demo_property: Property) -> Space:
|
||||
space = Space(
|
||||
name="Demo Room",
|
||||
type="sala",
|
||||
capacity=6,
|
||||
is_active=True,
|
||||
property_id=demo_property.id,
|
||||
)
|
||||
db.add(space)
|
||||
db.commit()
|
||||
db.refresh(space)
|
||||
return space
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def property_manager_user(db: Session, public_property: Property) -> User:
|
||||
from app.core.security import get_password_hash
|
||||
|
||||
user = User(
|
||||
email="propmanager@example.com",
|
||||
full_name="Property Manager",
|
||||
hashed_password=get_password_hash("managerpass"),
|
||||
role="manager",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
db.add(PropertyManager(property_id=public_property.id, user_id=user.id))
|
||||
db.commit()
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def manager_headers(property_manager_user: User) -> dict[str, str]:
|
||||
from app.core.security import create_access_token
|
||||
|
||||
token = create_access_token(subject=int(property_manager_user.id))
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
def _guest_payload(space_id: int, start: datetime, end: datetime, email: str = "guest@example.com") -> dict:
|
||||
return {
|
||||
"space_id": space_id,
|
||||
"start_datetime": start.isoformat(),
|
||||
"end_datetime": end.isoformat(),
|
||||
"title": "Guest Meeting",
|
||||
"description": None,
|
||||
"guest_name": "Guest Person",
|
||||
"guest_email": email,
|
||||
"guest_organization": None,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# busy endpoint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_busy_returns_only_start_end_times(
|
||||
client: TestClient, db: Session, public_space: Space
|
||||
) -> None:
|
||||
approved = Booking(
|
||||
space_id=public_space.id,
|
||||
title="Secret Title",
|
||||
description=None,
|
||||
start_datetime=datetime(2025, 6, 2, 10, 0, 0),
|
||||
end_datetime=datetime(2025, 6, 2, 11, 0, 0),
|
||||
status="approved",
|
||||
guest_name="Secret Guest",
|
||||
guest_email="secret@example.com",
|
||||
is_anonymous=True,
|
||||
)
|
||||
pending = Booking(
|
||||
space_id=public_space.id,
|
||||
title="Pending Title",
|
||||
start_datetime=datetime(2025, 6, 2, 13, 0, 0),
|
||||
end_datetime=datetime(2025, 6, 2, 14, 0, 0),
|
||||
status="pending",
|
||||
guest_name="Pending Guest",
|
||||
guest_email="pending@example.com",
|
||||
is_anonymous=True,
|
||||
)
|
||||
cancelled = Booking(
|
||||
space_id=public_space.id,
|
||||
title="Cancelled Title",
|
||||
start_datetime=datetime(2025, 6, 2, 16, 0, 0),
|
||||
end_datetime=datetime(2025, 6, 2, 17, 0, 0),
|
||||
status="cancelled",
|
||||
guest_name="Cancelled Guest",
|
||||
guest_email="cancelled@example.com",
|
||||
is_anonymous=True,
|
||||
)
|
||||
db.add_all([approved, pending, cancelled])
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{public_space.id}/busy",
|
||||
params={"start": "2025-06-01T00:00:00", "end": "2025-06-05T00:00:00"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 2 # approved + pending, cancelled excluded
|
||||
|
||||
for item in data:
|
||||
assert set(item.keys()) == {"start_time", "end_time"}
|
||||
for forbidden in ("title", "user_name", "guest_name", "guest_email", "email", "status", "id"):
|
||||
assert forbidden not in item
|
||||
|
||||
|
||||
def test_busy_orphan_space_returns_404(client: TestClient, orphan_space: Space) -> None:
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{orphan_space.id}/busy",
|
||||
params={"start": "2025-06-01T00:00:00", "end": "2025-06-05T00:00:00"},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_busy_private_property_returns_404(client: TestClient, private_space: Space) -> None:
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{private_space.id}/busy",
|
||||
params={"start": "2025-06-01T00:00:00", "end": "2025-06-05T00:00:00"},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_busy_missing_space_returns_404(client: TestClient) -> None:
|
||||
response = client.get(
|
||||
"/api/public/spaces/999999/busy",
|
||||
params={"start": "2025-06-01T00:00:00", "end": "2025-06-05T00:00:00"},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_busy_rejects_timezone_aware_params(client: TestClient, public_space: Space) -> None:
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{public_space.id}/busy",
|
||||
params={
|
||||
"start": "2025-06-01T00:00:00+02:00",
|
||||
"end": "2025-06-05T00:00:00+02:00",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_busy_rejects_end_before_or_equal_start(client: TestClient, public_space: Space) -> None:
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{public_space.id}/busy",
|
||||
params={"start": "2025-06-05T00:00:00", "end": "2025-06-01T00:00:00"},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
response_eq = client.get(
|
||||
f"/api/public/spaces/{public_space.id}/busy",
|
||||
params={"start": "2025-06-01T00:00:00", "end": "2025-06-01T00:00:00"},
|
||||
)
|
||||
assert response_eq.status_code == 422
|
||||
|
||||
|
||||
def test_busy_rejects_range_over_60_days(client: TestClient, public_space: Space) -> None:
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{public_space.id}/busy",
|
||||
params={"start": "2025-01-01T00:00:00", "end": "2025-06-01T00:00:00"},
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# slug resolve
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_property_by_slug(client: TestClient, public_property: Property) -> None:
|
||||
response = client.get(f"/api/public/properties/{public_property.slug}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["id"] == public_property.id
|
||||
assert response.json()["slug"] == public_property.slug
|
||||
|
||||
|
||||
def test_get_property_by_numeric_id(client: TestClient, public_property: Property) -> None:
|
||||
response = client.get(f"/api/public/properties/{public_property.id}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["slug"] == public_property.slug
|
||||
|
||||
|
||||
def test_get_property_wrong_slug_404(client: TestClient) -> None:
|
||||
response = client.get("/api/public/properties/does-not-exist")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_get_property_private_returns_404_not_403(client: TestClient, private_property: Property) -> None:
|
||||
response = client.get(f"/api/public/properties/{private_property.slug}")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_get_property_spaces_by_slug(client: TestClient, public_property: Property, public_space: Space) -> None:
|
||||
response = client.get(f"/api/public/properties/{public_property.slug}/spaces")
|
||||
assert response.status_code == 200
|
||||
ids = [s["id"] for s in response.json()]
|
||||
assert public_space.id in ids
|
||||
|
||||
|
||||
def test_get_property_spaces_private_returns_404(client: TestClient, private_property: Property) -> None:
|
||||
response = client.get(f"/api/public/properties/{private_property.slug}/spaces")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_get_property_spaces_missing_returns_404(client: TestClient) -> None:
|
||||
response = client.get("/api/public/properties/does-not-exist/spaces")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# availability regression: no PII leak
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_availability_scrubs_user_name_and_title(client: TestClient, db: Session, public_space: Space) -> None:
|
||||
booking = Booking(
|
||||
space_id=public_space.id,
|
||||
title="Confidential Planning Session",
|
||||
start_datetime=datetime(2025, 7, 1, 10, 0, 0),
|
||||
end_datetime=datetime(2025, 7, 1, 11, 0, 0),
|
||||
status="approved",
|
||||
guest_name="Confidential Guest",
|
||||
guest_email="confidential@example.com",
|
||||
is_anonymous=True,
|
||||
)
|
||||
db.add(booking)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
f"/api/public/spaces/{public_space.id}/availability",
|
||||
params={"start_datetime": "2025-07-01T10:30:00", "end_datetime": "2025-07-01T10:45:00"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["available"] is False
|
||||
assert len(data["conflicts"]) == 1
|
||||
conflict = data["conflicts"][0]
|
||||
# No PII leaked...
|
||||
assert conflict["user_name"] == ""
|
||||
assert conflict["title"] == ""
|
||||
# ...but useful scheduling info is preserved.
|
||||
assert conflict["status"] == "approved"
|
||||
assert conflict["start_datetime"]
|
||||
assert conflict["end_datetime"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PATCH slug via PUT /manager/properties/{id}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_patch_slug_duplicate_returns_409(
|
||||
client: TestClient, db: Session, public_property: Property, manager_headers: dict[str, str]
|
||||
) -> None:
|
||||
other = Property(name="Other Property", is_public=True, is_active=True, slug="other-property")
|
||||
db.add(other)
|
||||
db.commit()
|
||||
db.refresh(other)
|
||||
|
||||
response = client.put(
|
||||
f"/api/manager/properties/{public_property.id}",
|
||||
json={"slug": other.slug},
|
||||
headers=manager_headers,
|
||||
)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bad_slug",
|
||||
["ab", "has space", "123456", "admin", "-bad-start", "bad-end-", "a--b"],
|
||||
)
|
||||
def test_patch_slug_invalid_format_returns_422(
|
||||
client: TestClient, public_property: Property, manager_headers: dict[str, str], bad_slug: str
|
||||
) -> None:
|
||||
response = client.put(
|
||||
f"/api/manager/properties/{public_property.id}",
|
||||
json={"slug": bad_slug},
|
||||
headers=manager_headers,
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_patch_slug_valid_succeeds(
|
||||
client: TestClient, public_property: Property, manager_headers: dict[str, str]
|
||||
) -> None:
|
||||
response = client.put(
|
||||
f"/api/manager/properties/{public_property.id}",
|
||||
json={"slug": "brand-new-slug"},
|
||||
headers=manager_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["slug"] == "brand-new-slug"
|
||||
|
||||
|
||||
def test_patch_slug_non_manager_returns_403(
|
||||
client: TestClient, public_property: Property, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
response = client.put(
|
||||
f"/api/manager/properties/{public_property.id}",
|
||||
json={"slug": "some-new-slug"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /public/bookings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_post_anonymous_booking_happy_path(client: TestClient, public_space: Space) -> None:
|
||||
start = datetime(2025, 8, 1, 9, 0, 0)
|
||||
end = datetime(2025, 8, 1, 10, 0, 0)
|
||||
response = client.post("/api/public/bookings", json=_guest_payload(public_space.id, start, end))
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["is_anonymous"] is True
|
||||
assert data["guest_email"] == "guest@example.com"
|
||||
assert data["status"] == "pending"
|
||||
|
||||
|
||||
def test_post_anonymous_booking_pending_overlap_rejected(client: TestClient, public_space: Space) -> None:
|
||||
start = datetime(2025, 8, 2, 9, 0, 0)
|
||||
end = datetime(2025, 8, 2, 10, 0, 0)
|
||||
first = client.post("/api/public/bookings", json=_guest_payload(public_space.id, start, end, "first@example.com"))
|
||||
assert first.status_code == 201
|
||||
|
||||
overlap_start = datetime(2025, 8, 2, 9, 30, 0)
|
||||
overlap_end = datetime(2025, 8, 2, 10, 30, 0)
|
||||
second = client.post(
|
||||
"/api/public/bookings",
|
||||
json=_guest_payload(public_space.id, overlap_start, overlap_end, "second@example.com"),
|
||||
)
|
||||
assert second.status_code == 400
|
||||
|
||||
|
||||
def test_post_anonymous_booking_cap_five_pending_per_email_per_day(client: TestClient, public_space: Space) -> None:
|
||||
email = "capped@example.com"
|
||||
for hour in range(9, 14): # 5 non-overlapping slots: 9-10,10-11,...,13-14
|
||||
start = datetime(2025, 8, 3, hour, 0, 0)
|
||||
end = datetime(2025, 8, 3, hour + 1, 0, 0)
|
||||
response = client.post("/api/public/bookings", json=_guest_payload(public_space.id, start, end, email))
|
||||
assert response.status_code == 201
|
||||
|
||||
# 6th request for same guest/space/day must be rejected regardless of overlap.
|
||||
sixth_start = datetime(2025, 8, 3, 15, 0, 0)
|
||||
sixth_end = datetime(2025, 8, 3, 16, 0, 0)
|
||||
sixth = client.post("/api/public/bookings", json=_guest_payload(public_space.id, sixth_start, sixth_end, email))
|
||||
assert sixth.status_code == 400
|
||||
|
||||
|
||||
def test_post_anonymous_booking_demo_property_no_notifications(
|
||||
client: TestClient, db: Session, demo_space: Space, test_admin: User
|
||||
) -> None:
|
||||
start = datetime(2025, 8, 4, 9, 0, 0)
|
||||
end = datetime(2025, 8, 4, 10, 0, 0)
|
||||
response = client.post("/api/public/bookings", json=_guest_payload(demo_space.id, start, end, "demoguest@example.com"))
|
||||
assert response.status_code == 201
|
||||
booking_id = response.json()["id"]
|
||||
|
||||
notifications = db.query(Notification).filter(Notification.booking_id == booking_id).all()
|
||||
assert notifications == []
|
||||
|
||||
|
||||
def test_post_anonymous_booking_invalid_email_returns_422(client: TestClient, public_space: Space) -> None:
|
||||
start = datetime(2025, 8, 5, 9, 0, 0)
|
||||
end = datetime(2025, 8, 5, 10, 0, 0)
|
||||
payload = _guest_payload(public_space.id, start, end, "not-an-email")
|
||||
response = client.post("/api/public/bookings", json=payload)
|
||||
assert response.status_code == 422
|
||||
Reference in New Issue
Block a user