phase 10
All checks were successful
CI/CD Pipeline / VM Test - backend-integration (push) Successful in 7s
CI/CD Pipeline / VM Test - full-stack (push) Successful in 7s
CI/CD Pipeline / VM Test - performance (push) Successful in 7s
CI/CD Pipeline / VM Test - security (push) Successful in 7s
CI/CD Pipeline / Backend Linting (push) Successful in 2s
CI/CD Pipeline / Frontend Linting (push) Successful in 15s
CI/CD Pipeline / Nix Flake Check (push) Successful in 41s
CI/CD Pipeline / CI Summary (push) Successful in 1s
All checks were successful
CI/CD Pipeline / VM Test - backend-integration (push) Successful in 7s
CI/CD Pipeline / VM Test - full-stack (push) Successful in 7s
CI/CD Pipeline / VM Test - performance (push) Successful in 7s
CI/CD Pipeline / VM Test - security (push) Successful in 7s
CI/CD Pipeline / Backend Linting (push) Successful in 2s
CI/CD Pipeline / Frontend Linting (push) Successful in 15s
CI/CD Pipeline / Nix Flake Check (push) Successful in 41s
CI/CD Pipeline / CI Summary (push) Successful in 1s
This commit is contained in:
377
backend/tests/api/test_bulk_operations.py
Normal file
377
backend/tests/api/test_bulk_operations.py
Normal file
@@ -0,0 +1,377 @@
|
||||
"""Integration tests for bulk image operations."""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from uuid import uuid4
|
||||
|
||||
from app.database.models.user import User
|
||||
from app.database.models.board import Board
|
||||
from app.database.models.image import Image
|
||||
from app.database.models.board_image import BoardImage
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_position_delta(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk updating positions with delta."""
|
||||
# Create board
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
# Create images
|
||||
images = []
|
||||
board_images = []
|
||||
|
||||
for i in range(3):
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename=f"test{i}.jpg",
|
||||
storage_path=f"{test_user.id}/test{i}.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": f"abc{i}"},
|
||||
)
|
||||
db.add(image)
|
||||
images.append(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100 * i, "y": 100 * i},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=i,
|
||||
)
|
||||
db.add(board_image)
|
||||
board_images.append(board_image)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Bulk update position
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [str(img.id) for img in images[:2]], # First 2 images
|
||||
"position_delta": {"dx": 50, "dy": 75},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["updated_count"] == 2
|
||||
assert data["failed_count"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_transformations(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk updating transformations."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
images = []
|
||||
for i in range(2):
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename=f"test{i}.jpg",
|
||||
storage_path=f"{test_user.id}/test{i}.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": f"abc{i}"},
|
||||
)
|
||||
db.add(image)
|
||||
images.append(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=0,
|
||||
)
|
||||
db.add(board_image)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Bulk update transformations
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [str(img.id) for img in images],
|
||||
"transformations": {
|
||||
"scale": 2.0,
|
||||
"rotation": 45,
|
||||
"opacity": 0.8,
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["updated_count"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_z_order_delta(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk updating Z-order with delta."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
images = []
|
||||
for i in range(3):
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename=f"test{i}.jpg",
|
||||
storage_path=f"{test_user.id}/test{i}.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": f"abc{i}"},
|
||||
)
|
||||
db.add(image)
|
||||
images.append(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=i,
|
||||
)
|
||||
db.add(board_image)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Bulk update Z-order
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [str(images[0].id), str(images[1].id)],
|
||||
"z_order_delta": 10,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["updated_count"] == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_mixed_operations(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk update with position, transformations, and z-order together."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
images = []
|
||||
for i in range(2):
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename=f"test{i}.jpg",
|
||||
storage_path=f"{test_user.id}/test{i}.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": f"abc{i}"},
|
||||
)
|
||||
db.add(image)
|
||||
images.append(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=0,
|
||||
)
|
||||
db.add(board_image)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# Bulk update everything
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [str(img.id) for img in images],
|
||||
"position_delta": {"dx": 50, "dy": 50},
|
||||
"transformations": {"scale": 2.0},
|
||||
"z_order_delta": 5,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["updated_count"] == 2
|
||||
assert data["failed_count"] == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_non_existent_image(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk update with some non-existent images."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename="test.jpg",
|
||||
storage_path=f"{test_user.id}/test.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": "abc"},
|
||||
)
|
||||
db.add(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=0,
|
||||
)
|
||||
db.add(board_image)
|
||||
await db.commit()
|
||||
|
||||
# Try to update with one valid and one invalid ID
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [str(image.id), str(uuid4())], # One valid, one invalid
|
||||
"transformations": {"scale": 2.0},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["updated_count"] == 1 # Only valid one updated
|
||||
assert data["failed_count"] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_unauthorized(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk update on board not owned by user."""
|
||||
# Create another user
|
||||
other_user = User(id=uuid4(), email="other@example.com", password_hash="hashed")
|
||||
db.add(other_user)
|
||||
|
||||
# Create board owned by other user
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=other_user.id,
|
||||
title="Other Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
await db.commit()
|
||||
|
||||
# Try bulk update as current user
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [str(uuid4())],
|
||||
"transformations": {"scale": 2.0},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_empty_image_list(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test bulk update with empty image list."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
await db.commit()
|
||||
|
||||
response = await client.patch(
|
||||
f"/api/images/boards/{board.id}/images/bulk",
|
||||
json={
|
||||
"image_ids": [],
|
||||
"transformations": {"scale": 2.0},
|
||||
},
|
||||
)
|
||||
|
||||
# Should succeed with 0 updated
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["updated_count"] == 0
|
||||
|
||||
220
backend/tests/api/test_image_delete.py
Normal file
220
backend/tests/api/test_image_delete.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""Integration tests for image deletion endpoints."""
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from uuid import uuid4
|
||||
|
||||
from app.database.models.user import User
|
||||
from app.database.models.board import Board
|
||||
from app.database.models.image import Image
|
||||
from app.database.models.board_image import BoardImage
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_image_from_board(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test removing image from board (not deleting)."""
|
||||
# Create board and image
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename="test.jpg",
|
||||
storage_path=f"{test_user.id}/test.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": "abc123"},
|
||||
reference_count=1,
|
||||
)
|
||||
db.add(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=0,
|
||||
)
|
||||
db.add(board_image)
|
||||
await db.commit()
|
||||
|
||||
# Remove from board
|
||||
response = await client.delete(f"/api/images/boards/{board.id}/images/{image.id}")
|
||||
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_image_not_on_board(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test removing image that's not on the board."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename="test.jpg",
|
||||
storage_path=f"{test_user.id}/test.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": "abc123"},
|
||||
)
|
||||
db.add(image)
|
||||
await db.commit()
|
||||
|
||||
# Try to remove (image not on board)
|
||||
response = await client.delete(f"/api/images/boards/{board.id}/images/{image.id}")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_remove_image_unauthorized(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test removing image from board not owned by user."""
|
||||
# Create another user
|
||||
other_user = User(id=uuid4(), email="other@example.com", password_hash="hashed")
|
||||
db.add(other_user)
|
||||
|
||||
# Create board owned by other user
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=other_user.id,
|
||||
title="Other Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=other_user.id,
|
||||
filename="test.jpg",
|
||||
storage_path=f"{other_user.id}/test.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": "abc123"},
|
||||
)
|
||||
db.add(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=0,
|
||||
)
|
||||
db.add(board_image)
|
||||
await db.commit()
|
||||
|
||||
# Try to remove as current user
|
||||
response = await client.delete(f"/api/images/boards/{board.id}/images/{image.id}")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_permanent_delete_image(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test permanently deleting image from library."""
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename="test.jpg",
|
||||
storage_path=f"{test_user.id}/test.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": "abc123"},
|
||||
reference_count=0, # Not used on any boards
|
||||
)
|
||||
db.add(image)
|
||||
await db.commit()
|
||||
|
||||
# Delete permanently
|
||||
response = await client.delete(f"/api/images/{image.id}")
|
||||
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cannot_delete_image_in_use(client: AsyncClient, test_user: User, db: AsyncSession):
|
||||
"""Test that images in use cannot be permanently deleted."""
|
||||
board = Board(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
title="Test Board",
|
||||
viewport_state={"x": 0, "y": 0, "zoom": 1.0, "rotation": 0},
|
||||
)
|
||||
db.add(board)
|
||||
|
||||
image = Image(
|
||||
id=uuid4(),
|
||||
user_id=test_user.id,
|
||||
filename="test.jpg",
|
||||
storage_path=f"{test_user.id}/test.jpg",
|
||||
file_size=1024,
|
||||
mime_type="image/jpeg",
|
||||
width=800,
|
||||
height=600,
|
||||
metadata={"format": "jpeg", "checksum": "abc123"},
|
||||
reference_count=1, # Used on a board
|
||||
)
|
||||
db.add(image)
|
||||
|
||||
board_image = BoardImage(
|
||||
id=uuid4(),
|
||||
board_id=board.id,
|
||||
image_id=image.id,
|
||||
position={"x": 100, "y": 100},
|
||||
transformations={
|
||||
"scale": 1.0,
|
||||
"rotation": 0,
|
||||
"opacity": 1.0,
|
||||
"flipped_h": False,
|
||||
"flipped_v": False,
|
||||
"greyscale": False,
|
||||
},
|
||||
z_order=0,
|
||||
)
|
||||
db.add(board_image)
|
||||
await db.commit()
|
||||
|
||||
# Try to delete
|
||||
response = await client.delete(f"/api/images/{image.id}")
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "still used" in response.json()["detail"].lower()
|
||||
|
||||
Reference in New Issue
Block a user