"""Tests for booking calendar endpoints.""" from datetime import datetime import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session from app.models.booking import Booking from app.models.space import Space from app.models.user import User def test_get_bookings_as_user_shows_only_public_data( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that regular users see only public booking data (no user_id, description).""" # Create a booking with confidential info booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Confidential discussion about project X", start_datetime=datetime(2024, 3, 15, 10, 0, 0), end_datetime=datetime(2024, 3, 15, 12, 0, 0), status="approved", rejection_reason=None, cancellation_reason=None, ) db.add(booking) db.commit() # Request bookings as regular user response = client.get( f"/api/spaces/{test_space.id}/bookings", params={ "start": "2024-03-01T00:00:00", "end": "2024-03-31T23:59:59", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 1 booking_data = bookings[0] # Public fields should be present assert booking_data["id"] == booking.id assert booking_data["title"] == "Team Meeting" assert booking_data["status"] == "approved" assert "start_datetime" in booking_data assert "end_datetime" in booking_data # Private fields should NOT be present assert "user_id" not in booking_data assert "description" not in booking_data assert "rejection_reason" not in booking_data assert "cancellation_reason" not in booking_data assert "approved_by" not in booking_data assert "created_at" not in booking_data def test_get_bookings_as_admin_shows_all_details( client: TestClient, admin_token: str, test_space: Space, test_user: User, test_admin: User, db: Session, ) -> None: """Test that admins see all booking details including user info.""" # Create a booking with all details booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Executive Meeting", description="Confidential strategic planning", start_datetime=datetime(2024, 3, 20, 14, 0, 0), end_datetime=datetime(2024, 3, 20, 16, 0, 0), status="approved", rejection_reason=None, cancellation_reason=None, approved_by=test_admin.id, ) db.add(booking) db.commit() db.refresh(booking) # Request bookings as admin response = client.get( f"/api/spaces/{test_space.id}/bookings", params={ "start": "2024-03-01T00:00:00", "end": "2024-03-31T23:59:59", }, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 1 booking_data = bookings[0] # All fields should be present for admin assert booking_data["id"] == booking.id assert booking_data["user_id"] == test_user.id assert booking_data["space_id"] == test_space.id assert booking_data["title"] == "Executive Meeting" assert booking_data["description"] == "Confidential strategic planning" assert booking_data["status"] == "approved" assert booking_data["approved_by"] == test_admin.id assert "start_datetime" in booking_data assert "end_datetime" in booking_data assert "created_at" in booking_data def test_get_bookings_filters_by_time_range( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that bookings are filtered by the provided time range.""" # Create bookings in different time periods booking1 = Booking( user_id=test_user.id, space_id=test_space.id, title="March Meeting", start_datetime=datetime(2024, 3, 15, 10, 0, 0), end_datetime=datetime(2024, 3, 15, 12, 0, 0), status="approved", ) booking2 = Booking( user_id=test_user.id, space_id=test_space.id, title="April Meeting", start_datetime=datetime(2024, 4, 10, 10, 0, 0), end_datetime=datetime(2024, 4, 10, 12, 0, 0), status="approved", ) booking3 = Booking( user_id=test_user.id, space_id=test_space.id, title="May Meeting", start_datetime=datetime(2024, 5, 5, 10, 0, 0), end_datetime=datetime(2024, 5, 5, 12, 0, 0), status="approved", ) db.add_all([booking1, booking2, booking3]) db.commit() # Request only April bookings response = client.get( f"/api/spaces/{test_space.id}/bookings", params={ "start": "2024-04-01T00:00:00", "end": "2024-04-30T23:59:59", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 1 assert bookings[0]["title"] == "April Meeting" def test_get_bookings_space_not_found( client: TestClient, user_token: str, ) -> None: """Test that 404 is returned for non-existent space.""" response = client.get( "/api/spaces/99999/bookings", params={ "start": "2024-03-01T00:00:00", "end": "2024-03-31T23:59:59", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 404 assert "not found" in response.json()["detail"].lower() def test_get_bookings_requires_authentication( client: TestClient, test_space: Space, ) -> None: """Test that authentication is required to get bookings.""" response = client.get( f"/api/spaces/{test_space.id}/bookings", params={ "start": "2024-03-01T00:00:00", "end": "2024-03-31T23:59:59", }, ) # HTTPBearer returns 403 when no Authorization header is provided assert response.status_code == 403 # ===== POST /api/bookings Tests ===== def test_create_booking_success( client: TestClient, user_token: str, test_space: Space, db: Session, ) -> None: """Test successful booking creation.""" booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", "title": "Team Planning Session", "description": "Q3 planning and retrospective", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 201 data = response.json() # Verify response fields assert data["space_id"] == test_space.id assert data["title"] == "Team Planning Session" assert data["description"] == "Q3 planning and retrospective" assert data["status"] == "pending" assert "id" in data assert "user_id" in data assert "created_at" in data # Verify booking was saved in database booking = db.query(Booking).filter(Booking.id == data["id"]).first() assert booking is not None assert booking.title == "Team Planning Session" assert booking.status == "pending" def test_create_booking_validation_fails_duration( client: TestClient, user_token: str, test_space: Space, db: Session, ) -> None: """Test booking creation fails when duration is invalid.""" # Create a booking with only 15 minutes (below min_duration_minutes=30) booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T10:15:00", "title": "Short Meeting", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 400 assert "Durata rezervării" in response.json()["detail"] def test_create_booking_validation_fails_overlap( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test booking creation fails when there's an overlapping booking.""" # Create an existing approved booking existing_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Existing Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", ) db.add(existing_booking) db.commit() # Try to create a new booking that overlaps booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T11:00:00", "end_datetime": "2024-06-15T13:00:00", "title": "Conflicting Meeting", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 400 assert "deja rezervat" in response.json()["detail"] def test_create_booking_space_not_found( client: TestClient, user_token: str, ) -> None: """Test booking creation fails when space doesn't exist.""" booking_data = { "space_id": 99999, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", "title": "Meeting in Non-existent Space", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 404 assert "not found" in response.json()["detail"].lower() def test_create_booking_validation_fails_working_hours( client: TestClient, user_token: str, test_space: Space, ) -> None: """Test booking creation fails when outside working hours.""" # Try to book at 6 AM (before working_hours_start=8) booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T06:00:00", "end_datetime": "2024-06-15T07:00:00", "title": "Early Morning Meeting", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 400 assert "Rezervările sunt permise" in response.json()["detail"] def test_create_booking_validation_fails_max_daily_limit( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test booking creation fails when user exceeds daily booking limit.""" # Create 3 approved bookings for the same day (max_bookings_per_day_per_user=3) for hour in [9, 11, 13]: booking = Booking( user_id=test_user.id, space_id=test_space.id, title=f"Meeting at {hour}:00", start_datetime=datetime(2024, 6, 15, hour, 0, 0), end_datetime=datetime(2024, 6, 15, hour + 1, 0, 0), status="approved", ) db.add(booking) db.commit() # Try to create a 4th booking on the same day booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T15:00:00", "end_datetime": "2024-06-15T16:00:00", "title": "Fourth Meeting", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 400 assert "limita de" in response.json()["detail"] def test_create_booking_requires_authentication( client: TestClient, test_space: Space, ) -> None: """Test that authentication is required to create bookings.""" booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", "title": "Unauthenticated Meeting", } response = client.post("/api/bookings", json=booking_data) # HTTPBearer returns 403 when no Authorization header is provided assert response.status_code == 403 # ===== GET /api/bookings/my Tests ===== def test_get_my_bookings_returns_user_bookings( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that GET /api/bookings/my returns only current user's bookings with space details.""" # Create bookings for test_user booking1 = Booking( user_id=test_user.id, space_id=test_space.id, title="My First Meeting", description="Team sync", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", created_at=datetime(2024, 6, 1, 10, 0, 0), ) booking2 = Booking( user_id=test_user.id, space_id=test_space.id, title="My Second Meeting", description="Client meeting", start_datetime=datetime(2024, 6, 16, 14, 0, 0), end_datetime=datetime(2024, 6, 16, 16, 0, 0), status="pending", created_at=datetime(2024, 6, 2, 10, 0, 0), ) # Create booking for another user (should NOT be returned) other_user = User( full_name="Other User", email="other@example.com", hashed_password="dummy", role="user", ) db.add(other_user) db.flush() booking3 = Booking( user_id=other_user.id, space_id=test_space.id, title="Other User Meeting", start_datetime=datetime(2024, 6, 17, 10, 0, 0), end_datetime=datetime(2024, 6, 17, 12, 0, 0), status="approved", created_at=datetime(2024, 6, 3, 10, 0, 0), ) db.add_all([booking1, booking2, booking3]) db.commit() # Request user's bookings response = client.get( "/api/bookings/my", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() # Should return only 2 bookings (test_user's bookings) assert len(bookings) == 2 # Check that bookings are sorted by created_at desc (most recent first) assert bookings[0]["title"] == "My Second Meeting" assert bookings[1]["title"] == "My First Meeting" # Verify space details are included for booking in bookings: assert "space" in booking assert booking["space"]["id"] == test_space.id assert booking["space"]["name"] == test_space.name assert booking["space"]["type"] == test_space.type # Verify all required fields are present first_booking = bookings[0] assert first_booking["id"] == booking2.id assert first_booking["space_id"] == test_space.id assert first_booking["title"] == "My Second Meeting" assert first_booking["description"] == "Client meeting" assert first_booking["status"] == "pending" assert "start_datetime" in first_booking assert "end_datetime" in first_booking assert "created_at" in first_booking def test_get_my_bookings_filter_by_status( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that status filter works correctly.""" # Create bookings with different statuses booking1 = Booking( user_id=test_user.id, space_id=test_space.id, title="Pending Booking", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime(2024, 6, 1, 10, 0, 0), ) booking2 = Booking( user_id=test_user.id, space_id=test_space.id, title="Approved Booking", start_datetime=datetime(2024, 6, 16, 10, 0, 0), end_datetime=datetime(2024, 6, 16, 12, 0, 0), status="approved", created_at=datetime(2024, 6, 2, 10, 0, 0), ) booking3 = Booking( user_id=test_user.id, space_id=test_space.id, title="Rejected Booking", start_datetime=datetime(2024, 6, 17, 10, 0, 0), end_datetime=datetime(2024, 6, 17, 12, 0, 0), status="rejected", rejection_reason="Room unavailable", created_at=datetime(2024, 6, 3, 10, 0, 0), ) db.add_all([booking1, booking2, booking3]) db.commit() # Test filter by "pending" response = client.get( "/api/bookings/my?status=pending", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 1 assert bookings[0]["title"] == "Pending Booking" assert bookings[0]["status"] == "pending" # Test filter by "approved" response = client.get( "/api/bookings/my?status=approved", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 1 assert bookings[0]["title"] == "Approved Booking" assert bookings[0]["status"] == "approved" # Test filter by "rejected" response = client.get( "/api/bookings/my?status=rejected", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 1 assert bookings[0]["title"] == "Rejected Booking" assert bookings[0]["status"] == "rejected" # Test without filter - should return all 3 response = client.get( "/api/bookings/my", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 bookings = response.json() assert len(bookings) == 3 # ===== GET /api/admin/bookings/pending Tests ===== def test_get_pending_bookings_admin_only( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that admin can retrieve all pending bookings with full details.""" # Create multiple pending bookings booking1 = Booking( user_id=test_user.id, space_id=test_space.id, title="Pending Meeting 1", description="First pending request", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="pending", created_at=datetime(2024, 6, 1, 9, 0, 0), ) booking2 = Booking( user_id=test_user.id, space_id=test_space.id, title="Pending Meeting 2", description="Second pending request", start_datetime=datetime(2024, 6, 21, 14, 0, 0), end_datetime=datetime(2024, 6, 21, 16, 0, 0), status="pending", created_at=datetime(2024, 6, 1, 10, 0, 0), ) # Create an approved booking that should NOT appear booking3 = Booking( user_id=test_user.id, space_id=test_space.id, title="Approved Meeting", start_datetime=datetime(2024, 6, 22, 10, 0, 0), end_datetime=datetime(2024, 6, 22, 12, 0, 0), status="approved", created_at=datetime(2024, 6, 1, 11, 0, 0), ) db.add_all([booking1, booking2, booking3]) db.commit() db.refresh(booking1) db.refresh(booking2) response = client.get( "/api/admin/bookings/pending", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Should return only the 2 pending bookings assert len(data) == 2 # Should be ordered by created_at ascending (FIFO - oldest first) assert data[0]["id"] == booking1.id assert data[0]["title"] == "Pending Meeting 1" assert data[0]["status"] == "pending" assert data[0]["user"]["full_name"] == test_user.full_name assert data[0]["user"]["email"] == test_user.email assert data[0]["space"]["name"] == test_space.name assert data[0]["space"]["type"] == test_space.type assert data[1]["id"] == booking2.id assert data[1]["title"] == "Pending Meeting 2" def test_get_pending_bookings_filter_by_space( client: TestClient, admin_token: str, test_user: User, db: Session, ) -> None: """Test filtering pending bookings by space_id.""" # Create two spaces space1 = Space( name="Conference Room A", type="sala", capacity=10, is_active=True, ) space2 = Space( name="Conference Room B", type="sala", capacity=8, is_active=True, ) db.add_all([space1, space2]) db.commit() db.refresh(space1) db.refresh(space2) # Create pending bookings for both spaces booking1 = Booking( user_id=test_user.id, space_id=space1.id, title="Meeting in Room A", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="pending", ) booking2 = Booking( user_id=test_user.id, space_id=space2.id, title="Meeting in Room B", start_datetime=datetime(2024, 6, 21, 14, 0, 0), end_datetime=datetime(2024, 6, 21, 16, 0, 0), status="pending", ) db.add_all([booking1, booking2]) db.commit() db.refresh(booking1) # Filter by space1 response = client.get( f"/api/admin/bookings/pending?space_id={space1.id}", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Should return only booking for space1 assert len(data) == 1 assert data[0]["id"] == booking1.id assert data[0]["space_id"] == space1.id assert data[0]["title"] == "Meeting in Room A" def test_get_pending_bookings_non_admin_forbidden( client: TestClient, user_token: str, ) -> None: """Test that non-admin users cannot access pending bookings endpoint.""" response = client.get( "/api/admin/bookings/pending", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 403 assert "permissions" in response.json()["detail"].lower() # ===== PUT /api/admin/bookings/{id}/approve Tests ===== def test_approve_booking_success( client: TestClient, admin_token: str, test_admin: User, test_space: Space, test_user: User, db: Session, ) -> None: """Test successful booking approval.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Approve the booking response = client.put( f"/api/admin/bookings/{booking.id}/approve", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["status"] == "approved" assert data["user_id"] == test_user.id # Verify database update db.refresh(booking) assert booking.status == "approved" assert booking.approved_by == test_admin.id def test_approve_booking_conflict_overlap( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that approval detects overlapping bookings (race condition protection).""" # Create an existing approved booking existing_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Existing Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(existing_booking) db.commit() # Create a pending booking that overlaps (simulating race condition) pending_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Conflicting Meeting", start_datetime=datetime(2024, 6, 15, 11, 0, 0), end_datetime=datetime(2024, 6, 15, 13, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(pending_booking) db.commit() db.refresh(pending_booking) # Try to approve the overlapping booking response = client.put( f"/api/admin/bookings/{pending_booking.id}/approve", headers={"Authorization": f"Bearer {admin_token}"}, ) # Should get 409 Conflict assert response.status_code == 409 assert "deja rezervat" in response.json()["detail"] # Verify booking is still pending db.refresh(pending_booking) assert pending_booking.status == "pending" assert pending_booking.approved_by is None def test_approve_booking_not_found( client: TestClient, admin_token: str, ) -> None: """Test approval of non-existent booking returns 404.""" response = client.put( "/api/admin/bookings/99999/approve", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 404 assert "not found" in response.json()["detail"].lower() def test_approve_booking_not_pending( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that only pending bookings can be approved.""" # Create an already approved booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Already Approved", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to approve again response = client.put( f"/api/admin/bookings/{booking.id}/approve", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 400 assert "Cannot approve" in response.json()["detail"] def test_approve_booking_non_admin_forbidden( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that non-admin users cannot approve bookings.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to approve with user token response = client.put( f"/api/admin/bookings/{booking.id}/approve", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 403 assert "permissions" in response.json()["detail"].lower() # ===== PUT /api/admin/bookings/{id}/reject Tests ===== def test_reject_booking_with_reason( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test successful booking rejection with reason.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Reject the booking with reason response = client.put( f"/api/admin/bookings/{booking.id}/reject", json={"reason": "Space maintenance scheduled"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["status"] == "rejected" # Verify database update db.refresh(booking) assert booking.status == "rejected" assert booking.rejection_reason == "Space maintenance scheduled" def test_reject_booking_without_reason( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test successful booking rejection without reason.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Reject the booking without reason response = client.put( f"/api/admin/bookings/{booking.id}/reject", json={}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["status"] == "rejected" # Verify database update db.refresh(booking) assert booking.status == "rejected" assert booking.rejection_reason is None def test_reject_booking_not_found( client: TestClient, admin_token: str, ) -> None: """Test rejection of non-existent booking returns 404.""" response = client.put( "/api/admin/bookings/99999/reject", json={}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 404 assert "not found" in response.json()["detail"].lower() def test_reject_booking_not_pending( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that only pending bookings can be rejected.""" # Create an already rejected booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Already Rejected", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="rejected", rejection_reason="Previous rejection", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to reject again response = client.put( f"/api/admin/bookings/{booking.id}/reject", json={"reason": "Another reason"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 400 assert "Cannot reject" in response.json()["detail"] def test_reject_booking_non_admin_forbidden( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that non-admin users cannot reject bookings.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to reject with user token response = client.put( f"/api/admin/bookings/{booking.id}/reject", json={"reason": "Not authorized"}, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 403 assert "permissions" in response.json()["detail"].lower() # ===== PUT /api/admin/bookings/{id}/cancel Tests ===== def test_admin_cancel_booking_with_reason( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test admin can cancel any booking with a reason.""" # Create an approved booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Cancel the booking with reason response = client.put( f"/api/admin/bookings/{booking.id}/cancel", json={"cancellation_reason": "Emergency maintenance required"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["status"] == "canceled" # Verify database update db.refresh(booking) assert booking.status == "canceled" assert booking.cancellation_reason == "Emergency maintenance required" def test_admin_cancel_booking_without_reason( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test admin can cancel any booking without a reason.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Client Meeting", start_datetime=datetime(2024, 6, 16, 14, 0, 0), end_datetime=datetime(2024, 6, 16, 16, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Cancel the booking without reason response = client.put( f"/api/admin/bookings/{booking.id}/cancel", json={}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["status"] == "canceled" # Verify database update db.refresh(booking) assert booking.status == "canceled" assert booking.cancellation_reason is None # ===== PUT /api/bookings/{id}/cancel Tests (User Cancel) ===== def test_user_cancel_booking_success( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that user can successfully cancel their own booking if enough time before start.""" from app.models.settings import Settings # Create settings with min_hours_before_cancel=24 settings = Settings( id=1, min_duration_minutes=30, max_duration_minutes=480, working_hours_start=8, working_hours_end=20, max_bookings_per_day_per_user=3, min_hours_before_cancel=24, ) db.add(settings) db.commit() # Create a booking that starts in 48 hours start_time = datetime.utcnow() + __import__("datetime").timedelta(hours=48) end_time = start_time + __import__("datetime").timedelta(hours=2) booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Future Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Cancel the booking response = client.put( f"/api/bookings/{booking.id}/cancel", headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["status"] == "canceled" # Verify database update db.refresh(booking) assert booking.status == "canceled" def test_user_cancel_booking_too_late( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that user cannot cancel booking if too close to start time.""" from app.models.settings import Settings # Create settings with min_hours_before_cancel=24 settings = Settings( id=1, min_duration_minutes=30, max_duration_minutes=480, working_hours_start=8, working_hours_end=20, max_bookings_per_day_per_user=3, min_hours_before_cancel=24, ) db.add(settings) db.commit() # Create a booking that starts in 2 hours (less than 24 hours minimum) start_time = datetime.utcnow() + __import__("datetime").timedelta(hours=2) end_time = start_time + __import__("datetime").timedelta(hours=2) booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Soon Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to cancel the booking response = client.put( f"/api/bookings/{booking.id}/cancel", headers={"Authorization": f"Bearer {user_token}"}, ) # Should get 403 Forbidden assert response.status_code == 403 assert "Cannot cancel booking less than" in response.json()["detail"] assert "24 hours" in response.json()["detail"] # Verify booking is still approved db.refresh(booking) assert booking.status == "approved" def test_user_cancel_booking_not_owner( client: TestClient, user_token: str, test_space: Space, db: Session, ) -> None: """Test that user cannot cancel another user's booking.""" from app.models.settings import Settings # Create settings settings = Settings( id=1, min_duration_minutes=30, max_duration_minutes=480, working_hours_start=8, working_hours_end=20, max_bookings_per_day_per_user=3, min_hours_before_cancel=24, ) db.add(settings) db.commit() # Create another user from app.core.security import get_password_hash other_user = User( email="other@example.com", full_name="Other User", hashed_password=get_password_hash("password"), role="user", is_active=True, ) db.add(other_user) db.flush() # Create a booking for the other user start_time = datetime.utcnow() + __import__("datetime").timedelta(hours=48) end_time = start_time + __import__("datetime").timedelta(hours=2) booking = Booking( user_id=other_user.id, space_id=test_space.id, title="Other User Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to cancel another user's booking response = client.put( f"/api/bookings/{booking.id}/cancel", headers={"Authorization": f"Bearer {user_token}"}, ) # Should get 403 Forbidden assert response.status_code == 403 assert "your own bookings" in response.json()["detail"].lower() # Verify booking is still approved db.refresh(booking) assert booking.status == "approved" # ===== PUT /api/bookings/{id} Tests (User Edit) ===== def test_user_edit_pending_booking( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that user can successfully edit their own pending booking.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Original Title", description="Original description", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Update the booking update_data = { "title": "Updated Title", "description": "Updated description", "start_datetime": "2024-06-20T14:00:00", "end_datetime": "2024-06-20T16:00:00", } response = client.put( f"/api/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["title"] == "Updated Title" assert data["description"] == "Updated description" assert data["status"] == "pending" # Status should remain pending # Verify database update db.refresh(booking) assert booking.title == "Updated Title" assert booking.description == "Updated description" def test_user_cannot_edit_approved_booking( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that user cannot edit an approved booking.""" # Create an approved booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Approved Meeting", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to update the approved booking update_data = { "title": "Updated Title", } response = client.put( f"/api/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {user_token}"}, ) # Should get 400 Bad Request assert response.status_code == 400 assert "Can only edit pending bookings" in response.json()["detail"] # Verify booking was not changed db.refresh(booking) assert booking.title == "Approved Meeting" def test_user_cannot_edit_others_booking( client: TestClient, user_token: str, test_space: Space, db: Session, ) -> None: """Test that user cannot edit another user's booking.""" from app.core.security import get_password_hash # Create another user other_user = User( email="other@example.com", full_name="Other User", hashed_password=get_password_hash("password"), role="user", is_active=True, ) db.add(other_user) db.flush() # Create a pending booking for the other user booking = Booking( user_id=other_user.id, space_id=test_space.id, title="Other User Booking", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to update another user's booking update_data = { "title": "Hacked Title", } response = client.put( f"/api/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {user_token}"}, ) # Should get 403 Forbidden assert response.status_code == 403 assert "Can only edit your own bookings" in response.json()["detail"] # Verify booking was not changed db.refresh(booking) assert booking.title == "Other User Booking" def test_user_edit_booking_validation_fails( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that editing a booking with invalid data fails validation.""" # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Original Title", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to update with invalid duration (15 minutes - below min_duration_minutes=30) update_data = { "start_datetime": "2024-06-20T10:00:00", "end_datetime": "2024-06-20T10:15:00", } response = client.put( f"/api/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {user_token}"}, ) # Should get 400 Bad Request assert response.status_code == 400 assert "Durata rezervării" in response.json()["detail"] # Verify booking was not changed db.refresh(booking) assert booking.start_datetime == datetime(2024, 6, 20, 10, 0, 0) assert booking.end_datetime == datetime(2024, 6, 20, 12, 0, 0) # ===== PUT /api/admin/bookings/{id} Tests (Admin Edit) ===== def test_admin_edit_any_booking( client: TestClient, admin_token: str, test_admin: User, test_user: User, test_space: Space, db: Session, ) -> None: """Test that admin can edit any booking (pending or approved).""" # Create a pending booking for a user booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Original Title", description="Original description", start_datetime=datetime(2024, 6, 20, 10, 0, 0), end_datetime=datetime(2024, 6, 20, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Admin updates the booking update_data = { "title": "Admin Updated Title", "description": "Admin updated description", } response = client.put( f"/api/admin/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["title"] == "Admin Updated Title" assert data["description"] == "Admin updated description" # Verify database update db.refresh(booking) assert booking.title == "Admin Updated Title" assert booking.description == "Admin updated description" # Verify audit log was created from app.models.audit_log import AuditLog audit = db.query(AuditLog).filter( AuditLog.action == "booking_updated", AuditLog.target_id == booking.id ).first() assert audit is not None assert audit.user_id == test_admin.id assert audit.details == {"updated_by": "admin"} def test_admin_cannot_edit_started_booking( client: TestClient, admin_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that admin cannot edit bookings that already started.""" # Create an approved booking that already started start_time = datetime.utcnow() - __import__("datetime").timedelta(hours=1) end_time = datetime.utcnow() + __import__("datetime").timedelta(hours=1) booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Started Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to update the started booking update_data = { "title": "Updated Title", } response = client.put( f"/api/admin/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {admin_token}"}, ) # Should get 400 Bad Request assert response.status_code == 400 assert "Cannot edit bookings that already started" in response.json()["detail"] # Verify booking was not changed db.refresh(booking) assert booking.title == "Started Meeting" def test_admin_edit_approved_booking_future( client: TestClient, admin_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that admin can edit approved bookings that haven't started yet.""" # Create an approved booking in the future start_time = datetime.utcnow() + __import__("datetime").timedelta(hours=24) end_time = start_time + __import__("datetime").timedelta(hours=2) booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Future Approved Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Admin updates the booking update_data = { "title": "Admin Updated Future Meeting", } response = client.put( f"/api/admin/bookings/{booking.id}", json=update_data, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["title"] == "Admin Updated Future Meeting" # Verify database update db.refresh(booking) assert booking.title == "Admin Updated Future Meeting" # ===== POST /api/admin/bookings Tests (Admin Direct Booking) ===== def test_admin_create_booking_success( client: TestClient, admin_token: str, test_admin: User, test_user: User, test_space: Space, db: Session, ) -> None: """Test successful admin direct booking creation (bypass approval).""" booking_data = { "space_id": test_space.id, "user_id": test_user.id, "start_datetime": "2024-07-15T10:00:00", "end_datetime": "2024-07-15T12:00:00", "title": "Admin Direct Booking", "description": "VIP client meeting", } response = client.post( "/api/admin/bookings", json=booking_data, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 201 data = response.json() # Verify response fields assert data["space_id"] == test_space.id assert data["user_id"] == test_user.id assert data["title"] == "Admin Direct Booking" assert data["description"] == "VIP client meeting" assert data["status"] == "approved" # Should be approved, not pending assert "id" in data assert "created_at" in data # Verify booking was saved in database with approved status booking = db.query(Booking).filter(Booking.id == data["id"]).first() assert booking is not None assert booking.title == "Admin Direct Booking" assert booking.status == "approved" assert booking.approved_by == test_admin.id def test_admin_create_booking_overlap_error( client: TestClient, admin_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test admin direct booking fails when overlapping with existing approved booking.""" # Create an existing approved booking existing_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Existing Approved Booking", start_datetime=datetime(2024, 7, 20, 10, 0, 0), end_datetime=datetime(2024, 7, 20, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(existing_booking) db.commit() # Try to create admin booking that overlaps booking_data = { "space_id": test_space.id, "user_id": test_user.id, "start_datetime": "2024-07-20T11:00:00", "end_datetime": "2024-07-20T13:00:00", "title": "Conflicting Admin Booking", } response = client.post( "/api/admin/bookings", json=booking_data, headers={"Authorization": f"Bearer {admin_token}"}, ) # Should get 400 Bad Request for overlap assert response.status_code == 400 assert "deja rezervat" in response.json()["detail"] def test_admin_create_booking_outside_operating_hours( client: TestClient, admin_token: str, test_user: User, test_space: Space, ) -> None: """Test admin direct booking fails when outside operating hours.""" # Try to book outside working hours (working_hours_start=8, working_hours_end=20) booking_data = { "space_id": test_space.id, "user_id": test_user.id, "start_datetime": "2024-07-15T06:00:00", # 6 AM - before working hours "end_datetime": "2024-07-15T07:00:00", # 7 AM "title": "Early Morning Admin Booking", } response = client.post( "/api/admin/bookings", json=booking_data, headers={"Authorization": f"Bearer {admin_token}"}, ) # Should get 400 Bad Request for working hours violation assert response.status_code == 400 assert "Rezervările sunt permise" in response.json()["detail"] # ===== Notification Integration Tests ===== def test_booking_creation_notifies_admins( client: TestClient, user_token: str, test_space: Space, test_user: User, test_admin: User, db: Session, ) -> None: """Test that creating a booking notifies all admin users.""" from app.models.notification import Notification # Create another admin user from app.core.security import get_password_hash admin2 = User( email="admin2@example.com", full_name="Second Admin", hashed_password=get_password_hash("password"), role="admin", is_active=True, ) db.add(admin2) db.commit() db.refresh(admin2) # Count existing notifications initial_count = db.query(Notification).count() # Create a booking booking_data = { "space_id": test_space.id, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", "title": "Team Planning Session", "description": "Q3 planning and retrospective", } response = client.post( "/api/bookings", json=booking_data, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 201 booking_id = response.json()["id"] # Check that notifications were created for both admins notifications = db.query(Notification).filter( Notification.booking_id == booking_id ).all() assert len(notifications) == 2 # One for each admin # Verify notification details admin_ids = {test_admin.id, admin2.id} notified_admin_ids = {notif.user_id for notif in notifications} assert notified_admin_ids == admin_ids for notif in notifications: assert notif.type == "booking_created" assert notif.title == "Noua Cerere de Rezervare" assert test_user.full_name in notif.message assert test_space.name in notif.message assert "15.06.2024 10:00" in notif.message assert notif.is_read is False def test_booking_approval_notifies_user( client: TestClient, admin_token: str, test_admin: User, test_space: Space, test_user: User, db: Session, ) -> None: """Test that approving a booking notifies the user.""" from app.models.notification import Notification # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Approve the booking response = client.put( f"/api/admin/bookings/{booking.id}/approve", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 # Check that notification was created for the user notifications = db.query(Notification).filter( Notification.user_id == test_user.id, Notification.booking_id == booking.id, ).all() assert len(notifications) == 1 notif = notifications[0] assert notif.type == "booking_approved" assert notif.title == "Rezervare Aprobată" assert test_space.name in notif.message assert "15.06.2024 10:00" in notif.message assert "aprobată" in notif.message assert notif.is_read is False def test_booking_rejection_notifies_user( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that rejecting a booking notifies the user with rejection reason.""" from app.models.notification import Notification # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Reject the booking with reason response = client.put( f"/api/admin/bookings/{booking.id}/reject", json={"reason": "Space maintenance scheduled"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 # Check that notification was created for the user notifications = db.query(Notification).filter( Notification.user_id == test_user.id, Notification.booking_id == booking.id, ).all() assert len(notifications) == 1 notif = notifications[0] assert notif.type == "booking_rejected" assert notif.title == "Rezervare Respinsă" assert test_space.name in notif.message assert "15.06.2024 10:00" in notif.message assert "respinsă" in notif.message assert "Space maintenance scheduled" in notif.message assert notif.is_read is False def test_admin_cancel_notifies_user( client: TestClient, admin_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test that admin canceling a booking notifies the user with cancellation reason.""" from app.models.notification import Notification # Create an approved booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Cancel the booking with reason response = client.put( f"/api/admin/bookings/{booking.id}/cancel", json={"cancellation_reason": "Emergency maintenance required"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 # Check that notification was created for the user notifications = db.query(Notification).filter( Notification.user_id == test_user.id, Notification.booking_id == booking.id, ).all() assert len(notifications) == 1 notif = notifications[0] assert notif.type == "booking_canceled" assert notif.title == "Rezervare Anulată" assert test_space.name in notif.message assert "15.06.2024 10:00" in notif.message assert "anulată" in notif.message assert "Emergency maintenance required" in notif.message assert notif.is_read is False # ===== Audit Log Integration Tests ===== def test_booking_approval_creates_audit_log( client: TestClient, admin_token: str, test_admin: User, test_space: Space, test_user: User, db: Session, ) -> None: """Test that approving a booking creates an audit log entry.""" from app.models.audit_log import AuditLog # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Approve the booking response = client.put( f"/api/admin/bookings/{booking.id}/approve", headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 # Check audit log was created audit = db.query(AuditLog).filter( AuditLog.action == "booking_approved", AuditLog.target_id == booking.id ).first() assert audit is not None assert audit.target_type == "booking" assert audit.user_id == test_admin.id assert audit.details == {} # Empty dict when no details provided def test_booking_rejection_creates_audit_log( client: TestClient, admin_token: str, test_admin: User, test_space: Space, test_user: User, db: Session, ) -> None: """Test that rejecting a booking creates an audit log entry with reason.""" from app.models.audit_log import AuditLog # Create a pending booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Reject the booking response = client.put( f"/api/admin/bookings/{booking.id}/reject", json={"reason": "Space maintenance"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 # Check audit log was created with rejection reason audit = db.query(AuditLog).filter( AuditLog.action == "booking_rejected", AuditLog.target_id == booking.id ).first() assert audit is not None assert audit.target_type == "booking" assert audit.user_id == test_admin.id assert audit.details == {"rejection_reason": "Space maintenance"} def test_booking_cancellation_creates_audit_log( client: TestClient, admin_token: str, test_admin: User, test_space: Space, test_user: User, db: Session, ) -> None: """Test that admin canceling a booking creates an audit log entry.""" from app.models.audit_log import AuditLog # Create an approved booking booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Cancel the booking response = client.put( f"/api/admin/bookings/{booking.id}/cancel", json={"cancellation_reason": "Emergency maintenance"}, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 200 # Check audit log was created with cancellation reason audit = db.query(AuditLog).filter( AuditLog.action == "booking_canceled", AuditLog.target_id == booking.id ).first() assert audit is not None assert audit.target_type == "booking" assert audit.user_id == test_admin.id assert audit.details == {"cancellation_reason": "Emergency maintenance"} def test_check_availability_no_conflicts( client: TestClient, user_token: str, test_space: Space, ) -> None: """Test availability check with no conflicts.""" response = client.get( "/api/bookings/check-availability", params={ "space_id": test_space.id, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["available"] is True assert len(data["conflicts"]) == 0 assert data["message"] == "Time slot is available" def test_check_availability_with_pending_conflict( client: TestClient, user_token: str, test_space: Space, test_user: User, db: Session, ) -> None: """Test availability check with pending conflict.""" # Create a pending booking pending_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Pending Meeting", description="Test pending", start_datetime=datetime(2024, 6, 15, 10, 0, 0), end_datetime=datetime(2024, 6, 15, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add(pending_booking) db.commit() db.refresh(pending_booking) # Check availability for overlapping time response = client.get( "/api/bookings/check-availability", params={ "space_id": test_space.id, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["available"] is True # Still available (only pending) assert len(data["conflicts"]) == 1 assert "pending" in data["message"].lower() conflict = data["conflicts"][0] assert conflict["id"] == pending_booking.id assert conflict["status"] == "pending" assert conflict["title"] == "Pending Meeting" def test_check_availability_with_approved_conflict( client: TestClient, user_token: str, test_space: Space, test_user: User, test_admin: User, db: Session, ) -> None: """Test availability check with approved conflict.""" # Create an approved booking approved_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Approved Meeting", description="Test approved", start_datetime=datetime(2024, 6, 15, 14, 0, 0), end_datetime=datetime(2024, 6, 15, 16, 0, 0), status="approved", approved_by=test_admin.id, created_at=datetime.utcnow(), ) db.add(approved_booking) db.commit() db.refresh(approved_booking) # Check availability for overlapping time response = client.get( "/api/bookings/check-availability", params={ "space_id": test_space.id, "start_datetime": "2024-06-15T14:00:00", "end_datetime": "2024-06-15T16:00:00", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["available"] is False # Not available (approved conflict) assert len(data["conflicts"]) == 1 assert "approved" in data["message"].lower() conflict = data["conflicts"][0] assert conflict["id"] == approved_booking.id assert conflict["status"] == "approved" assert conflict["title"] == "Approved Meeting" assert conflict["user_name"] == test_user.full_name def test_check_availability_with_multiple_conflicts( client: TestClient, user_token: str, test_space: Space, test_user: User, test_admin: User, db: Session, ) -> None: """Test availability check with both pending and approved conflicts.""" # Create an approved booking approved_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Approved Meeting", start_datetime=datetime(2024, 6, 16, 10, 0, 0), end_datetime=datetime(2024, 6, 16, 11, 0, 0), status="approved", approved_by=test_admin.id, created_at=datetime.utcnow(), ) # Create a pending booking pending_booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Pending Meeting", start_datetime=datetime(2024, 6, 16, 11, 0, 0), end_datetime=datetime(2024, 6, 16, 12, 0, 0), status="pending", created_at=datetime.utcnow(), ) db.add_all([approved_booking, pending_booking]) db.commit() # Check availability for time that overlaps both response = client.get( "/api/bookings/check-availability", params={ "space_id": test_space.id, "start_datetime": "2024-06-16T10:30:00", "end_datetime": "2024-06-16T11:30:00", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 200 data = response.json() assert data["available"] is False # Not available (has approved) assert len(data["conflicts"]) == 2 assert "approved" in data["message"].lower() def test_check_availability_invalid_space( client: TestClient, user_token: str, ) -> None: """Test availability check with invalid space ID.""" response = client.get( "/api/bookings/check-availability", params={ "space_id": 99999, "start_datetime": "2024-06-15T10:00:00", "end_datetime": "2024-06-15T12:00:00", }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 404 assert "Space not found" in response.json()["detail"] # ===== PUT /api/admin/bookings/{id}/reschedule Tests ===== def test_admin_reschedule_booking_success( client: TestClient, admin_token: str, test_admin: User, test_user: User, test_space: Space, db: Session, ) -> None: """Test successful booking reschedule (drag-and-drop).""" # Create an approved booking in the future (during working hours) # Use specific date/time to ensure it's within working hours (8:00-20:00) start_time = datetime(2026, 7, 15, 10, 0, 0) # 10 AM (future date) end_time = datetime(2026, 7, 15, 12, 0, 0) # 12 PM booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", description="Q3 Planning", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Reschedule to 2 hours later (still within working hours: 12 PM - 2 PM) new_start = datetime(2026, 7, 15, 12, 0, 0) # 12 PM new_end = datetime(2026, 7, 15, 14, 0, 0) # 2 PM response = client.put( f"/api/admin/bookings/{booking.id}/reschedule", json={ "start_datetime": new_start.isoformat(), "end_datetime": new_end.isoformat(), }, headers={"Authorization": f"Bearer {admin_token}"}, ) if response.status_code != 200: print(f"Error: {response.json()}") assert response.status_code == 200 data = response.json() # Verify response assert data["id"] == booking.id assert data["start_datetime"] == new_start.isoformat() assert data["end_datetime"] == new_end.isoformat() # Verify database update db.refresh(booking) assert booking.start_datetime == new_start assert booking.end_datetime == new_end # Verify audit log was created from app.models.audit_log import AuditLog audit = db.query(AuditLog).filter( AuditLog.action == "booking_rescheduled", AuditLog.target_id == booking.id, ).first() assert audit is not None assert audit.user_id == test_admin.id assert "old_start" in audit.details assert "new_start" in audit.details # Verify notification was sent to user from app.models.notification import Notification notification = db.query(Notification).filter( Notification.user_id == test_user.id, Notification.booking_id == booking.id, Notification.type == "booking_rescheduled", ).first() assert notification is not None assert "Reprogramată" in notification.title def test_admin_reschedule_booking_overlap( client: TestClient, admin_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test reschedule fails when overlapping with another booking.""" # Create two approved bookings start_time1 = datetime.utcnow() + __import__("datetime").timedelta(hours=48) end_time1 = start_time1 + __import__("datetime").timedelta(hours=2) booking1 = Booking( user_id=test_user.id, space_id=test_space.id, title="First Meeting", start_datetime=start_time1, end_datetime=end_time1, status="approved", created_at=datetime.utcnow(), ) start_time2 = datetime.utcnow() + __import__("datetime").timedelta(hours=52) end_time2 = start_time2 + __import__("datetime").timedelta(hours=2) booking2 = Booking( user_id=test_user.id, space_id=test_space.id, title="Second Meeting", start_datetime=start_time2, end_datetime=end_time2, status="approved", created_at=datetime.utcnow(), ) db.add_all([booking1, booking2]) db.commit() db.refresh(booking1) db.refresh(booking2) # Try to reschedule booking1 to overlap with booking2 response = client.put( f"/api/admin/bookings/{booking1.id}/reschedule", json={ "start_datetime": start_time2.isoformat(), "end_datetime": end_time2.isoformat(), }, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 400 assert "deja rezervat" in response.json()["detail"].lower() # Verify booking was not changed db.refresh(booking1) assert booking1.start_datetime == start_time1 def test_admin_reschedule_past_booking( client: TestClient, admin_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test cannot reschedule bookings that already started.""" # Create a booking that already started start_time = datetime.utcnow() - __import__("datetime").timedelta(hours=1) end_time = datetime.utcnow() + __import__("datetime").timedelta(hours=1) booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Started Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to reschedule new_start = datetime.utcnow() + __import__("datetime").timedelta(hours=2) new_end = new_start + __import__("datetime").timedelta(hours=2) response = client.put( f"/api/admin/bookings/{booking.id}/reschedule", json={ "start_datetime": new_start.isoformat(), "end_datetime": new_end.isoformat(), }, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 400 assert "already started" in response.json()["detail"] def test_admin_reschedule_booking_not_found( client: TestClient, admin_token: str, ) -> None: """Test reschedule of non-existent booking returns 404.""" new_start = datetime.utcnow() + __import__("datetime").timedelta(hours=2) new_end = new_start + __import__("datetime").timedelta(hours=2) response = client.put( "/api/admin/bookings/99999/reschedule", json={ "start_datetime": new_start.isoformat(), "end_datetime": new_end.isoformat(), }, headers={"Authorization": f"Bearer {admin_token}"}, ) assert response.status_code == 404 assert "not found" in response.json()["detail"].lower() def test_admin_reschedule_booking_non_admin_forbidden( client: TestClient, user_token: str, test_user: User, test_space: Space, db: Session, ) -> None: """Test that non-admin users cannot reschedule bookings.""" # Create an approved booking start_time = datetime.utcnow() + __import__("datetime").timedelta(hours=48) end_time = start_time + __import__("datetime").timedelta(hours=2) booking = Booking( user_id=test_user.id, space_id=test_space.id, title="Team Meeting", start_datetime=start_time, end_datetime=end_time, status="approved", created_at=datetime.utcnow(), ) db.add(booking) db.commit() db.refresh(booking) # Try to reschedule with user token new_start = start_time + __import__("datetime").timedelta(hours=2) new_end = end_time + __import__("datetime").timedelta(hours=2) response = client.put( f"/api/admin/bookings/{booking.id}/reschedule", json={ "start_datetime": new_start.isoformat(), "end_datetime": new_end.isoformat(), }, headers={"Authorization": f"Bearer {user_token}"}, ) assert response.status_code == 403 assert "permissions" in response.json()["detail"].lower()