Added full suite of tests & added testing to CI/CD
This commit is contained in:
c-d-p
@@ -1,44 +1,389 @@
|
||||
import pytest
|
||||
from fastapi import status
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from tests.helpers import generators
|
||||
from modules.calendar.models import CalendarEvent # Assuming model exists
|
||||
from tests.conftest import fake
|
||||
|
||||
# Helper function to create an event payload
|
||||
def create_event_payload(start_offset_days=0, end_offset_days=1):
|
||||
start_time = datetime.utcnow() + timedelta(days=start_offset_days)
|
||||
end_time = datetime.utcnow() + timedelta(days=end_offset_days)
|
||||
return {
|
||||
"title": fake.sentence(nb_words=3),
|
||||
"description": fake.text(),
|
||||
"start": start_time.isoformat(), # Rename start_time to start
|
||||
"end": end_time.isoformat(), # Rename end_time to end
|
||||
"all_day": fake.boolean(),
|
||||
}
|
||||
|
||||
# --- Test Create Event ---
|
||||
|
||||
def test_create_event_unauthorized(client: TestClient) -> None:
|
||||
"""Test creating an event without authentication."""
|
||||
payload = create_event_payload()
|
||||
response = client.post("/api/calendar/events", json=payload)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_create_event_success(db: Session, client: TestClient) -> None:
|
||||
"""Test creating a calendar event successfully."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
|
||||
response = client.post(
|
||||
"/api/calendar/events",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
json=payload
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED # Change expected status to 201
|
||||
data = response.json()
|
||||
assert data["title"] == payload["title"]
|
||||
assert data["description"] == payload["description"]
|
||||
# Remove the '+ "Z"' as the API doesn't add it
|
||||
assert data["start"] == payload["start"]
|
||||
assert data["end"] == payload["end"]
|
||||
assert data["all_day"] == payload["all_day"]
|
||||
assert "id" in data
|
||||
assert data["user_id"] == user.id
|
||||
|
||||
# Verify in DB
|
||||
event_in_db = db.query(CalendarEvent).filter(CalendarEvent.id == data["id"]).first()
|
||||
assert event_in_db is not None
|
||||
assert event_in_db.user_id == user.id
|
||||
assert event_in_db.title == payload["title"]
|
||||
|
||||
# --- Test Get Events ---
|
||||
|
||||
def test_get_events_unauthorized(client: TestClient) -> None:
|
||||
"""Test getting events without authentication."""
|
||||
response = client.get("/api/calendar/events")
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_get_events_success(db: Session, client: TestClient) -> None:
|
||||
"""Test getting all calendar events for a user."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
|
||||
# Create a couple of events for the user
|
||||
payload1 = create_event_payload(0, 1)
|
||||
client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload1)
|
||||
payload2 = create_event_payload(2, 3)
|
||||
client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload2)
|
||||
|
||||
# Create an event for another user (should not be returned)
|
||||
other_user, other_password = generators.create_user(db)
|
||||
other_login_rsp = generators.login(db, other_user.username, other_password)
|
||||
other_access_token = other_login_rsp["access_token"]
|
||||
other_payload = create_event_payload(4, 5)
|
||||
client.post("/api/calendar/events", headers={"Authorization": f"Bearer {other_access_token}"}, json=other_payload)
|
||||
|
||||
|
||||
def test_create_event(client: TestClient, db: Session) -> None:
|
||||
user, unhashed_password = generators.create_user(db)
|
||||
rsp = generators.login(db, user.username, unhashed_password)
|
||||
access_token = rsp["access_token"]
|
||||
refresh_token = rsp["refresh_token"]
|
||||
|
||||
response = client.post("/api/calendar/events",
|
||||
json={
|
||||
"title": "Test Event",
|
||||
"start_time": "2024-03-20T15:00:00Z"
|
||||
},
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
cookies={"refresh_token": refresh_token},
|
||||
response = client.get(
|
||||
"/api/calendar/events",
|
||||
headers={"Authorization": f"Bearer {access_token}"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["title"] == "Test Event"
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) == 2
|
||||
assert data[0]["title"] == payload1["title"]
|
||||
assert data[1]["title"] == payload2["title"]
|
||||
assert data[0]["user_id"] == user.id
|
||||
assert data[1]["user_id"] == user.id
|
||||
|
||||
def test_get_events(client: TestClient, db: Session) -> None:
|
||||
user, unhashed_password = generators.create_user(db)
|
||||
rsp = generators.login(db, user.username, unhashed_password)
|
||||
access_token = rsp["access_token"]
|
||||
refresh_token = rsp["refresh_token"]
|
||||
|
||||
# Create an event to retrieve
|
||||
client.post("/api/calendar/events",
|
||||
json={
|
||||
"title": "Test Event",
|
||||
"start_time": "2024-03-20T15:00:00Z"
|
||||
},
|
||||
|
||||
def test_get_events_filtered(db: Session, client: TestClient) -> None:
|
||||
"""Test getting filtered calendar events for a user."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
|
||||
# Create events
|
||||
payload1 = create_event_payload(0, 1) # Today -> Tomorrow
|
||||
client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload1)
|
||||
payload2 = create_event_payload(5, 6) # In 5 days -> In 6 days
|
||||
client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload2)
|
||||
payload3 = create_event_payload(10, 11) # In 10 days -> In 11 days
|
||||
client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload3)
|
||||
|
||||
# Filter for events starting within the next week
|
||||
start_filter = datetime.utcnow().isoformat()
|
||||
end_filter = (datetime.utcnow() + timedelta(days=7)).isoformat()
|
||||
|
||||
response = client.get(
|
||||
"/api/calendar/events",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
cookies={"refresh_token": refresh_token},
|
||||
params={"start": start_filter, "end": end_filter}
|
||||
)
|
||||
|
||||
response = client.get("/api/calendar/events",
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) == 2 # Should get event 1 and 2
|
||||
assert data[0]["title"] == payload1["title"]
|
||||
assert data[1]["title"] == payload2["title"]
|
||||
|
||||
# Filter for events starting after 8 days
|
||||
start_filter_late = (datetime.utcnow() + timedelta(days=8)).isoformat()
|
||||
response = client.get(
|
||||
"/api/calendar/events",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
cookies={"refresh_token": refresh_token},
|
||||
params={"start": start_filter_late}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()) > 0
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert len(data) == 1 # Should get event 3
|
||||
assert data[0]["title"] == payload3["title"]
|
||||
|
||||
|
||||
# --- Test Get Event By ID ---
|
||||
|
||||
def test_get_event_by_id_unauthorized(db: Session, client: TestClient) -> None:
|
||||
"""Test getting a specific event without authentication."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
response = client.get(f"/api/calendar/events/{event_id}")
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_get_event_by_id_success(db: Session, client: TestClient) -> None:
|
||||
"""Test getting a specific event successfully."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
response = client.get(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["id"] == event_id
|
||||
assert data["title"] == payload["title"]
|
||||
assert data["user_id"] == user.id
|
||||
|
||||
def test_get_event_by_id_not_found(db: Session, client: TestClient) -> None:
|
||||
"""Test getting a non-existent event."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
non_existent_id = 99999
|
||||
|
||||
response = client.get(
|
||||
f"/api/calendar/events/{non_existent_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_get_event_by_id_forbidden(db: Session, client: TestClient) -> None:
|
||||
"""Test getting another user's event."""
|
||||
user1, password_user1 = generators.create_user(db)
|
||||
user2, password_user2 = generators.create_user(db)
|
||||
|
||||
# Log in as user1 and create an event
|
||||
login_rsp1 = generators.login(db, user1.username, password_user1)
|
||||
access_token1 = login_rsp1["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token1}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
# Log in as user2 and try to get user1's event
|
||||
login_rsp2 = generators.login(db, user2.username, password_user2)
|
||||
access_token2 = login_rsp2["access_token"]
|
||||
|
||||
response = client.get(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token2}"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND # Service layer returns 404 if user_id doesn't match
|
||||
|
||||
# --- Test Update Event ---
|
||||
|
||||
def test_update_event_unauthorized(db: Session, client: TestClient) -> None:
|
||||
"""Test updating an event without authentication."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
update_payload = {"title": "Updated Title"}
|
||||
|
||||
response = client.patch(f"/api/calendar/events/{event_id}", json=update_payload)
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_update_event_success(db: Session, client: TestClient) -> None:
|
||||
"""Test updating an event successfully."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload)
|
||||
assert create_response.status_code == status.HTTP_201_CREATED # Ensure creation check uses 201
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
update_payload = {
|
||||
"title": "Updated Title",
|
||||
"description": "Updated description.",
|
||||
"all_day": not payload["all_day"] # Toggle all_day
|
||||
}
|
||||
|
||||
response = client.patch(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
json=update_payload
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
data = response.json()
|
||||
assert data["id"] == event_id
|
||||
assert data["title"] == update_payload["title"]
|
||||
assert data["description"] == update_payload["description"]
|
||||
assert data["all_day"] == update_payload["all_day"]
|
||||
assert data["start"] == payload["start"] # Check correct field name 'start'
|
||||
assert data["user_id"] == user.id
|
||||
|
||||
# Verify in DB
|
||||
event_in_db = db.query(CalendarEvent).filter(CalendarEvent.id == event_id).first()
|
||||
assert event_in_db is not None
|
||||
assert event_in_db.title == update_payload["title"]
|
||||
assert event_in_db.description == update_payload["description"]
|
||||
assert event_in_db.all_day == update_payload["all_day"]
|
||||
|
||||
def test_update_event_not_found(db: Session, client: TestClient) -> None:
|
||||
"""Test updating a non-existent event."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
non_existent_id = 99999
|
||||
update_payload = {"title": "Updated Title"}
|
||||
|
||||
response = client.patch(
|
||||
f"/api/calendar/events/{non_existent_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
json=update_payload
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
def test_update_event_forbidden(db: Session, client: TestClient) -> None:
|
||||
"""Test updating another user's event."""
|
||||
user1, password_user1 = generators.create_user(db)
|
||||
user2, password_user2 = generators.create_user(db)
|
||||
|
||||
# Log in as user1 and create an event
|
||||
login_rsp1 = generators.login(db, user1.username, password_user1)
|
||||
access_token1 = login_rsp1["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token1}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
# Log in as user2 and try to update user1's event
|
||||
login_rsp2 = generators.login(db, user2.username, password_user2)
|
||||
access_token2 = login_rsp2["access_token"]
|
||||
update_payload = {"title": "Updated by User 2"}
|
||||
|
||||
response = client.patch(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token2}"},
|
||||
json=update_payload
|
||||
)
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND # Service layer returns 404 if user_id doesn't match
|
||||
|
||||
# --- Test Delete Event ---
|
||||
|
||||
def test_delete_event_unauthorized(db: Session, client: TestClient) -> None:
|
||||
"""Test deleting an event without authentication."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
response = client.delete(f"/api/calendar/events/{event_id}")
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
|
||||
def test_delete_event_success(db: Session, client: TestClient) -> None:
|
||||
"""Test deleting an event successfully."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token}"}, json=payload)
|
||||
assert create_response.status_code == status.HTTP_201_CREATED # Ensure creation check uses 201
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
# Verify event exists before delete
|
||||
event_in_db = db.query(CalendarEvent).filter(CalendarEvent.id == event_id).first()
|
||||
assert event_in_db is not None
|
||||
|
||||
response = client.delete(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_204_NO_CONTENT
|
||||
|
||||
# Verify event is deleted from DB
|
||||
event_in_db = db.query(CalendarEvent).filter(CalendarEvent.id == event_id).first()
|
||||
assert event_in_db is None
|
||||
|
||||
# Try getting the deleted event (should be 404)
|
||||
get_response = client.get(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"}
|
||||
)
|
||||
assert get_response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
def test_delete_event_not_found(db: Session, client: TestClient) -> None:
|
||||
"""Test deleting a non-existent event."""
|
||||
user, password = generators.create_user(db)
|
||||
login_rsp = generators.login(db, user.username, password)
|
||||
access_token = login_rsp["access_token"]
|
||||
non_existent_id = 99999
|
||||
|
||||
response = client.delete(
|
||||
f"/api/calendar/events/{non_existent_id}",
|
||||
headers={"Authorization": f"Bearer {access_token}"}
|
||||
)
|
||||
# The service layer raises NotFound, which should result in 404
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
|
||||
def test_delete_event_forbidden(db: Session, client: TestClient) -> None:
|
||||
"""Test deleting another user's event."""
|
||||
user1, password_user1 = generators.create_user(db)
|
||||
user2, password_user2 = generators.create_user(db)
|
||||
|
||||
# Log in as user1 and create an event
|
||||
login_rsp1 = generators.login(db, user1.username, password_user1)
|
||||
access_token1 = login_rsp1["access_token"]
|
||||
payload = create_event_payload()
|
||||
create_response = client.post("/api/calendar/events", headers={"Authorization": f"Bearer {access_token1}"}, json=payload)
|
||||
event_id = create_response.json()["id"]
|
||||
|
||||
# Log in as user2 and try to delete user1's event
|
||||
login_rsp2 = generators.login(db, user2.username, password_user2)
|
||||
access_token2 = login_rsp2["access_token"]
|
||||
|
||||
response = client.delete(
|
||||
f"/api/calendar/events/{event_id}",
|
||||
headers={"Authorization": f"Bearer {access_token2}"}
|
||||
)
|
||||
# The service layer raises NotFound if user_id doesn't match, resulting in 404
|
||||
assert response.status_code == status.HTTP_404_NOT_FOUND
|
||||
|
||||
# Verify event still exists for user1
|
||||
event_in_db = db.query(CalendarEvent).filter(CalendarEvent.id == event_id).first()
|
||||
assert event_in_db is not None
|
||||
assert event_in_db.user_id == user1.id
|
||||
|
||||
|
||||
Reference in New Issue
Block a user