jeet/grateful-journal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

testing

Browse Source
This commit is contained in:
Jeet Debnath
2026-03-24 10:48:20 +05:30
parent bd1af0bf44
commit 6e425e2f04
21 changed files with 3021 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/pytest.ini Normal file
View File

@@ -0,0 +1,3 @@
[pytest]
pythonpath = .
testpaths = tests

11
backend/requirements.txt
View File

@@ -1,7 +1,12 @@
fastapi==0.104.1
fastapi>=0.115.0
uvicorn==0.24.0
pymongo==4.6.0
pydantic==2.5.0
pydantic>=2.5.0
python-dotenv==1.0.0
pydantic-settings==2.1.0
pydantic-settings>=2.1.0
python-multipart==0.0.6
# Testing
pytest>=7.4.0
httpx>=0.25.0
mongomock>=4.1.2

BIN
backend/routers/__pycache__/__init__.cpython-312.pyc
View File

Binary file not shown.

BIN
backend/routers/__pycache__/entries.cpython-312.pyc
View File

Binary file not shown.

BIN
backend/routers/__pycache__/users.cpython-312.pyc
View File

Binary file not shown.

54
backend/routers/entries.py
View File

@@ -5,6 +5,7 @@ from models import JournalEntryCreate, JournalEntryUpdate, JournalEntry, Entries
from datetime import datetime, timedelta
from typing import List, Optional
from bson import ObjectId
from bson.errors import InvalidId
from utils import format_ist_timestamp
router = APIRouter()
@@ -50,7 +51,10 @@ async def create_entry(user_id: str, entry_data: JournalEntryCreate):
try:
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
# Verify user exists
user = db.users.find_one({"_id": user_oid})
if not user:
@@ -91,9 +95,6 @@ async def create_entry(user_id: str, entry_data: JournalEntryCreate):
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(
status_code=500, detail=f"Failed to create entry: {str(e)}")
@@ -113,7 +114,10 @@ async def get_user_entries(
try:
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
# Verify user exists
user = db.users.find_one({"_id": user_oid})
if not user:
@@ -142,10 +146,9 @@ async def get_user_entries(
"hasMore": has_more
}
}
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(
status_code=500, detail=f"Failed to fetch entries: {str(e)}")
@@ -158,7 +161,10 @@ async def get_entry(user_id: str, entry_id: str):
try:
user_oid = ObjectId(user_id)
entry_oid = ObjectId(entry_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid ID format")
try:
entry = db.entries.find_one({
"_id": entry_oid,
"userId": user_oid
@@ -168,9 +174,9 @@ async def get_entry(user_id: str, entry_id: str):
raise HTTPException(status_code=404, detail="Entry not found")
return _format_entry(entry)
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(status_code=400, detail="Invalid ID format")
raise HTTPException(
status_code=500, detail=f"Failed to fetch entry: {str(e)}")
@@ -183,7 +189,10 @@ async def update_entry(user_id: str, entry_id: str, entry_data: JournalEntryUpda
try:
user_oid = ObjectId(user_id)
entry_oid = ObjectId(entry_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid ID format")
try:
update_data = entry_data.model_dump(exclude_unset=True)
update_data["updatedAt"] = datetime.utcnow()
@@ -206,9 +215,9 @@ async def update_entry(user_id: str, entry_id: str, entry_data: JournalEntryUpda
# Fetch and return updated entry
entry = db.entries.find_one({"_id": entry_oid})
return _format_entry(entry)
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(status_code=400, detail="Invalid ID format")
raise HTTPException(
status_code=500, detail=f"Failed to update entry: {str(e)}")
@@ -221,7 +230,10 @@ async def delete_entry(user_id: str, entry_id: str):
try:
user_oid = ObjectId(user_id)
entry_oid = ObjectId(entry_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid ID format")
try:
result = db.entries.delete_one({
"_id": entry_oid,
"userId": user_oid
@@ -231,9 +243,9 @@ async def delete_entry(user_id: str, entry_id: str):
raise HTTPException(status_code=404, detail="Entry not found")
return {"message": "Entry deleted successfully"}
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(status_code=400, detail="Invalid ID format")
raise HTTPException(
status_code=500, detail=f"Failed to delete entry: {str(e)}")
@@ -249,7 +261,10 @@ async def get_entries_by_date(user_id: str, date_str: str):
try:
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
# Parse date
target_date = datetime.strptime(date_str, "%Y-%m-%d")
next_date = target_date + timedelta(days=1)
@@ -274,10 +289,9 @@ async def get_entries_by_date(user_id: str, date_str: str):
except ValueError:
raise HTTPException(
status_code=400, detail="Invalid date format. Use YYYY-MM-DD")
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(
status_code=500, detail=f"Failed to fetch entries: {str(e)}")
@@ -293,7 +307,10 @@ async def get_entries_by_month(user_id: str, year: int, month: int, limit: int =
try:
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
if not (1 <= month <= 12):
raise HTTPException(
status_code=400, detail="Month must be between 1 and 12")
@@ -325,10 +342,9 @@ async def get_entries_by_month(user_id: str, year: int, month: int, limit: int =
}
except ValueError:
raise HTTPException(status_code=400, detail="Invalid year or month")
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(
status_code=500, detail=f"Failed to fetch entries: {str(e)}")
@@ -347,6 +363,8 @@ async def convert_utc_to_ist(data: dict):
"utc": utc_timestamp,
"ist": ist_timestamp
}
except HTTPException:
raise
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:

56
backend/routers/users.py
View File

@@ -1,11 +1,11 @@
"""User management routes"""
from fastapi import APIRouter, HTTPException
from pymongo.errors import DuplicateKeyError, WriteError
from db import get_database
from models import UserCreate, UserUpdate, User
from datetime import datetime
from typing import Optional
from bson import ObjectId
from bson.errors import InvalidId
router = APIRouter()
@@ -56,6 +56,8 @@ async def register_user(user_data: UserCreate):
"updatedAt": user["updatedAt"].isoformat(),
"message": "User registered successfully" if result.upserted_id else "User already exists"
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Registration failed: {str(e)}")
@@ -80,6 +82,8 @@ async def get_user_by_email(email: str):
"createdAt": user["createdAt"].isoformat(),
"updatedAt": user["updatedAt"].isoformat()
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to fetch user: {str(e)}")
@@ -91,7 +95,12 @@ async def get_user_by_id(user_id: str):
db = get_database()
try:
user = db.users.find_one({"_id": ObjectId(user_id)})
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
user = db.users.find_one({"_id": user_oid})
if not user:
raise HTTPException(status_code=404, detail="User not found")
@@ -104,10 +113,9 @@ async def get_user_by_id(user_id: str):
"createdAt": user["createdAt"].isoformat(),
"updatedAt": user["updatedAt"].isoformat()
}
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(
status_code=500, detail=f"Failed to fetch user: {str(e)}")
@@ -117,13 +125,18 @@ async def update_user(user_id: str, user_data: UserUpdate):
"""Update user profile."""
db = get_database()
try:
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
# Prepare update data (exclude None values)
update_data = user_data.model_dump(exclude_unset=True)
update_data["updatedAt"] = datetime.utcnow()
result = db.users.update_one(
{"_id": ObjectId(user_id)},
{"_id": user_oid},
{"$set": update_data}
)
@@ -131,7 +144,7 @@ async def update_user(user_id: str, user_data: UserUpdate):
raise HTTPException(status_code=404, detail="User not found")
# Fetch and return updated user
user = db.users.find_one({"_id": ObjectId(user_id)})
user = db.users.find_one({"_id": user_oid})
return {
"id": str(user["_id"]),
"email": user["email"],
@@ -142,10 +155,9 @@ async def update_user(user_id: str, user_data: UserUpdate):
"updatedAt": user["updatedAt"].isoformat(),
"message": "User updated successfully"
}
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(status_code=500, detail=f"Update failed: {str(e)}")
@@ -154,33 +166,27 @@ async def delete_user(user_id: str):
"""Delete user account and all associated data."""
db = get_database()
try:
user_oid = ObjectId(user_id)
except InvalidId:
raise HTTPException(status_code=400, detail="Invalid user ID format")
try:
# Delete user
user_result = db.users.delete_one({"_id": ObjectId(user_id)})
user_result = db.users.delete_one({"_id": user_oid})
if user_result.deleted_count == 0:
raise HTTPException(status_code=404, detail="User not found")
# Delete all user's entries
entry_result = db.entries.delete_many({"userId": ObjectId(user_id)})
entry_result = db.entries.delete_many({"userId": user_oid})
return {
"message": "User deleted successfully",
"user_deleted": user_result.deleted_count,
"entries_deleted": entry_result.deleted_count
}
except HTTPException:
raise
except Exception as e:
if "invalid ObjectId" in str(e).lower():
raise HTTPException(
status_code=400, detail="Invalid user ID format")
raise HTTPException(
status_code=500, detail=f"Deletion failed: {str(e)}")
# Delete all entries by user
db.entries.delete_many({"userId": user_id})
# Delete user settings
db.settings.delete_one({"userId": user_id})
return {"message": "User and associated data deleted"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

0
backend/tests/__init__.py Normal file
View File

41
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,41 @@
"""
Shared pytest fixtures for all backend tests.
Strategy:
- Use mongomock to create an in-memory MongoDB per test.
- Directly set MongoDB.db to the mock database so get_database() returns it.
- Patch MongoDB.connect_db / close_db so FastAPI's lifespan doesn't try
to connect to a real MongoDB server.
"""
import pytest
import mongomock
from unittest.mock import patch
from fastapi.testclient import TestClient
@pytest.fixture
def mock_db():
"""Fresh in-memory MongoDB database for each test."""
client = mongomock.MongoClient()
return client["test_grateful_journal"]
@pytest.fixture
def client(mock_db):
"""
FastAPI TestClient with MongoDB replaced by an in-memory mock.
Yields (TestClient, mock_db) so tests can inspect the database directly.
"""
from db import MongoDB
from main import app
with (
patch.object(MongoDB, "connect_db"),
patch.object(MongoDB, "close_db"),
):
MongoDB.db = mock_db
with TestClient(app) as c:
yield c, mock_db
MongoDB.db = None

454
backend/tests/test_entries.py Normal file
View File

@@ -0,0 +1,454 @@
"""Tests for journal entry endpoints (/api/entries/*)."""
import pytest
# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------
VALID_ENCRYPTION = {
"encrypted": True,
"ciphertext": "dGVzdF9jaXBoZXJ0ZXh0", # base64("test_ciphertext")
"nonce": "dGVzdF9ub25jZQ==", # base64("test_nonce")
"algorithm": "XSalsa20-Poly1305",
}
@pytest.fixture
def user(client):
"""Register and return a test user."""
c, _ = client
response = c.post("/api/users/register", json={"email": "entry_test@example.com"})
return response.json()
@pytest.fixture
def entry(client, user):
"""Create and return a test entry."""
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={"encryption": VALID_ENCRYPTION})
assert response.status_code == 200
return response.json()
# ---------------------------------------------------------------------------
# POST /api/entries/{user_id}
# ---------------------------------------------------------------------------
class TestCreateEntry:
def test_create_encrypted_entry_returns_200(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={"encryption": VALID_ENCRYPTION})
assert response.status_code == 200
def test_create_entry_returns_id_and_message(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={"encryption": VALID_ENCRYPTION})
data = response.json()
assert "id" in data
assert data["message"] == "Entry created successfully"
def test_create_entry_with_mood(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"mood": "grateful",
})
assert response.status_code == 200
def test_create_entry_with_invalid_mood_returns_422(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"mood": "ecstatic", # Not in MoodEnum
})
assert response.status_code == 422
def test_create_entry_with_tags(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"tags": ["family", "gratitude"],
})
assert response.status_code == 200
def test_create_entry_missing_ciphertext_returns_400(self, client, user):
"""Encryption metadata without ciphertext must be rejected."""
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={
"encryption": {
"encrypted": True,
"nonce": "bm9uY2U=",
"algorithm": "XSalsa20-Poly1305",
# ciphertext intentionally missing
}
})
# Pydantic requires ciphertext field → 422
assert response.status_code == 422
def test_create_entry_encryption_missing_nonce_returns_400(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={
"encryption": {
"encrypted": True,
"ciphertext": "dGVzdA==",
"algorithm": "XSalsa20-Poly1305",
# nonce intentionally missing
}
})
assert response.status_code == 422
def test_create_entry_for_nonexistent_user_returns_404(self, client):
c, _ = client
response = c.post("/api/entries/507f1f77bcf86cd799439011", json={"encryption": VALID_ENCRYPTION})
assert response.status_code == 404
assert "User not found" in response.json()["detail"]
def test_create_entry_with_invalid_user_id_returns_400(self, client):
c, _ = client
response = c.post("/api/entries/not-a-valid-id", json={"encryption": VALID_ENCRYPTION})
assert response.status_code == 400
def test_create_entry_with_specific_entry_date(self, client, user):
c, _ = client
response = c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"entryDate": "2024-06-15T00:00:00",
})
assert response.status_code == 200
# ---------------------------------------------------------------------------
# GET /api/entries/{user_id}
# ---------------------------------------------------------------------------
class TestGetUserEntries:
def test_returns_entries_and_pagination(self, client, user, entry):
c, _ = client
response = c.get(f"/api/entries/{user['id']}")
assert response.status_code == 200
data = response.json()
assert "entries" in data
assert "pagination" in data
def test_returns_entry_that_was_created(self, client, user, entry):
c, _ = client
response = c.get(f"/api/entries/{user['id']}")
entries = response.json()["entries"]
assert len(entries) == 1
assert entries[0]["id"] == entry["id"]
def test_entry_includes_encryption_metadata(self, client, user, entry):
c, _ = client
response = c.get(f"/api/entries/{user['id']}")
fetched_entry = response.json()["entries"][0]
assert fetched_entry["encryption"]["ciphertext"] == VALID_ENCRYPTION["ciphertext"]
assert fetched_entry["encryption"]["nonce"] == VALID_ENCRYPTION["nonce"]
def test_empty_list_when_no_entries(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}")
assert response.status_code == 200
assert response.json()["entries"] == []
assert response.json()["pagination"]["total"] == 0
def test_pagination_limit(self, client, user):
c, _ = client
for _ in range(5):
c.post(f"/api/entries/{user['id']}", json={"encryption": VALID_ENCRYPTION})
response = c.get(f"/api/entries/{user['id']}?limit=2&skip=0")
assert response.status_code == 200
data = response.json()
assert len(data["entries"]) == 2
assert data["pagination"]["hasMore"] is True
assert data["pagination"]["total"] == 5
def test_pagination_skip(self, client, user):
c, _ = client
for _ in range(4):
c.post(f"/api/entries/{user['id']}", json={"encryption": VALID_ENCRYPTION})
response = c.get(f"/api/entries/{user['id']}?limit=10&skip=3")
assert len(response.json()["entries"]) == 1
def test_pagination_has_more_false_at_end(self, client, user):
c, _ = client
for _ in range(3):
c.post(f"/api/entries/{user['id']}", json={"encryption": VALID_ENCRYPTION})
response = c.get(f"/api/entries/{user['id']}?limit=10&skip=0")
assert response.json()["pagination"]["hasMore"] is False
def test_nonexistent_user_returns_404(self, client):
c, _ = client
response = c.get("/api/entries/507f1f77bcf86cd799439011")
assert response.status_code == 404
def test_invalid_user_id_returns_400(self, client):
c, _ = client
response = c.get("/api/entries/bad-id")
assert response.status_code == 400
# ---------------------------------------------------------------------------
# GET /api/entries/{user_id}/{entry_id}
# ---------------------------------------------------------------------------
class TestGetSingleEntry:
def test_returns_entry_by_id(self, client, user, entry):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/{entry['id']}")
assert response.status_code == 200
assert response.json()["id"] == entry["id"]
def test_returned_entry_has_encryption_field(self, client, user, entry):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/{entry['id']}")
data = response.json()
assert "encryption" in data
assert data["encryption"]["ciphertext"] == VALID_ENCRYPTION["ciphertext"]
def test_entry_belongs_to_correct_user(self, client, user, entry):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/{entry['id']}")
assert response.json()["userId"] == user["id"]
def test_entry_from_different_user_returns_404(self, client, user, entry):
"""User isolation: another user cannot access this entry."""
c, _ = client
other = c.post("/api/users/register", json={"email": "other@example.com"}).json()
response = c.get(f"/api/entries/{other['id']}/{entry['id']}")
assert response.status_code == 404
def test_nonexistent_entry_returns_404(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/507f1f77bcf86cd799439099")
assert response.status_code == 404
def test_invalid_entry_id_returns_400(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/not-valid-id")
assert response.status_code == 400
def test_invalid_user_id_returns_400(self, client, entry):
c, _ = client
response = c.get(f"/api/entries/bad-user-id/{entry['id']}")
assert response.status_code == 400
# ---------------------------------------------------------------------------
# PUT /api/entries/{user_id}/{entry_id}
# ---------------------------------------------------------------------------
class TestUpdateEntry:
def test_update_mood(self, client, user, entry):
c, _ = client
response = c.put(f"/api/entries/{user['id']}/{entry['id']}", json={"mood": "happy"})
assert response.status_code == 200
assert response.json()["mood"] == "happy"
def test_update_encryption_ciphertext(self, client, user, entry):
c, _ = client
new_enc = {**VALID_ENCRYPTION, "ciphertext": "bmV3Y2lwaGVydGV4dA=="}
response = c.put(f"/api/entries/{user['id']}/{entry['id']}", json={"encryption": new_enc})
assert response.status_code == 200
assert response.json()["encryption"]["ciphertext"] == "bmV3Y2lwaGVydGV4dA=="
def test_update_persists(self, client, user, entry):
c, _ = client
c.put(f"/api/entries/{user['id']}/{entry['id']}", json={"mood": "sad"})
response = c.get(f"/api/entries/{user['id']}/{entry['id']}")
assert response.json()["mood"] == "sad"
def test_update_invalid_mood_returns_422(self, client, user, entry):
c, _ = client
response = c.put(f"/api/entries/{user['id']}/{entry['id']}", json={"mood": "furious"})
assert response.status_code == 422
def test_update_nonexistent_entry_returns_404(self, client, user):
c, _ = client
response = c.put(f"/api/entries/{user['id']}/507f1f77bcf86cd799439099", json={"mood": "happy"})
assert response.status_code == 404
def test_update_invalid_entry_id_returns_400(self, client, user):
c, _ = client
response = c.put(f"/api/entries/{user['id']}/bad-id", json={"mood": "happy"})
assert response.status_code == 400
# ---------------------------------------------------------------------------
# DELETE /api/entries/{user_id}/{entry_id}
# ---------------------------------------------------------------------------
class TestDeleteEntry:
def test_delete_entry_returns_200(self, client, user, entry):
c, _ = client
response = c.delete(f"/api/entries/{user['id']}/{entry['id']}")
assert response.status_code == 200
assert "deleted" in response.json()["message"].lower()
def test_deleted_entry_is_not_retrievable(self, client, user, entry):
c, _ = client
c.delete(f"/api/entries/{user['id']}/{entry['id']}")
response = c.get(f"/api/entries/{user['id']}/{entry['id']}")
assert response.status_code == 404
def test_deleted_entry_not_in_list(self, client, user, entry):
c, _ = client
c.delete(f"/api/entries/{user['id']}/{entry['id']}")
response = c.get(f"/api/entries/{user['id']}")
assert response.json()["entries"] == []
def test_delete_entry_wrong_user_returns_404(self, client, user, entry):
"""User isolation: another user cannot delete this entry."""
c, _ = client
other = c.post("/api/users/register", json={"email": "other_del@example.com"}).json()
response = c.delete(f"/api/entries/{other['id']}/{entry['id']}")
assert response.status_code == 404
def test_delete_nonexistent_entry_returns_404(self, client, user):
c, _ = client
response = c.delete(f"/api/entries/{user['id']}/507f1f77bcf86cd799439099")
assert response.status_code == 404
def test_delete_invalid_entry_id_returns_400(self, client, user):
c, _ = client
response = c.delete(f"/api/entries/{user['id']}/bad-id")
assert response.status_code == 400
# ---------------------------------------------------------------------------
# GET /api/entries/{user_id}/by-date/{date_str}
# ---------------------------------------------------------------------------
class TestGetEntriesByDate:
def test_returns_entry_for_matching_date(self, client, user):
c, _ = client
c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"entryDate": "2024-06-15T00:00:00",
})
response = c.get(f"/api/entries/{user['id']}/by-date/2024-06-15")
assert response.status_code == 200
data = response.json()
assert data["count"] == 1
assert data["date"] == "2024-06-15"
def test_returns_empty_for_date_with_no_entries(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/by-date/2020-01-01")
assert response.status_code == 200
assert response.json()["count"] == 0
def test_does_not_return_entries_from_other_dates(self, client, user):
c, _ = client
c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"entryDate": "2024-06-15T00:00:00",
})
response = c.get(f"/api/entries/{user['id']}/by-date/2024-06-16") # Next day
assert response.json()["count"] == 0
def test_invalid_date_format_returns_400(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/by-date/not-a-date")
assert response.status_code == 400
def test_invalid_date_13th_month_returns_400(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/by-date/2024-13-01")
assert response.status_code == 400
def test_invalid_user_id_returns_400(self, client):
c, _ = client
response = c.get("/api/entries/bad-id/by-date/2024-06-15")
assert response.status_code == 400
# ---------------------------------------------------------------------------
# GET /api/entries/{user_id}/by-month/{year}/{month}
# ---------------------------------------------------------------------------
class TestGetEntriesByMonth:
def test_returns_entries_for_matching_month(self, client, user):
c, _ = client
c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"entryDate": "2024-06-15T00:00:00",
})
response = c.get(f"/api/entries/{user['id']}/by-month/2024/6")
assert response.status_code == 200
data = response.json()
assert data["count"] == 1
assert data["year"] == 2024
assert data["month"] == 6
def test_does_not_return_entries_from_other_months(self, client, user):
c, _ = client
c.post(f"/api/entries/{user['id']}", json={
"encryption": VALID_ENCRYPTION,
"entryDate": "2024-05-10T00:00:00", # May, not June
})
response = c.get(f"/api/entries/{user['id']}/by-month/2024/6")
assert response.json()["count"] == 0
def test_december_january_rollover_works(self, client, user):
"""Month 12 boundary (year+1 rollover) must not crash."""
c, _ = client
response = c.get(f"/api/entries/{user['id']}/by-month/2024/12")
assert response.status_code == 200
def test_invalid_month_0_returns_400(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/by-month/2024/0")
assert response.status_code == 400
def test_invalid_month_13_returns_400(self, client, user):
c, _ = client
response = c.get(f"/api/entries/{user['id']}/by-month/2024/13")
assert response.status_code == 400
def test_invalid_user_id_returns_400(self, client):
c, _ = client
response = c.get("/api/entries/bad-id/by-month/2024/6")
assert response.status_code == 400
# ---------------------------------------------------------------------------
# POST /api/entries/convert-timestamp/utc-to-ist
# ---------------------------------------------------------------------------
class TestConvertTimestamp:
def test_converts_utc_z_suffix_to_ist(self, client):
c, _ = client
response = c.post("/api/entries/convert-timestamp/utc-to-ist", json={
"timestamp": "2024-01-01T00:00:00Z"
})
assert response.status_code == 200
data = response.json()
assert "utc" in data
assert "ist" in data
assert "+05:30" in data["ist"]
def test_ist_is_5h30m_ahead_of_utc(self, client):
c, _ = client
response = c.post("/api/entries/convert-timestamp/utc-to-ist", json={
"timestamp": "2024-01-01T00:00:00Z"
})
assert "05:30:00+05:30" in response.json()["ist"]
def test_missing_timestamp_field_returns_400(self, client):
c, _ = client
response = c.post("/api/entries/convert-timestamp/utc-to-ist", json={})
assert response.status_code == 400
def test_invalid_timestamp_string_returns_400(self, client):
c, _ = client
response = c.post("/api/entries/convert-timestamp/utc-to-ist", json={
"timestamp": "not-a-date"
})
assert response.status_code == 400
def test_returns_original_utc_in_response(self, client):
c, _ = client
utc = "2024-06-15T12:00:00Z"
response = c.post("/api/entries/convert-timestamp/utc-to-ist", json={"timestamp": utc})
assert response.json()["utc"] == utc

196
backend/tests/test_models.py Normal file
View File

@@ -0,0 +1,196 @@
"""Tests for Pydantic data models (backend/models.py)."""
import pytest
from pydantic import ValidationError
from models import (
UserCreate,
UserUpdate,
EncryptionMetadata,
JournalEntryCreate,
JournalEntryUpdate,
MoodEnum,
)
# ---------------------------------------------------------------------------
# UserCreate
# ---------------------------------------------------------------------------
class TestUserCreate:
def test_requires_email(self):
with pytest.raises(ValidationError):
UserCreate()
def test_valid_email_only(self):
user = UserCreate(email="test@example.com")
assert user.email == "test@example.com"
def test_display_name_is_optional(self):
user = UserCreate(email="test@example.com")
assert user.displayName is None
def test_photo_url_is_optional(self):
user = UserCreate(email="test@example.com")
assert user.photoURL is None
def test_all_fields(self):
user = UserCreate(
email="test@example.com",
displayName="Alice",
photoURL="https://example.com/pic.jpg",
)
assert user.displayName == "Alice"
assert user.photoURL == "https://example.com/pic.jpg"
# ---------------------------------------------------------------------------
# UserUpdate
# ---------------------------------------------------------------------------
class TestUserUpdate:
def test_all_fields_optional(self):
update = UserUpdate()
assert update.displayName is None
assert update.photoURL is None
assert update.theme is None
def test_update_only_theme(self):
update = UserUpdate(theme="dark")
assert update.theme == "dark"
assert update.displayName is None
def test_update_only_display_name(self):
update = UserUpdate(displayName="New Name")
assert update.displayName == "New Name"
assert update.theme is None
def test_model_dump_excludes_unset(self):
update = UserUpdate(theme="dark")
dumped = update.model_dump(exclude_unset=True)
assert "theme" in dumped
assert "displayName" not in dumped
# ---------------------------------------------------------------------------
# EncryptionMetadata
# ---------------------------------------------------------------------------
class TestEncryptionMetadata:
def test_requires_ciphertext(self):
with pytest.raises(ValidationError):
EncryptionMetadata(nonce="abc")
def test_requires_nonce(self):
with pytest.raises(ValidationError):
EncryptionMetadata(ciphertext="abc")
def test_requires_both_ciphertext_and_nonce(self):
with pytest.raises(ValidationError):
EncryptionMetadata()
def test_default_algorithm_is_xsalsa20(self):
meta = EncryptionMetadata(ciphertext="abc", nonce="xyz")
assert meta.algorithm == "XSalsa20-Poly1305"
def test_default_encrypted_is_true(self):
meta = EncryptionMetadata(ciphertext="abc", nonce="xyz")
assert meta.encrypted is True
def test_valid_full_metadata(self):
meta = EncryptionMetadata(
encrypted=True,
ciphertext="dGVzdA==",
nonce="bm9uY2U=",
algorithm="XSalsa20-Poly1305",
)
assert meta.ciphertext == "dGVzdA=="
assert meta.nonce == "bm9uY2U="
def test_custom_algorithm_accepted(self):
meta = EncryptionMetadata(ciphertext="abc", nonce="xyz", algorithm="AES-256-GCM")
assert meta.algorithm == "AES-256-GCM"
# ---------------------------------------------------------------------------
# JournalEntryCreate
# ---------------------------------------------------------------------------
class TestJournalEntryCreate:
def test_all_fields_optional(self):
entry = JournalEntryCreate()
assert entry.title is None
assert entry.content is None
assert entry.encryption is None
assert entry.mood is None
def test_encrypted_entry_has_no_plaintext(self):
"""Encrypted entries legitimately have no title or content."""
entry = JournalEntryCreate(
encryption=EncryptionMetadata(ciphertext="abc", nonce="xyz")
)
assert entry.title is None
assert entry.content is None
assert entry.encryption is not None
def test_valid_mood_values(self):
for mood in ("happy", "sad", "neutral", "anxious", "grateful"):
entry = JournalEntryCreate(mood=mood)
assert entry.mood == mood
def test_invalid_mood_raises_validation_error(self):
with pytest.raises(ValidationError):
JournalEntryCreate(mood="ecstatic")
def test_default_is_public_is_false(self):
entry = JournalEntryCreate()
assert entry.isPublic is False
def test_tags_default_is_none(self):
entry = JournalEntryCreate()
assert entry.tags is None
def test_tags_list_accepted(self):
entry = JournalEntryCreate(tags=["family", "work", "health"])
assert entry.tags == ["family", "work", "health"]
# ---------------------------------------------------------------------------
# JournalEntryUpdate
# ---------------------------------------------------------------------------
class TestJournalEntryUpdate:
def test_all_fields_optional(self):
update = JournalEntryUpdate()
assert update.title is None
assert update.mood is None
def test_update_mood_only(self):
update = JournalEntryUpdate(mood="happy")
dumped = update.model_dump(exclude_unset=True)
assert dumped == {"mood": MoodEnum.happy}
def test_invalid_mood_raises_error(self):
with pytest.raises(ValidationError):
JournalEntryUpdate(mood="angry")
def test_update_encryption(self):
update = JournalEntryUpdate(
encryption=EncryptionMetadata(ciphertext="new_ct", nonce="new_nonce")
)
assert update.encryption.ciphertext == "new_ct"
# ---------------------------------------------------------------------------
# MoodEnum
# ---------------------------------------------------------------------------
class TestMoodEnum:
def test_all_enum_values(self):
assert MoodEnum.happy == "happy"
assert MoodEnum.sad == "sad"
assert MoodEnum.neutral == "neutral"
assert MoodEnum.anxious == "anxious"
assert MoodEnum.grateful == "grateful"
def test_enum_used_in_entry_create(self):
entry = JournalEntryCreate(mood=MoodEnum.grateful)
assert entry.mood == "grateful"

236
backend/tests/test_users.py Normal file
View File

@@ -0,0 +1,236 @@
"""Tests for user management endpoints (/api/users/*)."""
import pytest
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def registered_user(client):
"""Register a test user and return the API response data."""
c, _ = client
response = c.post("/api/users/register", json={
"email": "test@example.com",
"displayName": "Test User",
"photoURL": "https://example.com/photo.jpg",
})
assert response.status_code == 200
return response.json()
# ---------------------------------------------------------------------------
# POST /api/users/register
# ---------------------------------------------------------------------------
class TestRegisterUser:
def test_register_new_user_returns_200(self, client):
c, _ = client
response = c.post("/api/users/register", json={"email": "new@example.com", "displayName": "New User"})
assert response.status_code == 200
def test_register_returns_user_fields(self, client):
c, _ = client
response = c.post("/api/users/register", json={"email": "new@example.com", "displayName": "New User"})
data = response.json()
assert data["email"] == "new@example.com"
assert data["displayName"] == "New User"
assert "id" in data
assert "createdAt" in data
assert "updatedAt" in data
def test_register_returns_registered_message(self, client):
c, _ = client
response = c.post("/api/users/register", json={"email": "brand_new@example.com"})
assert response.json()["message"] == "User registered successfully"
def test_register_existing_user_is_idempotent(self, client):
c, _ = client
payload = {"email": "existing@example.com"}
c.post("/api/users/register", json=payload)
response = c.post("/api/users/register", json=payload)
assert response.status_code == 200
assert response.json()["message"] == "User already exists"
def test_register_idempotent_returns_same_id(self, client):
c, _ = client
payload = {"email": "same@example.com"}
r1 = c.post("/api/users/register", json=payload).json()
r2 = c.post("/api/users/register", json=payload).json()
assert r1["id"] == r2["id"]
def test_register_uses_email_prefix_as_default_display_name(self, client):
c, _ = client
response = c.post("/api/users/register", json={"email": "johndoe@example.com"})
assert response.json()["displayName"] == "johndoe"
def test_register_default_theme_is_light(self, client):
c, _ = client
response = c.post("/api/users/register", json={"email": "x@example.com"})
assert response.json()["theme"] == "light"
def test_register_missing_email_returns_422(self, client):
c, _ = client
response = c.post("/api/users/register", json={"displayName": "No Email"})
assert response.status_code == 422
def test_register_without_optional_fields(self, client):
c, _ = client
response = c.post("/api/users/register", json={"email": "minimal@example.com"})
assert response.status_code == 200
assert response.json()["photoURL"] is None
# ---------------------------------------------------------------------------
# GET /api/users/by-email/{email}
# ---------------------------------------------------------------------------
class TestGetUserByEmail:
def test_returns_existing_user(self, client, registered_user):
c, _ = client
email = registered_user["email"]
response = c.get(f"/api/users/by-email/{email}")
assert response.status_code == 200
assert response.json()["email"] == email
def test_returns_all_user_fields(self, client, registered_user):
c, _ = client
response = c.get(f"/api/users/by-email/{registered_user['email']}")
data = response.json()
for field in ("id", "email", "displayName", "theme", "createdAt", "updatedAt"):
assert field in data
def test_nonexistent_email_returns_404(self, client):
c, _ = client
response = c.get("/api/users/by-email/ghost@example.com")
assert response.status_code == 404
assert "User not found" in response.json()["detail"]
# ---------------------------------------------------------------------------
# GET /api/users/{user_id}
# ---------------------------------------------------------------------------
class TestGetUserById:
def test_returns_existing_user(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
response = c.get(f"/api/users/{user_id}")
assert response.status_code == 200
assert response.json()["id"] == user_id
def test_invalid_object_id_format_returns_400(self, client):
c, _ = client
response = c.get("/api/users/not-a-valid-objectid")
assert response.status_code == 400
def test_nonexistent_valid_id_returns_404(self, client):
c, _ = client
response = c.get("/api/users/507f1f77bcf86cd799439011")
assert response.status_code == 404
# ---------------------------------------------------------------------------
# PUT /api/users/{user_id}
# ---------------------------------------------------------------------------
class TestUpdateUser:
def test_update_display_name(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
response = c.put(f"/api/users/{user_id}", json={"displayName": "Updated Name"})
assert response.status_code == 200
assert response.json()["displayName"] == "Updated Name"
def test_update_theme_to_dark(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
response = c.put(f"/api/users/{user_id}", json={"theme": "dark"})
assert response.status_code == 200
assert response.json()["theme"] == "dark"
def test_update_photo_url(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
new_url = "https://new-photo.example.com/pic.jpg"
response = c.put(f"/api/users/{user_id}", json={"photoURL": new_url})
assert response.status_code == 200
assert response.json()["photoURL"] == new_url
def test_update_persists_to_database(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
c.put(f"/api/users/{user_id}", json={"displayName": "Persisted Name"})
response = c.get(f"/api/users/{user_id}")
assert response.json()["displayName"] == "Persisted Name"
def test_partial_update_does_not_clear_other_fields(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
# Update only theme
c.put(f"/api/users/{user_id}", json={"theme": "dark"})
response = c.get(f"/api/users/{user_id}")
data = response.json()
assert data["theme"] == "dark"
assert data["displayName"] == "Test User" # original value preserved
def test_update_nonexistent_user_returns_404(self, client):
c, _ = client
response = c.put("/api/users/507f1f77bcf86cd799439011", json={"displayName": "X"})
assert response.status_code == 404
def test_update_invalid_id_format_returns_400(self, client):
c, _ = client
response = c.put("/api/users/bad-id", json={"displayName": "X"})
assert response.status_code == 400
# ---------------------------------------------------------------------------
# DELETE /api/users/{user_id}
# ---------------------------------------------------------------------------
class TestDeleteUser:
def test_delete_user_returns_200(self, client, registered_user):
c, _ = client
response = c.delete(f"/api/users/{registered_user['id']}")
assert response.status_code == 200
def test_delete_user_returns_deletion_counts(self, client, registered_user):
c, _ = client
response = c.delete(f"/api/users/{registered_user['id']}")
data = response.json()
assert data["user_deleted"] == 1
assert "entries_deleted" in data
def test_delete_user_makes_them_unretrievable(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
c.delete(f"/api/users/{user_id}")
response = c.get(f"/api/users/{user_id}")
assert response.status_code == 404
def test_delete_user_also_deletes_their_entries(self, client, registered_user):
c, _ = client
user_id = registered_user["id"]
# Create 2 entries for this user
for _ in range(2):
c.post(f"/api/entries/{user_id}", json={
"encryption": {
"encrypted": True,
"ciphertext": "dGVzdA==",
"nonce": "bm9uY2U=",
"algorithm": "XSalsa20-Poly1305",
}
})
response = c.delete(f"/api/users/{user_id}")
assert response.json()["entries_deleted"] == 2
def test_delete_nonexistent_user_returns_404(self, client):
c, _ = client
response = c.delete("/api/users/507f1f77bcf86cd799439011")
assert response.status_code == 404
def test_delete_invalid_id_format_returns_400(self, client):
c, _ = client
response = c.delete("/api/users/bad-id")
assert response.status_code == 400

89
backend/tests/test_utils.py Normal file
View File

@@ -0,0 +1,89 @@
"""Tests for utility functions (backend/utils.py)."""
import pytest
from datetime import datetime, timezone, timedelta
from utils import utc_to_ist, format_ist_timestamp
IST = timezone(timedelta(hours=5, minutes=30))
class TestUtcToIst:
def test_midnight_utc_becomes_530_ist(self):
utc = datetime(2024, 1, 1, 0, 0, 0)
ist = utc_to_ist(utc)
assert ist.hour == 5
assert ist.minute == 30
def test_adds_five_hours_thirty_minutes(self):
utc = datetime(2024, 6, 15, 10, 0, 0)
ist = utc_to_ist(utc)
assert ist.hour == 15
assert ist.minute == 30
def test_rolls_over_to_next_day(self):
utc = datetime(2024, 1, 1, 22, 0, 0) # 22:00 UTC → 03:30 next day IST
ist = utc_to_ist(utc)
assert ist.day == 2
assert ist.hour == 3
assert ist.minute == 30
def test_rolls_over_to_next_month(self):
utc = datetime(2024, 1, 31, 23, 0, 0) # Jan 31 → Feb 1 IST
ist = utc_to_ist(utc)
assert ist.month == 2
assert ist.day == 1
def test_output_has_ist_timezone_offset(self):
utc = datetime(2024, 1, 1, 12, 0, 0)
ist = utc_to_ist(utc)
assert ist.utcoffset() == timedelta(hours=5, minutes=30)
def test_preserves_seconds(self):
utc = datetime(2024, 3, 15, 8, 45, 30)
ist = utc_to_ist(utc)
assert ist.second == 30
def test_noon_utc_is_1730_ist(self):
utc = datetime(2024, 7, 4, 12, 0, 0)
ist = utc_to_ist(utc)
assert ist.hour == 17
assert ist.minute == 30
class TestFormatIstTimestamp:
def test_converts_z_suffix_timestamp(self):
result = format_ist_timestamp("2024-01-01T00:00:00Z")
assert "+05:30" in result
def test_converts_explicit_utc_offset_timestamp(self):
result = format_ist_timestamp("2024-01-01T00:00:00+00:00")
assert "+05:30" in result
def test_midnight_utc_produces_0530_ist(self):
result = format_ist_timestamp("2024-01-01T00:00:00Z")
assert "05:30:00+05:30" in result
def test_noon_utc_produces_1730_ist(self):
result = format_ist_timestamp("2024-01-01T12:00:00Z")
assert "17:30:00+05:30" in result
def test_returns_iso_format_string(self):
result = format_ist_timestamp("2024-06-15T08:00:00Z")
# Should be parseable as ISO datetime
parsed = datetime.fromisoformat(result)
assert parsed is not None
def test_invalid_text_raises_value_error(self):
with pytest.raises(ValueError):
format_ist_timestamp("not-a-date")
def test_invalid_month_raises_value_error(self):
with pytest.raises(ValueError):
format_ist_timestamp("2024-13-01T00:00:00Z")
def test_empty_string_raises_value_error(self):
with pytest.raises(ValueError):
format_ist_timestamp("")
def test_slash_separated_date_raises_value_error(self):
with pytest.raises(ValueError):
format_ist_timestamp("2024/01/01T00:00:00") # Slashes not valid ISO format