Skip to content
Merged
30 changes: 13 additions & 17 deletions backend/app/database/albums.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,16 +176,24 @@ def db_delete_album(album_id: str):
cursor.execute("DELETE FROM albums WHERE album_id = ?", (album_id,))


def db_update_album_cover_image(album_id: str, cover_image_path: str):
"""Update the cover image path for an album"""
def db_get_album_cover_path(album_id: str) -> str | None:
"""Path of the album's cover: its first image, by insertion order."""
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
try:
cursor.execute(
"UPDATE albums SET cover_image_path = ? WHERE album_id = ?",
(cover_image_path, album_id),
"""
SELECT images.path
FROM album_images
JOIN images ON images.id = album_images.image_id
WHERE album_images.album_id = ?
ORDER BY album_images.rowid
LIMIT 1
""",
(album_id,),
)
conn.commit()
result = cursor.fetchone()
return result[0] if result else None
finally:
conn.close()

Expand Down Expand Up @@ -287,15 +295,3 @@ def verify_album_password(album_id: str, password: str) -> bool:
return bcrypt.checkpw(password.encode("utf-8"), row[0].encode("utf-8"))
finally:
conn.close()


def db_get_image_path(image_id: str) -> str | None:
"""Get the path of an image by its ID."""
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
try:
cursor.execute("SELECT path FROM images WHERE id = ?", (image_id,))
result = cursor.fetchone()
return result[0] if result else None
finally:
conn.close()
76 changes: 12 additions & 64 deletions backend/app/routes/albums.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
SuccessResponse,
ErrorResponse,
ImageIdsRequest,
SetCoverImageRequest,
Album,
)
from app.database.albums import (
Expand All @@ -25,9 +24,8 @@
db_add_images_to_album,
db_remove_image_from_album,
db_remove_images_from_album,
db_update_album_cover_image,
db_get_album_cover_path,
verify_album_password,
db_get_image_path,
)

router = APIRouter()
Expand All @@ -43,14 +41,19 @@ def get_albums():
# Get image count for each album
image_ids = db_get_album_images(album[0])
image_count = len(image_ids)
is_locked = bool(album[3])

album_list.append(
Album(
album_id=album[0],
album_name=album[1],
description=album[2] or "",
is_locked=bool(album[3]),
cover_image_path=album[5] if len(album) > 5 else None,
is_locked=is_locked,
# A locked album's cover would show the very content the
# password is protecting, so never send it.
cover_image_path=(
None if is_locked else db_get_album_cover_path(album[0])
),
image_count=image_count,
)
)
Expand Down Expand Up @@ -105,12 +108,14 @@ def get_album(album_id: str = Path(...)):
image_ids = db_get_album_images(album_id)
image_count = len(image_ids)

is_locked = bool(album[3])
album_obj = Album(
album_id=album[0],
album_name=album[1],
description=album[2] or "",
is_locked=bool(album[3]),
cover_image_path=album[5] if len(album) > 5 else None,
is_locked=is_locked,
# Same reasoning as the listing: the cover gives away the contents.
cover_image_path=(None if is_locked else db_get_album_cover_path(album_id)),
image_count=image_count,
)
return GetAlbumResponse(success=True, data=album_obj)
Expand Down Expand Up @@ -371,60 +376,3 @@ def remove_images_from_album(
success=False, error="Failed to Remove Images", message=str(e)
).model_dump(),
)


# PUT /albums/{album_id}/cover - Set album cover image
@router.put("/{album_id}/cover", response_model=SuccessResponse)
def set_album_cover_image(
album_id: str = Path(...), body: SetCoverImageRequest = Body(...)
):
"""Set or update the cover image for an album"""
album = db_get_album(album_id)
if not album:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=ErrorResponse(
success=False,
error="Album Not Found",
message="No album exists with the provided ID.",
).model_dump(),
)

# Verify the image exists in the album
album_image_ids = db_get_album_images(album_id)
if body.image_id not in album_image_ids:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=ErrorResponse(
success=False,
error="Image Not In Album",
message="The specified image is not in this album.",
).model_dump(),
)

try:
# Get the image path from the database
image_path = db_get_image_path(body.image_id)

if not image_path:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=ErrorResponse(
success=False,
error="Image Not Found",
message="The specified image does not exist.",
).model_dump(),
)

db_update_album_cover_image(album_id, image_path)

return SuccessResponse(
success=True, msg="Album cover image updated successfully"
)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=ErrorResponse(
success=False, error="Failed to Set Cover Image", message=str(e)
).model_dump(),
)
4 changes: 0 additions & 4 deletions backend/app/schemas/album.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,10 +70,6 @@ def validate_image_ids(cls, value: List[str]) -> List[str]:
return cleaned


class SetCoverImageRequest(BaseModel):
image_id: str = Field(..., min_length=1)


# ##############################
# Response Handler
# ##############################
Expand Down
57 changes: 57 additions & 0 deletions backend/tests/test_albums.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,63 @@ def test_get_all_albums_include_hidden(self, mock_db_album, mock_db_locked_album

mock_get_all.assert_called_once()

def test_locked_album_cover_is_withheld(self, mock_db_album, mock_db_locked_album):
"""A locked album's cover would reveal the content the password gates."""
with patch("app.routes.albums.db_get_all_albums") as mock_get_all, patch(
"app.routes.albums.db_get_album_cover_path"
) as mock_cover:
mock_get_all.return_value = [
(
mock_db_album["album_id"],
mock_db_album["album_name"],
mock_db_album["description"],
mock_db_album["is_locked"],
mock_db_album["password_hash"],
None,
),
(
mock_db_locked_album["album_id"],
mock_db_locked_album["album_name"],
mock_db_locked_album["description"],
mock_db_locked_album["is_locked"],
mock_db_locked_album["password_hash"],
None,
),
]
mock_cover.return_value = "/photos/secret.jpg"

response = client.get("/albums/")
assert response.status_code == 200

covers = {
album["album_id"]: album["cover_image_path"]
for album in response.json()["albums"]
}
assert covers[mock_db_album["album_id"]] == "/photos/secret.jpg"
assert covers[mock_db_locked_album["album_id"]] is None
# The path is never even looked up for a locked album
mock_cover.assert_called_once_with(mock_db_album["album_id"])

def test_get_album_by_id_withholds_a_locked_cover(self, mock_db_locked_album):
"""The single-album read must not leak what the listing hides."""
with patch("app.routes.albums.db_get_album") as mock_get_album, patch(
"app.routes.albums.db_get_album_cover_path"
) as mock_cover:
mock_get_album.return_value = (
mock_db_locked_album["album_id"],
mock_db_locked_album["album_name"],
mock_db_locked_album["description"],
mock_db_locked_album["is_locked"],
mock_db_locked_album["password_hash"],
None,
)
mock_cover.return_value = "/photos/secret.jpg"

response = client.get(f"/albums/{mock_db_locked_album['album_id']}")
assert response.status_code == 200
assert response.json()["data"]["cover_image_path"] is None
mock_cover.assert_not_called()

def test_get_all_albums_empty_list(self):
"""
Test fetching albums when none exist.
Expand Down
52 changes: 52 additions & 0 deletions backend/tests/test_albums_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
db_delete_album,
db_get_album_images,
db_remove_images_from_album,
db_get_album_cover_path,
verify_album_password,
)
from app.database.images import db_create_images_table
Expand Down Expand Up @@ -73,6 +74,17 @@ def link_images(db_path: str, album_id: str, image_ids: List[str]) -> None:
conn.close()


def make_images(db_path: str, image_ids: List[str]) -> None:
"""Seed real image rows so cover lookups have something to join against."""
conn = sqlite3.connect(db_path)
conn.executemany(
"INSERT INTO images (id, path) VALUES (?, ?)",
[(image_id, f"/photos/{image_id}.jpg") for image_id in image_ids],
)
conn.commit()
conn.close()


def stored_hash(db_path: str, album_id: str) -> Optional[str]:
"""Read an album's raw password_hash straight from the table."""
conn = sqlite3.connect(db_path)
Expand Down Expand Up @@ -275,3 +287,43 @@ def test_update_without_password_keeps_the_existing_one(self, test_db):

assert stored_hash(test_db, "album-1") == before
assert verify_album_password("album-1", "oldpass") is True


# ##############################
# Cover image
# ##############################


class TestAlbumCoverPath:
def test_uses_the_first_image_added(self, test_db):
make_album("album-1")
make_images(test_db, ["img-a", "img-b", "img-c"])
link_images(test_db, "album-1", ["img-b", "img-a", "img-c"])

assert db_get_album_cover_path("album-1") == "/photos/img-b.jpg"

def test_returns_none_for_an_empty_album(self, test_db):
make_album("album-1")

assert db_get_album_cover_path("album-1") is None

def test_returns_none_for_a_missing_album(self, test_db):
assert db_get_album_cover_path("nope") is None

def test_follows_the_album_when_the_first_image_is_removed(self, test_db):
make_album("album-1")
make_images(test_db, ["img-a", "img-b"])
link_images(test_db, "album-1", ["img-a", "img-b"])

db_remove_images_from_album("album-1", ["img-a"])

assert db_get_album_cover_path("album-1") == "/photos/img-b.jpg"

def test_ignores_images_in_other_albums(self, test_db):
make_album("album-1")
make_album("album-2", name="Other")
make_images(test_db, ["img-a", "img-b"])
link_images(test_db, "album-1", ["img-a"])
link_images(test_db, "album-2", ["img-b"])

assert db_get_album_cover_path("album-2") == "/photos/img-b.jpg"
67 changes: 0 additions & 67 deletions docs/backend/backend_python/openapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -804,59 +804,6 @@
}
}
},
"/albums/{album_id}/cover": {
"put": {
"tags": [
"Albums"
],
"summary": "Set Album Cover Image",
"description": "Set or update the cover image for an album",
"operationId": "set_album_cover_image_albums__album_id__cover_put",
"parameters": [
{
"name": "album_id",
"in": "path",
"required": true,
"schema": {
"type": "string",
"title": "Album Id"
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SetCoverImageRequest"
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SuccessResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/images/": {
"get": {
"tags": [
Expand Down Expand Up @@ -5176,20 +5123,6 @@
],
"title": "SemanticSearchImage"
},
"SetCoverImageRequest": {
"properties": {
"image_id": {
"type": "string",
"minLength": 1,
"title": "Image Id"
}
},
"type": "object",
"required": [
"image_id"
],
"title": "SetCoverImageRequest"
},
"SetupRequest": {
"properties": {
"tier": {
Expand Down
Loading
Loading