Mirrors/mealie
1
0
Fork 0
mirror of https://github.com/mealie-recipes/mealie.git synced 2026-05-26 03:30:26 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'mealie-next' into fix/plan-to-eat-import-6360

Browse Source
This commit is contained in:
Hayden
2026-05-24 11:59:39 -05:00
committed by GitHub
parent 23cb4bdf36 5ce448af7a
commit 91bc1bd263
181 changed files with 4217 additions and 756 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
tests/conftest.py
View File

@@ -20,7 +20,6 @@ mp = MonkeyPatch()
mp.setenv("PRODUCTION", "True")
mp.setenv("TESTING", "True")
mp.setenv("ALLOW_SIGNUP", "True")
mp.setenv("OPENAI_API_KEY", "dummy-api-key")
from pathlib import Path
from fastapi.testclient import TestClient

136
tests/integration_tests/admin_tests/test_admin_ai_providers.py Normal file
View File

@@ -0,0 +1,136 @@
"""
Integration tests for admin AI provider management across groups.
"""
from fastapi.testclient import TestClient
from mealie.schema.group.ai_providers import AIProviderCreate
from tests.utils import api_routes
from tests.utils.factories import random_string
from tests.utils.fixture_schemas import TestUser
# =======================================================================
# Permissions
def test_admin_ai_provider_routes_require_admin(api_client: TestClient, unique_user: TestUser, admin_user: TestUser):
"""Non-admin users cannot access admin AI provider routes."""
group_id = unique_user.group_id
response = api_client.post(
api_routes.admin_groups_group_id_ai_providers_providers(group_id),
json={"name": random_string(), "model": "gpt-4o", "apiKey": "key"},
headers=unique_user.token,
)
assert response.status_code == 403
# =======================================================================
# Provider CRUD
def test_admin_create_ai_provider_for_group(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Admin can create an AI provider for any group."""
provider_name = random_string()
response = api_client.post(
api_routes.admin_groups_group_id_ai_providers_providers(unique_user.group_id),
json={"name": provider_name, "model": "gpt-4o", "apiKey": "admin-created-key"},
headers=admin_user.token,
)
assert response.status_code == 200
provider = response.json()
try:
assert provider["name"] == provider_name
assert provider["model"] == "gpt-4o"
assert "id" in provider
assert "apiKey" not in provider
finally:
unique_user.repos.group_ai_providers.delete(provider["id"])
def test_admin_get_ai_provider_for_group(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Admin can retrieve an AI provider from any group."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="secret")
)
try:
response = api_client.get(
api_routes.admin_groups_group_id_ai_providers_providers_provider_id(unique_user.group_id, provider.id),
headers=admin_user.token,
)
assert response.status_code == 200
data = response.json()
assert data["id"] == str(provider.id)
assert data["name"] == provider.name
finally:
unique_user.repos.group_ai_providers.delete(provider.id)
def test_admin_update_ai_provider_for_group(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Admin can update an AI provider in any group."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="original-key")
)
try:
new_name = random_string()
response = api_client.put(
api_routes.admin_groups_group_id_ai_providers_providers_provider_id(unique_user.group_id, provider.id),
json={"name": new_name, "model": "gpt-4-turbo", "apiKey": "updated-key"},
headers=admin_user.token,
)
assert response.status_code == 200
data = response.json()
assert data["name"] == new_name
assert data["model"] == "gpt-4-turbo"
finally:
unique_user.repos.group_ai_providers.delete(provider.id)
def test_admin_delete_ai_provider_for_group(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Admin can delete an AI provider from any group."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="to-delete")
)
response = api_client.delete(
api_routes.admin_groups_group_id_ai_providers_providers_provider_id(unique_user.group_id, provider.id),
headers=admin_user.token,
)
assert response.status_code == 200
# Confirm gone
response = api_client.get(
api_routes.admin_groups_group_id_ai_providers_providers_provider_id(unique_user.group_id, provider.id),
headers=admin_user.token,
)
assert response.status_code == 404
def test_admin_can_manage_providers_across_groups(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Admin can create providers in a group they are not a member of."""
# Create a brand new group the admin doesn't belong to
group_name = random_string()
create_resp = api_client.post(api_routes.admin_groups, json={"name": group_name}, headers=admin_user.token)
assert create_resp.status_code == 201
foreign_group_id = create_resp.json()["id"]
try:
response = api_client.post(
api_routes.admin_groups_group_id_ai_providers_providers(foreign_group_id),
json={"name": random_string(), "model": "gpt-4o", "apiKey": "cross-group-key"},
headers=admin_user.token,
)
assert response.status_code == 200
provider_id = response.json()["id"]
# Settings should also be accessible
# Cleanup provider before deleting group
api_client.delete(
api_routes.admin_groups_group_id_ai_providers_providers_provider_id(foreign_group_id, provider_id),
headers=admin_user.token,
)
finally:
api_client.delete(api_routes.admin_groups_item_id(foreign_group_id), headers=admin_user.token)

95
tests/integration_tests/admin_tests/test_admin_group_actions.py
View File

@@ -2,6 +2,7 @@ from fastapi.testclient import TestClient
from mealie.core.config import get_app_settings
from mealie.repos.repository_factory import AllRepositories
from mealie.schema.group.ai_providers import AIProviderCreate, AIProviderSettingsUpdate
from mealie.schema.user.user import GroupInDB
from tests.utils import api_routes
from tests.utils.assertion_helpers import assert_ignore_keys
@@ -79,6 +80,100 @@ def test_admin_update_group(api_client: TestClient, admin_user: TestUser, unique
assert_ignore_keys(as_json["preferences"], update_payload["preferences"]) # type: ignore
def test_admin_update_group_name_only(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Updating only the name leaves preferences and ai_provider_settings untouched."""
new_name = random_string()
response = api_client.put(
api_routes.admin_groups_item_id(unique_user.group_id),
json={"id": unique_user.group_id, "name": new_name},
headers=admin_user.token,
)
assert response.status_code == 200
assert response.json()["name"] == new_name
def test_admin_update_group_ai_provider_settings(api_client: TestClient, admin_user: TestUser, unique_user: TestUser):
"""Admin can update ai_provider_settings for a group via the PUT endpoint."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
update_payload = {
"id": unique_user.group_id,
"name": unique_user.group_id, # name is required but unchanged
"aiProviderSettings": {
"defaultProviderId": str(provider.id),
"audioProviderId": str(provider.id),
"imageProviderId": str(provider.id),
},
}
response = api_client.put(
api_routes.admin_groups_item_id(unique_user.group_id),
json=update_payload,
headers=admin_user.token,
)
assert response.status_code == 200
settings = response.json()["aiProviderSettings"]
assert settings["defaultProviderId"] == str(provider.id)
assert settings["audioProviderId"] == str(provider.id)
assert settings["imageProviderId"] == str(provider.id)
finally:
# Clear provider references before deleting the provider
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
unique_user.repos.group_ai_providers.delete(provider.id)
def test_admin_update_group_ai_provider_settings_clear(
api_client: TestClient, admin_user: TestUser, unique_user: TestUser
):
"""Admin can clear ai_provider_settings provider IDs for a group."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
# First set the providers
api_client.put(
api_routes.admin_groups_item_id(unique_user.group_id),
json={
"id": unique_user.group_id,
"name": unique_user.group_id,
"aiProviderSettings": {
"defaultProviderId": str(provider.id),
"audioProviderId": None,
"imageProviderId": None,
},
},
headers=admin_user.token,
)
# Now clear them
response = api_client.put(
api_routes.admin_groups_item_id(unique_user.group_id),
json={
"id": unique_user.group_id,
"name": unique_user.group_id,
"aiProviderSettings": {
"defaultProviderId": None,
"audioProviderId": None,
"imageProviderId": None,
},
},
headers=admin_user.token,
)
assert response.status_code == 200
settings = response.json()["aiProviderSettings"]
assert settings["defaultProviderId"] is None
assert settings["audioProviderId"] is None
assert settings["imageProviderId"] is None
finally:
unique_user.repos.group_ai_providers.delete(provider.id)
def test_admin_delete_group(unfiltered_database: AllRepositories, api_client: TestClient, admin_user: TestUser):
group = unfiltered_database.groups.create({"name": random_string()})
response = api_client.delete(api_routes.admin_groups_item_id(group.id), headers=admin_user.token)

18
tests/integration_tests/category_tag_tool_tests/test_organizers_common.py
View File

@@ -191,6 +191,24 @@ def test_organizer_association(
assert response.status_code == 200
@pytest.mark.parametrize("route", organizer_routes, ids=test_ids)
def test_organizer_create_duplicate_name_returns_400(
api_client: TestClient,
unique_user: TestUser,
route: RoutesBase,
):
# Regression test for #7582: POSTing a duplicate name to organizer endpoints
# leaked the sqlalchemy IntegrityError as an HTTP 500. The expected behavior,
# matching other organizer endpoints (foods, units, tools), is HTTP 400.
data = {"name": random_string(10)}
response = api_client.post(route.base, json=data, headers=unique_user.token)
assert response.status_code == 201
response = api_client.post(route.base, json=data, headers=unique_user.token)
assert response.status_code == 400
@pytest.mark.parametrize("route, recipe_key", association_data, ids=test_ids)
def test_organizer_get_by_slug(
api_client: TestClient,

584
tests/integration_tests/user_group_tests/test_group_ai_providers.py Normal file
View File

@@ -0,0 +1,584 @@
"""
Integration tests for AI provider CRUD, settings, permissions, and API key security.
"""
from uuid import uuid4
from fastapi.testclient import TestClient
from mealie.schema.group.ai_providers import AIProviderCreate, AIProviderSettingsUpdate
from tests.utils import api_routes
from tests.utils.factories import random_string, user_registration_factory
from tests.utils.fixture_schemas import TestUser
# ==========================================
# Provider CRUD
# ==========================================
def test_create_provider(api_client: TestClient, unique_user: TestUser):
data = {"name": random_string(), "model": "gpt-4o", "apiKey": "test-key"}
response = api_client.post(api_routes.groups_ai_providers_providers, json=data, headers=unique_user.token)
assert response.status_code == 200
provider = response.json()
assert provider["name"] == data["name"]
assert provider["model"] == data["model"]
assert "id" in provider
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider["id"]), headers=unique_user.token)
def test_get_provider(api_client: TestClient, unique_user: TestUser):
# Create a provider first
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
response = api_client.get(
api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token
)
assert response.status_code == 200
data = response.json()
assert data["id"] == str(provider.id)
assert data["name"] == provider.name
assert data["model"] == provider.model
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_get_provider_not_found(api_client: TestClient, unique_user: TestUser):
response = api_client.get(api_routes.groups_ai_providers_providers_provider_id(uuid4()), headers=unique_user.token)
assert response.status_code == 404
def test_update_provider(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="original-key")
)
try:
new_model = "gpt-4-turbo"
update_data = {"name": provider.name, "model": new_model, "apiKey": "updated-key"}
response = api_client.put(
api_routes.groups_ai_providers_providers_provider_id(provider.id),
json=update_data,
headers=unique_user.token,
)
assert response.status_code == 200
updated = response.json()
assert updated["model"] == new_model
assert updated["id"] == str(provider.id)
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_delete_provider(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
response = api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token
)
assert response.status_code == 200
# Verify it's gone
response = api_client.get(
api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token
)
assert response.status_code == 404
# ==========================================
# Provider Settings CRUD
# ==========================================
def test_get_settings(api_client: TestClient, unique_user: TestUser):
response = api_client.get(api_routes.groups_ai_providers_settings, headers=unique_user.token)
assert response.status_code == 200
settings = response.json()
assert "providers" in settings
assert "defaultProviderId" in settings
assert "audioProviderId" in settings
assert "imageProviderId" in settings
def test_update_settings_set_default_provider(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
update = {"defaultProviderId": str(provider.id), "audioProviderId": None, "imageProviderId": None}
response = api_client.put(api_routes.groups_ai_providers_settings, json=update, headers=unique_user.token)
assert response.status_code == 200
settings = response.json()
assert settings["defaultProviderId"] == str(provider.id)
finally:
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_update_settings_all_provider_types(api_client: TestClient, unique_user: TestUser):
default_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
audio_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="whisper-1", api_key="test-key")
)
image_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="dall-e-3", api_key="test-key")
)
try:
update = {
"defaultProviderId": str(default_provider.id),
"audioProviderId": str(audio_provider.id),
"imageProviderId": str(image_provider.id),
}
response = api_client.put(api_routes.groups_ai_providers_settings, json=update, headers=unique_user.token)
assert response.status_code == 200
settings = response.json()
assert settings["defaultProviderId"] == str(default_provider.id)
assert settings["audioProviderId"] == str(audio_provider.id)
assert settings["imageProviderId"] == str(image_provider.id)
finally:
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
for p in [default_provider, audio_provider, image_provider]:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(p.id), headers=unique_user.token)
def test_update_settings_clear_providers(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=provider.id, audio_provider_id=None, image_provider_id=None),
)
try:
# Now clear all
update = {"defaultProviderId": None, "audioProviderId": None, "imageProviderId": None}
response = api_client.put(api_routes.groups_ai_providers_settings, json=update, headers=unique_user.token)
assert response.status_code == 200
settings = response.json()
assert settings["defaultProviderId"] is None
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_settings_providers_list_populated(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
response = api_client.get(api_routes.groups_ai_providers_settings, headers=unique_user.token)
assert response.status_code == 200
provider_ids = [p["id"] for p in response.json()["providers"]]
assert str(provider.id) in provider_ids
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
# ==========================================
# Delete provider cascades to settings
# ==========================================
def test_delete_default_provider_clears_settings(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=provider.id, audio_provider_id=None, image_provider_id=None),
)
# Delete the provider
response = api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token
)
assert response.status_code == 200
# Settings should now have nulled-out default
settings = unique_user.repos.group_ai_provider_settings.get_one(unique_user.repos.group_id)
assert settings is not None
assert settings.default_provider_id is None
def test_delete_audio_provider_clears_settings(api_client: TestClient, unique_user: TestUser):
default_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
audio_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="whisper-1", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(
default_provider_id=default_provider.id,
audio_provider_id=audio_provider.id,
image_provider_id=None,
),
)
try:
# Delete only the audio provider
api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(audio_provider.id), headers=unique_user.token
)
settings = unique_user.repos.group_ai_provider_settings.get_one(unique_user.repos.group_id)
assert settings is not None
assert settings.audio_provider_id is None
assert settings.default_provider_id == default_provider.id
finally:
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(default_provider.id), headers=unique_user.token
)
def test_delete_image_provider_clears_settings(api_client: TestClient, unique_user: TestUser):
default_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
image_provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="dall-e-3", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(
default_provider_id=default_provider.id,
audio_provider_id=None,
image_provider_id=image_provider.id,
),
)
try:
# Delete only the image provider
api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(image_provider.id), headers=unique_user.token
)
settings = unique_user.repos.group_ai_provider_settings.get_one(unique_user.repos.group_id)
assert settings is not None
assert settings.image_provider_id is None
assert settings.default_provider_id == default_provider.id
finally:
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(default_provider.id), headers=unique_user.token
)
# ==========================================
# Permissions: can_manage required
# ==========================================
def test_providers_require_can_manage_get(api_client: TestClient, user_tuple: list[TestUser]):
usr, _ = user_tuple
# Ensure user does NOT have can_manage
user = usr.repos.users.get_one(usr.user_id)
assert user
user.can_manage = False
usr.repos.users.update(user.id, user)
# Create a provider so there's something to GET (using repos directly to bypass permission check)
provider = usr.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
response = api_client.get(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=usr.token)
assert response.status_code == 403
finally:
usr.repos.group_ai_providers.delete(provider.id)
def test_providers_require_can_manage_create(api_client: TestClient, user_tuple: list[TestUser]):
usr, _ = user_tuple
user = usr.repos.users.get_one(usr.user_id)
assert user
user.can_manage = False
usr.repos.users.update(user.id, user)
data = {"name": random_string(), "model": "gpt-4o", "apiKey": "test-key"}
response = api_client.post(api_routes.groups_ai_providers_providers, json=data, headers=usr.token)
assert response.status_code == 403
def test_providers_require_can_manage_update(api_client: TestClient, user_tuple: list[TestUser]):
usr, _ = user_tuple
user = usr.repos.users.get_one(usr.user_id)
assert user
user.can_manage = False
usr.repos.users.update(user.id, user)
provider = usr.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
update_data = {"name": provider.name, "model": "gpt-4-turbo", "apiKey": "key"}
response = api_client.put(
api_routes.groups_ai_providers_providers_provider_id(provider.id), json=update_data, headers=usr.token
)
assert response.status_code == 403
finally:
usr.repos.group_ai_providers.delete(provider.id)
def test_providers_require_can_manage_delete(api_client: TestClient, user_tuple: list[TestUser]):
usr, _ = user_tuple
user = usr.repos.users.get_one(usr.user_id)
assert user
user.can_manage = False
usr.repos.users.update(user.id, user)
provider = usr.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
try:
response = api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=usr.token
)
assert response.status_code == 403
finally:
usr.repos.group_ai_providers.delete(provider.id)
def test_settings_require_can_manage_get(api_client: TestClient, user_tuple: list[TestUser]):
usr, _ = user_tuple
user = usr.repos.users.get_one(usr.user_id)
assert user
user.can_manage = False
usr.repos.users.update(user.id, user)
response = api_client.get(api_routes.groups_ai_providers_settings, headers=usr.token)
assert response.status_code == 403
def test_settings_require_can_manage_update(api_client: TestClient, user_tuple: list[TestUser]):
usr, _ = user_tuple
user = usr.repos.users.get_one(usr.user_id)
assert user
user.can_manage = False
usr.repos.users.update(user.id, user)
update = {"defaultProviderId": None, "audioProviderId": None, "imageProviderId": None}
response = api_client.put(api_routes.groups_ai_providers_settings, json=update, headers=usr.token)
assert response.status_code == 403
# ==========================================
# API key not exposed in responses
# ==========================================
def test_api_key_not_in_create_response(api_client: TestClient, unique_user: TestUser):
data = {"name": random_string(), "model": "gpt-4o", "apiKey": "super-secret-key"}
response = api_client.post(api_routes.groups_ai_providers_providers, json=data, headers=unique_user.token)
assert response.status_code == 200
provider = response.json()
try:
assert "apiKey" not in provider
assert "api_key" not in provider
assert "super-secret-key" not in str(provider)
finally:
api_client.delete(
api_routes.groups_ai_providers_providers_provider_id(provider["id"]), headers=unique_user.token
)
def test_api_key_not_in_get_response(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="super-secret-key")
)
try:
response = api_client.get(
api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token
)
assert response.status_code == 200
data = response.json()
assert "apiKey" not in data
assert "api_key" not in data
assert "super-secret-key" not in str(data)
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_api_key_not_in_update_response(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="original-key")
)
try:
update_data = {"name": provider.name, "model": "gpt-4-turbo", "apiKey": "updated-secret-key"}
response = api_client.put(
api_routes.groups_ai_providers_providers_provider_id(provider.id),
json=update_data,
headers=unique_user.token,
)
assert response.status_code == 200
data = response.json()
assert "apiKey" not in data
assert "api_key" not in data
assert "updated-secret-key" not in str(data)
assert "original-key" not in str(data)
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_api_key_not_in_settings_response(api_client: TestClient, unique_user: TestUser):
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="secret-in-settings")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=provider.id, audio_provider_id=None, image_provider_id=None),
)
try:
response = api_client.get(api_routes.groups_ai_providers_settings, headers=unique_user.token)
assert response.status_code == 200
data = response.json()
assert "apiKey" not in data
assert "api_key" not in data
assert "secret-in-settings" not in str(data)
finally:
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
def test_api_key_not_in_groups_self_response(api_client: TestClient, unique_user: TestUser):
"""Ensure the groups/self endpoint does not expose any AI provider data including API keys."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="groups-self-secret")
)
try:
response = api_client.get(api_routes.groups_self, headers=unique_user.token)
assert response.status_code == 200
data = response.json()
assert "api_key" not in str(data)
assert "apiKey" not in str(data)
assert "groups-self-secret" not in str(data)
finally:
api_client.delete(api_routes.groups_ai_providers_providers_provider_id(provider.id), headers=unique_user.token)
# ==========================================
# New group creation creates empty settings singleton
# ==========================================
def test_new_group_has_empty_ai_provider_settings(api_client: TestClient):
"""When a user registers (creating a new group), empty AI provider settings are created."""
registration = user_registration_factory()
response = api_client.post(api_routes.users_register, json=registration.model_dump(by_alias=True))
assert response.status_code == 201
# Login
form_data = {"username": registration.email, "password": registration.password}
response = api_client.post(api_routes.auth_token, data=form_data)
assert response.status_code == 200
token = response.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
# Fetch AI provider settings for the newly created group
response = api_client.get(api_routes.groups_ai_providers_settings, headers=headers)
assert response.status_code == 200
settings = response.json()
assert settings["defaultProviderId"] is None
assert settings["audioProviderId"] is None
assert settings["imageProviderId"] is None
assert settings["providers"] == []
def test_new_group_created_via_admin_has_empty_ai_provider_settings(
api_client: TestClient,
admin_token: dict,
):
"""When an admin creates a group, empty AI provider settings are created."""
group_name = random_string()
response = api_client.post(api_routes.admin_groups, json={"name": group_name}, headers=admin_token)
assert response.status_code == 201
group_id = response.json()["id"]
try:
# Create a user in the new group with can_manage=True
user_data = {
"fullName": random_string(),
"username": random_string(),
"email": f"{random_string()}@example.com",
"password": "useruser",
"group": group_name,
"household": "Family",
"admin": False,
"canManage": True,
"tokens": [],
}
response = api_client.post(api_routes.admin_users, json=user_data, headers=admin_token)
assert response.status_code == 201
# Login as the new user
form_data = {"username": user_data["email"], "password": "useruser"}
response = api_client.post(api_routes.auth_token, data=form_data)
assert response.status_code == 200
token = response.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
response = api_client.get(api_routes.groups_ai_providers_settings, headers=headers)
assert response.status_code == 200
settings = response.json()
assert settings["defaultProviderId"] is None
assert settings["audioProviderId"] is None
assert settings["imageProviderId"] is None
assert settings["providers"] == []
finally:
api_client.delete(api_routes.admin_groups_item_id(group_id), headers=admin_token)

115
tests/integration_tests/user_household_tests/test_organize_permissions.py Normal file
View File

@@ -0,0 +1,115 @@
"""Tests that the "User can organize group data" permission gates the
mutation endpoints for foods, tags, and categories (GHSA / issue #6883).
Read endpoints remain open; only POST/PUT/DELETE require the permission.
"""
import pytest
from fastapi.testclient import TestClient
from tests.utils import api_routes
from tests.utils.factories import random_string
from tests.utils.fixture_schemas import TestUser
# (id, collection route, item-id route builder)
RESOURCES = [
("foods", api_routes.foods, api_routes.foods_item_id),
("tags", api_routes.organizers_tags, api_routes.organizers_tags_item_id),
("categories", api_routes.organizers_categories, api_routes.organizers_categories_item_id),
]
RESOURCE_IDS = [resource[0] for resource in RESOURCES]
def set_can_organize(user: TestUser, value: bool) -> None:
db_user = user.repos.users.get_one(user.user_id)
assert db_user
db_user.can_organize = value
user.repos.users.update(db_user.id, db_user)
def create_item(api_client: TestClient, user: TestUser, collection: str) -> str:
"""Creates an item as a user with the organize permission and returns its id."""
set_can_organize(user, True)
response = api_client.post(collection, json={"name": random_string(10)}, headers=user.token)
assert response.status_code == 201
return response.json()["id"]
@pytest.mark.parametrize("name, collection, item", RESOURCES, ids=RESOURCE_IDS)
def test_create_requires_organize_permission(
api_client: TestClient, unique_user_fn_scoped: TestUser, name: str, collection: str, item
):
user = unique_user_fn_scoped
payload = {"name": random_string(10)}
set_can_organize(user, False)
response = api_client.post(collection, json=payload, headers=user.token)
assert response.status_code == 403
set_can_organize(user, True)
response = api_client.post(collection, json=payload, headers=user.token)
assert response.status_code == 201
@pytest.mark.parametrize("name, collection, item", RESOURCES, ids=RESOURCE_IDS)
def test_update_requires_organize_permission(
api_client: TestClient, unique_user_fn_scoped: TestUser, name: str, collection: str, item
):
user = unique_user_fn_scoped
item_id = create_item(api_client, user, collection)
payload = {"id": item_id, "name": random_string(10)}
set_can_organize(user, False)
response = api_client.put(item(item_id), json=payload, headers=user.token)
assert response.status_code == 403
set_can_organize(user, True)
response = api_client.put(item(item_id), json=payload, headers=user.token)
assert response.status_code == 200
@pytest.mark.parametrize("name, collection, item", RESOURCES, ids=RESOURCE_IDS)
def test_delete_requires_organize_permission(
api_client: TestClient, unique_user_fn_scoped: TestUser, name: str, collection: str, item
):
user = unique_user_fn_scoped
item_id = create_item(api_client, user, collection)
set_can_organize(user, False)
response = api_client.delete(item(item_id), headers=user.token)
assert response.status_code == 403
set_can_organize(user, True)
response = api_client.delete(item(item_id), headers=user.token)
assert response.status_code == 200
def test_food_merge_requires_organize_permission(api_client: TestClient, unique_user_fn_scoped: TestUser):
user = unique_user_fn_scoped
from_food = create_item(api_client, user, api_routes.foods)
to_food = create_item(api_client, user, api_routes.foods)
payload = {"fromFood": from_food, "toFood": to_food}
set_can_organize(user, False)
response = api_client.put(api_routes.foods_merge, json=payload, headers=user.token)
assert response.status_code == 403
set_can_organize(user, True)
response = api_client.put(api_routes.foods_merge, json=payload, headers=user.token)
assert response.status_code == 200
@pytest.mark.parametrize("name, collection, item", RESOURCES, ids=RESOURCE_IDS)
def test_read_endpoints_do_not_require_organize_permission(
api_client: TestClient, unique_user_fn_scoped: TestUser, name: str, collection: str, item
):
user = unique_user_fn_scoped
item_id = create_item(api_client, user, collection)
set_can_organize(user, False)
response = api_client.get(collection, headers=user.token)
assert response.status_code == 200
response = api_client.get(item(item_id), headers=user.token)
assert response.status_code == 200

15
tests/integration_tests/user_household_tests/test_shopping_list_labels.py
View File

@@ -20,6 +20,21 @@ def create_labels(api_client: TestClient, unique_user: TestUser, count: int = 10
return labels
def test_label_create_duplicate_name_returns_400(api_client: TestClient, unique_user_fn_scoped: TestUser):
# Regression test for #7582: POSTing a duplicate label name leaked the
# sqlalchemy IntegrityError as an HTTP 500. The expected behavior, matching
# the other organizer endpoints (foods, units, tools, tags, categories),
# is HTTP 400. The function-scoped fixture avoids leaking the created label
# into the module-scoped `unique_user` group state used by sibling tests.
payload = {"name": random_string(), "color": "#ff0000"}
response = api_client.post(api_routes.groups_labels, json=payload, headers=unique_user_fn_scoped.token)
assert response.status_code == 200
response = api_client.post(api_routes.groups_labels, json=payload, headers=unique_user_fn_scoped.token)
assert response.status_code == 400
def test_new_list_creates_list_labels(api_client: TestClient, unique_user: TestUser):
labels = create_labels(api_client, unique_user)
response = api_client.post(

17
tests/integration_tests/user_recipe_tests/test_recipe_create_from_image.py
View File

@@ -3,6 +3,7 @@ import json
import pytest
from fastapi.testclient import TestClient
from mealie.schema.group.ai_providers import AIProviderCreate, AIProviderSettingsUpdate
from mealie.schema.openai.recipe import (
OpenAIRecipe,
OpenAIRecipeIngredient,
@@ -15,6 +16,22 @@ from tests.utils.factories import random_int, random_string
from tests.utils.fixture_schemas import TestUser
@pytest.fixture(scope="module", autouse=True)
def setup_ai_providers(unique_user: TestUser):
"""Create AI providers for the test group so image-based OpenAI routes are enabled."""
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name="test-provider", model="gpt-4o", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(
default_provider_id=provider.id,
audio_provider_id=None,
image_provider_id=provider.id,
),
)
def test_openai_create_recipe_from_image(
api_client: TestClient,
monkeypatch: pytest.MonkeyPatch,

23
tests/integration_tests/user_recipe_tests/test_recipe_create_from_openai.py
View File

@@ -4,7 +4,7 @@ import pytest
from fastapi.testclient import TestClient
import mealie.services.scraper.recipe_scraper as recipe_scraper_module
import mealie.services.scraper.scraper_strategies as scraper_strategies_module
from mealie.schema.group.ai_providers import AIProviderCreate, AIProviderSettingsUpdate
from mealie.schema.openai.general import OpenAIText
from mealie.services.openai import OpenAIService
from mealie.services.recipe.recipe_data_service import RecipeDataService
@@ -47,12 +47,17 @@ def recipe_url() -> str:
@pytest.fixture(autouse=True)
def openai_scraper_setup(monkeypatch: pytest.MonkeyPatch, bare_html: str):
"""Restrict to only RecipeScraperOpenAI, enable it unconditionally, and prevent real HTTP calls."""
def openai_scraper_setup(monkeypatch: pytest.MonkeyPatch, bare_html: str, unique_user: TestUser):
"""Restrict to only RecipeScraperOpenAI, create real DB provider data, and prevent real HTTP calls."""
monkeypatch.setattr(recipe_scraper_module, "DEFAULT_SCRAPER_STRATEGIES", [RecipeScraperOpenAI])
settings_stub = type("_Settings", (), {"OPENAI_ENABLED": True})()
monkeypatch.setattr(scraper_strategies_module, "get_app_settings", lambda: settings_stub)
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=provider.id, audio_provider_id=None, image_provider_id=None),
)
async def mock_safe_scrape_html(url: str) -> str:
return bare_html
@@ -171,9 +176,11 @@ def test_create_by_url_openai_disabled(
monkeypatch: pytest.MonkeyPatch,
recipe_url: str,
):
"""When OPENAI_ENABLED is False, can_scrape() returns False and the endpoint returns 400."""
disabled_settings = type("_Settings", (), {"OPENAI_ENABLED": False})()
monkeypatch.setattr(scraper_strategies_module, "get_app_settings", lambda: disabled_settings)
"""When no default provider is set, can_scrape() returns False and the endpoint returns 400."""
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
response = api_client.post(
api_routes.recipes_create_url,

22
tests/integration_tests/user_recipe_tests/test_recipe_create_from_video.py
View File

@@ -6,7 +6,7 @@ from fastapi.testclient import TestClient
import mealie.services.scraper.recipe_scraper as recipe_scraper_module
from mealie.core import exceptions
from mealie.core.config import get_app_settings
from mealie.schema.group.ai_providers import AIProviderCreate, AIProviderSettingsUpdate
from mealie.schema.openai.recipe import OpenAIRecipe, OpenAIRecipeIngredient, OpenAIRecipeInstruction
from mealie.services.openai import OpenAIService
from mealie.services.scraper.scraper_strategies import RecipeScraperOpenAITranscription
@@ -27,10 +27,22 @@ def _make_openai_recipe() -> OpenAIRecipe:
@pytest.fixture(autouse=True)
def video_scraper_setup(monkeypatch: pytest.MonkeyPatch):
def video_scraper_setup(monkeypatch: pytest.MonkeyPatch, unique_user: TestUser):
# Restrict to only the video scraper so other strategies don't interfere
monkeypatch.setattr(recipe_scraper_module, "DEFAULT_SCRAPER_STRATEGIES", [RecipeScraperOpenAITranscription])
provider = unique_user.repos.group_ai_providers.create(
AIProviderCreate(name=random_string(), model="gpt-4o", api_key="test-key")
)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(
default_provider_id=provider.id,
audio_provider_id=provider.id,
image_provider_id=None,
),
)
# Prevent any real HTTP calls during scraping
async def mock_safe_scrape_html(url: str) -> str:
return "<html></html>"
@@ -117,8 +129,10 @@ def test_create_recipe_from_video_transcription_disabled(
monkeypatch: pytest.MonkeyPatch,
unique_user: TestUser,
):
settings = get_app_settings()
monkeypatch.setattr(settings, "OPENAI_ENABLE_TRANSCRIPTION_SERVICES", False)
unique_user.repos.group_ai_provider_settings.update(
unique_user.repos.group_id,
AIProviderSettingsUpdate(default_provider_id=None, audio_provider_id=None, image_provider_id=None),
)
r = api_client.post(api_routes.recipes_create_url, json={"url": VIDEO_URL}, headers=unique_user.token)
assert r.status_code == 400

17
tests/integration_tests/user_recipe_tests/test_recipe_cross_household.py
View File

@@ -201,19 +201,12 @@ def test_delete_recipes_from_other_households(
assert recipe_json["id"] == h2_recipe_id
response = api_client.delete(api_routes.recipes_slug(recipe_json["slug"]), headers=unique_user.token)
if household_lock_recipe_edits:
assert response.status_code == 403
assert response.status_code == 403
# confirm the recipe still exists
response = api_client.get(api_routes.recipes_slug(h2_recipe_id), headers=unique_user.token)
assert response.status_code == 200
assert response.json()["id"] == h2_recipe_id
else:
assert response.status_code == 200
# confirm the recipe was deleted
response = api_client.get(api_routes.recipes_slug(h2_recipe_id), headers=unique_user.token)
assert response.status_code == 404
# confirm the recipe still exists
response = api_client.get(api_routes.recipes_slug(h2_recipe_id), headers=unique_user.token)
assert response.status_code == 200
assert response.json()["id"] == h2_recipe_id
@pytest.mark.parametrize("is_private_household", [True, False])

17
tests/integration_tests/user_recipe_tests/test_recipe_image_assets.py
View File

@@ -87,6 +87,23 @@ def test_recipe_asset_exploit(api_client: TestClient, unique_user: TestUser, rec
assert not (recipe.asset_dir / "test.txt").exists()
def test_recipe_asset_dangerous_extension_blocked(
api_client: TestClient, unique_user: TestUser, recipe_ingredient_only: Recipe
):
"""Ensure scriptable extensions are rejected to prevent stored XSS (GHSA-gfwc-pjx4-mg9p)."""
recipe = recipe_ingredient_only
for ext in ("html", "svg", "js", "htm", "xhtml"):
payload = {"name": random_string(10), "icon": "mdi-file", "extension": ext}
file_payload = {"file": b"<script>alert(1)</script>"}
response = api_client.post(
f"/api/recipes/{recipe.slug}/assets",
data=payload,
files=file_payload,
headers=unique_user.token,
)
assert response.status_code == 400, f"expected 400 for extension={ext}, got {response.status_code}"
def test_recipe_image_upload(api_client: TestClient, unique_user: TestUser, recipe_ingredient_only: Recipe):
data_payload = {"extension": "jpg"}
file_payload = {"image": data.images_test_image_1.read_bytes()}

42
tests/integration_tests/user_recipe_tests/test_recipe_owner.py
View File

@@ -160,6 +160,24 @@ def test_other_user_cant_delete_recipe(api_client: TestClient, user_tuple: list[
assert response.status_code == 403
def test_other_user_cant_delete_unlocked_recipe(api_client: TestClient, user_tuple: list[TestUser]):
"""Non-owner must not delete an unlocked recipe — BOLA regression (GHSA-x5v9-9jvh-7c7q)."""
slug = random_string(10)
unique_user, other_user = user_tuple
unique_user.repos.recipes.create(
Recipe(
user_id=unique_user.user_id,
group_id=unique_user.group_id,
name=slug,
settings=RecipeSettings(locked=False),
)
)
response = api_client.delete(api_routes.recipes_slug(slug), headers=other_user.token)
assert response.status_code == 403
def test_other_user_bulk_delete(api_client: TestClient, user_tuple: list[TestUser]):
slug_locked = random_string(10)
slug_unlocked = random_string(10)
@@ -190,6 +208,30 @@ def test_other_user_bulk_delete(api_client: TestClient, user_tuple: list[TestUse
assert response.status_code == 403
def test_other_user_cant_bulk_delete_unlocked_recipes(api_client: TestClient, user_tuple: list[TestUser]):
"""Non-owner must not bulk-delete unlocked recipes — BOLA regression (GHSA-x5v9-9jvh-7c7q)."""
slug_1 = random_string(10)
slug_2 = random_string(10)
unique_user, other_user = user_tuple
for slug in (slug_1, slug_2):
unique_user.repos.recipes.create(
Recipe(
user_id=unique_user.user_id,
group_id=unique_user.group_id,
name=slug,
settings=RecipeSettings(locked=False),
)
)
response = api_client.post(
api_routes.recipes_bulk_actions_delete,
json={"recipes": [slug_1, slug_2]},
headers=other_user.token,
)
assert response.status_code == 403
def test_admin_can_delete_locked_recipe_owned_by_another_user(
api_client: TestClient, unfiltered_database: AllRepositories, unique_user: TestUser, admin_user: TestUser
):

50
tests/integration_tests/user_recipe_tests/test_recipe_timeline_events.py
View File

@@ -3,18 +3,24 @@ from uuid import uuid4
import pytest
from fastapi.testclient import TestClient
from mealie.lang.providers import get_all_translations
from mealie.schema.recipe.recipe import Recipe
from mealie.schema.recipe.recipe_timeline_events import (
RecipeTimelineEventOut,
RecipeTimelineEventPagination,
TimelineEventImage,
TimelineEventType,
)
from mealie.schema.recipe.request_helpers import UpdateImageResponse
from mealie.services.recipe.recipe_service import RECIPE_CREATED_EVENT_SUBJECT
from tests.utils import api_routes
from tests.utils.factories import random_string
from tests.utils.fixture_schemas import TestUser
PERSISTED_TRANSLATION_KEYS = [RECIPE_CREATED_EVENT_SUBJECT]
@pytest.fixture(scope="function")
def recipes(api_client: TestClient, unique_user: TestUser):
recipes = []
@@ -341,6 +347,50 @@ def test_create_recipe_with_timeline_event(
assert events_pagination.items
@pytest.mark.parametrize("translation_key", PERSISTED_TRANSLATION_KEYS)
def test_persisted_translation_keys_have_translations(translation_key: str):
translations = get_all_translations(translation_key)
missing_translations = [locale for locale, translation in translations.items() if translation == translation_key]
assert missing_translations == []
def test_recipe_created_system_event_is_translated(
api_client: TestClient,
unique_user: TestUser,
recipes: list[Recipe],
):
recipe = recipes[0]
params = {"queryFilter": f"recipe_id={recipe.id}"}
# fetch events in French — the system "recipe created" event should be translated
fr_headers = {**unique_user.token, "Accept-Language": "fr-FR"}
events_response = api_client.get(api_routes.recipes_timeline_events, params=params, headers=fr_headers)
assert events_response.status_code == 200
events_pagination = RecipeTimelineEventPagination.model_validate(events_response.json())
system_events = [e for e in events_pagination.items if e.event_type == TimelineEventType.system.value]
assert system_events, "expected at least one system event for a newly created recipe"
for event in system_events:
assert event.subject == "Recette créée", f"expected French translation, got: {event.subject!r}"
# also verify the individual GET endpoint translates correctly
single_response = api_client.get(api_routes.recipes_timeline_events_item_id(event.id), headers=fr_headers)
assert single_response.status_code == 200
single_event = RecipeTimelineEventOut.model_validate(single_response.json())
assert single_event.subject == "Recette créée"
# fetch the same events in English — subject should be the English string
en_headers = {**unique_user.token, "Accept-Language": "en-US"}
events_response = api_client.get(api_routes.recipes_timeline_events, params=params, headers=en_headers)
events_pagination = RecipeTimelineEventPagination.model_validate(events_response.json())
system_events = [e for e in events_pagination.items if e.event_type == TimelineEventType.system.value]
for event in system_events:
assert event.subject == "Recipe Created", f"expected English string, got: {event.subject!r}"
@pytest.mark.parametrize("use_other_household_user", [True, False])
def test_invalid_recipe_id(
api_client: TestClient, unique_user: TestUser, h2_user: TestUser, use_other_household_user: bool

83
tests/unit_tests/repository_tests/test_query_filter_builder.py
View File

@@ -1,5 +1,11 @@
import pytest
import sqlalchemy as sa
from mealie.db.models.recipe.recipe import RecipeModel
from mealie.db.models.users.users import LongLiveToken, User
from mealie.services.query_filter.builder import (
LogicalOperator,
NonFilterableValueError,
QueryFilterBuilder,
QueryFilterJSON,
QueryFilterJSONPart,
@@ -74,3 +80,80 @@ def test_query_filter_builder_json_uses_raw_value():
),
]
)
# ---------------------------------------------------------------------------
# FilterableColumn tests
# ---------------------------------------------------------------------------
def test_non_filterable_field_user_password_raises():
"""Filtering on User.password (plain Mapped, not FilterableColumn) should raise ValueError."""
with pytest.raises(NonFilterableValueError):
QueryFilterBuilder.get_model_and_model_attr_from_attr_string("password", User)
def test_non_filterable_field_user_email_raises():
"""Filtering on User.email (plain Mapped, not FilterableColumn) should raise ValueError."""
with pytest.raises(NonFilterableValueError):
QueryFilterBuilder.get_model_and_model_attr_from_attr_string("email", User)
def test_non_filterable_field_long_live_token_raises():
"""Filtering on LongLiveToken.token (plain Mapped, not FilterableColumn) should raise ValueError."""
with pytest.raises(NonFilterableValueError):
QueryFilterBuilder.get_model_and_model_attr_from_attr_string("token", LongLiveToken)
def test_filterable_field_does_not_raise():
"""Filtering on a FilterableColumn field should not raise."""
model, attr, _ = QueryFilterBuilder.get_model_and_model_attr_from_attr_string("full_name", User)
assert model is User
assert attr is User.full_name
# ---------------------------------------------------------------------------
# Relationship traversal tests
# ---------------------------------------------------------------------------
def test_deep_traversal_to_filterable_field_works():
"""Traversing a relationship to a FilterableColumn field should succeed."""
model, attr, _ = QueryFilterBuilder.get_model_and_model_attr_from_attr_string("user.full_name", RecipeModel)
assert model is User
assert attr is User.full_name
def test_deep_traversal_to_non_filterable_field_raises():
"""Traversing a relationship to a plain Mapped field should raise ValueError."""
with pytest.raises(NonFilterableValueError):
QueryFilterBuilder.get_model_and_model_attr_from_attr_string("user.email", RecipeModel)
def test_deep_traversal_user_password_raises():
"""Traversing RecipeModel.user.password should raise ValueError."""
with pytest.raises(NonFilterableValueError):
QueryFilterBuilder.get_model_and_model_attr_from_attr_string("user.password", RecipeModel)
def test_filter_query_user_email_raises():
"""filter_query on user.email should raise ValueError."""
query = sa.select(RecipeModel)
builder = QueryFilterBuilder('user.email = "test@example.com"')
with pytest.raises(NonFilterableValueError):
builder.filter_query(query, RecipeModel)
def test_filter_query_user_password_raises():
"""filter_query on user.password should raise ValueError."""
query = sa.select(RecipeModel)
builder = QueryFilterBuilder('user.password = "secret"')
with pytest.raises(NonFilterableValueError):
builder.filter_query(query, RecipeModel)
def test_association_proxy_resolving_to_filterable_field_works():
"""Single-hop association proxy (e.g. household_id) resolving to a FilterableColumn should succeed."""
model, attr, _ = QueryFilterBuilder.get_model_and_model_attr_from_attr_string("household_id", RecipeModel)
assert model is User
assert attr is User.household_id

206
tests/unit_tests/schema_tests/test_ai_providers.py Normal file
View File

@@ -0,0 +1,206 @@
from uuid import uuid4
import pytest
from pydantic import ValidationError
from mealie.schema.group.ai_providers import (
AIProviderCreate,
AIProviderSettingsOut,
AIProviderSummary,
)
class AIProviderCreateTests:
def test_valid_create(self):
provider = AIProviderCreate(name="test", api_key="key", model="gpt-4o")
assert provider.name == "test"
assert provider.model == "gpt-4o"
assert provider.timeout == 300
assert provider.base_url is None
@pytest.mark.parametrize("field", ["name", "api_key", "model"])
def test_empty_field_raises(self, field: str):
data: dict = {"name": "test", "api_key": "key", "model": "gpt-4o", field: ""}
with pytest.raises(ValidationError):
AIProviderCreate(**data)
@pytest.mark.parametrize("timeout", [-1, -100])
def test_negative_timeout_raises(self, timeout: int):
with pytest.raises(ValidationError):
AIProviderCreate(name="test", api_key="key", model="gpt-4o", timeout=timeout)
def test_zero_timeout_is_valid(self):
provider = AIProviderCreate(name="test", api_key="key", model="gpt-4o", timeout=0)
assert provider.timeout == 0
@pytest.mark.parametrize("base_url", ["", None])
def test_base_url_empty_becomes_none(self, base_url: str | None):
provider = AIProviderCreate(name="test", api_key="key", model="gpt-4o", base_url=base_url)
assert provider.base_url is None
def test_api_key_excluded_from_serialization(self):
provider = AIProviderCreate(name="test", api_key="secret", model="gpt-4o")
dumped = provider.model_dump()
assert "api_key" not in dumped
def test_api_key_excluded_from_json(self):
provider = AIProviderCreate(name="test", api_key="secret", model="gpt-4o")
json_str = provider.model_dump_json()
assert "api_key" not in json_str
assert "secret" not in json_str
class AIProviderSettingsOutTests:
def _make_settings(
self,
*,
default_provider_id=None,
audio_provider_id=None,
image_provider_id=None,
providers=None,
) -> AIProviderSettingsOut:
if providers is None:
providers = []
return AIProviderSettingsOut(
default_provider_id=default_provider_id,
audio_provider_id=audio_provider_id,
image_provider_id=image_provider_id,
providers=providers,
)
# --- ai_enabled ---
def test_ai_enabled_false_when_no_default(self):
s = self._make_settings()
assert not s.ai_enabled
def test_ai_enabled_true_when_default_set(self):
pid = uuid4()
s = self._make_settings(default_provider_id=pid, providers=[AIProviderSummary(id=pid, name="p")])
assert s.ai_enabled
# --- audio_provider_enabled ---
def test_audio_provider_disabled_when_no_default(self):
audio_id = uuid4()
s = self._make_settings(
audio_provider_id=audio_id,
providers=[AIProviderSummary(id=audio_id, name="audio")],
)
# audio_provider_id is valid, but validate_providers sets audio_provider_id to None
# because without default_provider_id, it would be fine; let's test audio_provider_enabled
# which requires ai_enabled to be True
assert not s.ai_enabled
assert not s.audio_provider_enabled
def test_audio_provider_disabled_when_only_default_set(self):
pid = uuid4()
s = self._make_settings(default_provider_id=pid, providers=[AIProviderSummary(id=pid, name="p")])
assert s.ai_enabled
assert not s.audio_provider_enabled
def test_audio_provider_enabled_when_both_set(self):
pid = uuid4()
audio_id = uuid4()
s = self._make_settings(
default_provider_id=pid,
audio_provider_id=audio_id,
providers=[AIProviderSummary(id=pid, name="p"), AIProviderSummary(id=audio_id, name="audio")],
)
assert s.ai_enabled
assert s.audio_provider_enabled
# --- image_provider_enabled ---
def test_image_provider_disabled_when_no_default(self):
image_id = uuid4()
s = self._make_settings(
image_provider_id=image_id,
providers=[AIProviderSummary(id=image_id, name="img")],
)
assert not s.ai_enabled
assert not s.image_provider_enabled
def test_image_provider_disabled_when_only_default_set(self):
pid = uuid4()
s = self._make_settings(default_provider_id=pid, providers=[AIProviderSummary(id=pid, name="p")])
assert s.ai_enabled
assert not s.image_provider_enabled
def test_image_provider_enabled_when_both_set(self):
pid = uuid4()
image_id = uuid4()
s = self._make_settings(
default_provider_id=pid,
image_provider_id=image_id,
providers=[AIProviderSummary(id=pid, name="p"), AIProviderSummary(id=image_id, name="img")],
)
assert s.ai_enabled
assert s.image_provider_enabled
# --- validate_providers model validator ---
def test_validate_providers_strips_unknown_default(self):
s = self._make_settings(default_provider_id=uuid4(), providers=[])
assert s.default_provider_id is None
assert not s.ai_enabled
def test_validate_providers_strips_unknown_audio(self):
pid = uuid4()
providers = [AIProviderSummary(id=pid, name="p")]
s = self._make_settings(default_provider_id=pid, audio_provider_id=uuid4(), providers=providers)
assert s.default_provider_id == pid
assert s.audio_provider_id is None
def test_validate_providers_strips_unknown_image(self):
pid = uuid4()
providers = [AIProviderSummary(id=pid, name="p")]
s = self._make_settings(default_provider_id=pid, image_provider_id=uuid4(), providers=providers)
assert s.default_provider_id == pid
assert s.image_provider_id is None
def test_validate_providers_keeps_valid_ids(self):
pid = uuid4()
audio_id = uuid4()
image_id = uuid4()
providers = [
AIProviderSummary(id=pid, name="p"),
AIProviderSummary(id=audio_id, name="audio"),
AIProviderSummary(id=image_id, name="img"),
]
s = self._make_settings(
default_provider_id=pid,
audio_provider_id=audio_id,
image_provider_id=image_id,
providers=providers,
)
assert s.default_provider_id == pid
assert s.audio_provider_id == audio_id
assert s.image_provider_id == image_id
def test_validate_providers_strips_all_if_empty_list(self):
pid = uuid4()
s = self._make_settings(
default_provider_id=pid,
audio_provider_id=uuid4(),
image_provider_id=uuid4(),
providers=[],
)
assert s.default_provider_id is None
assert s.audio_provider_id is None
assert s.image_provider_id is None
def test_validate_providers_partial_strip(self):
"""Only the IDs pointing to missing providers are stripped."""
pid = uuid4()
audio_id = uuid4()
providers = [AIProviderSummary(id=pid, name="p"), AIProviderSummary(id=audio_id, name="audio")]
s = self._make_settings(
default_provider_id=pid,
audio_provider_id=audio_id,
image_provider_id=uuid4(), # not in list → stripped
providers=providers,
)
assert s.default_provider_id == pid
assert s.audio_provider_id == audio_id
assert s.image_provider_id is None

2
tests/unit_tests/services_tests/backup_v2_tests/test_backup_v2.py
View File

@@ -13,8 +13,8 @@ from mealie.db.models.household.household import Household
from mealie.db.models.household.household_to_recipe import HouseholdToRecipe
from mealie.db.models.household.mealplan import GroupMealPlanRules
from mealie.db.models.household.shopping_list import ShoppingList
from mealie.db.models.labels import MultiPurposeLabel
from mealie.db.models.recipe.ingredient import IngredientFoodModel, IngredientUnitModel
from mealie.db.models.recipe.labels import MultiPurposeLabel
from mealie.db.models.recipe.recipe import RecipeModel
from mealie.db.models.recipe.tool import Tool
from mealie.db.models.users.user_to_recipe import UserToRecipe

19
tests/unit_tests/services_tests/ingredient_parser/test_openai_parser.py
View File

@@ -49,6 +49,12 @@ def test_openai_parser(
monkeypatch.setattr(OpenAIService, "get_response", mock_get_response)
def mock_openai_init(self, repos):
self.repos = repos
self.custom_prompt_dir = None
monkeypatch.setattr(OpenAIService, "__init__", mock_openai_init)
with session_context() as session:
loop = asyncio.get_event_loop()
parser = get_parser(RegisteredParser.openai, unique_local_group_id, session, get_locale_provider())
@@ -69,7 +75,7 @@ def test_openai_parser_sanitize_output(
parsed_ingredient_data: tuple[list[IngredientFood], list[IngredientUnit]], # required so database is populated
monkeypatch: pytest.MonkeyPatch,
):
async def mock_get_raw_response(self, prompt: str, content: list[dict], response_schema) -> MagicMock:
async def mock_get_raw_response(self, prompt: str, content: list[dict], response_schema, provider) -> MagicMock:
# Create data with null character in JSON to test preprocessing
data = OpenAIIngredients(
ingredients=[
@@ -91,6 +97,17 @@ def test_openai_parser_sanitize_output(
# Mock the raw response here since we want to make sure our service executes processing before loading the model
monkeypatch.setattr(OpenAIService, "_get_raw_response", mock_get_raw_response)
def mock_openai_init(self, repos):
from unittest.mock import MagicMock
self.repos = repos
self.custom_prompt_dir = None
self.default_provider = MagicMock()
self.audio_provider = None
self.image_provider = None
monkeypatch.setattr(OpenAIService, "__init__", mock_openai_init)
with session_context() as session:
loop = asyncio.get_event_loop()
parser = get_parser(RegisteredParser.openai, unique_local_group_id, session, get_locale_provider())

4
tests/unit_tests/services_tests/scraper_tests/test_cleaner.py
View File

@@ -49,7 +49,7 @@ def test_html_with_recipe_data():
url = "https://www.bbc.co.uk/food/recipes/healthy_pasta_bake_60759"
translator = get_locale_provider()
open_graph_strategy = RecipeScraperOpenGraph(url, translator)
open_graph_strategy = RecipeScraperOpenGraph(url, translator, None) # type: ignore[arg-type]
recipe_data = open_graph_strategy.get_recipe_fields(path.read_text())
@@ -78,7 +78,7 @@ def test_clean_scraper_preserves_notes():
html = RecipeScraperPackage.ld_json_to_html(ld_json)
scraped = scrape_html(html, org_url="https://example.com", supported_only=False)
translator = get_locale_provider()
strategy = RecipeScraperPackage("https://example.com", translator)
strategy = RecipeScraperPackage("https://example.com", translator, None) # type: ignore[arg-type]
recipe, _ = strategy.clean_scraper(scraped, "https://example.com")

37
tests/unit_tests/services_tests/test_openai_service.py
View File

@@ -1,23 +1,28 @@
from unittest.mock import MagicMock
from uuid import uuid4
import pytest
import mealie.services.openai.openai as openai_module
from mealie.services.openai.openai import OpenAIService
def _make_mock_repos() -> MagicMock:
provider_settings = MagicMock()
provider_settings.ai_enabled = True
provider_settings.default_provider_id = uuid4()
provider_settings.audio_provider_id = None
provider_settings.image_provider_id = None
repos = MagicMock()
repos.group_id = uuid4()
repos.group_ai_provider_settings.get_one.return_value = provider_settings
repos.group_ai_providers.get_one.return_value = MagicMock()
return repos
class _SettingsStub:
OPENAI_ENABLED = True
OPENAI_MODEL = "gpt-4o"
OPENAI_AUDIO_MODEL = "whisper-1"
OPENAI_WORKERS = 1
OPENAI_SEND_DATABASE_DATA = False
OPENAI_ENABLE_IMAGE_SERVICES = True
OPENAI_ENABLE_TRANSCRIPTION_SERVICES = True
OPENAI_CUSTOM_PROMPT_DIR: str | None = None
OPENAI_BASE_URL: str | None = None
OPENAI_API_KEY = "dummy"
OPENAI_REQUEST_TIMEOUT = 30
OPENAI_CUSTOM_HEADERS: dict = {}
OPENAI_CUSTOM_PARAMS: dict = {}
@pytest.fixture()
@@ -39,7 +44,7 @@ def settings_stub(tmp_path, monkeypatch):
def test_get_prompt_default_only(settings_stub):
svc = OpenAIService()
svc = OpenAIService(_make_mock_repos())
out = svc.get_prompt("recipes.parse-recipe-ingredients")
assert out == "DEFAULT PROMPT"
@@ -51,7 +56,7 @@ def test_get_prompt_custom_dir_used(settings_stub, tmp_path):
settings_stub.OPENAI_CUSTOM_PROMPT_DIR = str(custom_dir)
svc = OpenAIService()
svc = OpenAIService(_make_mock_repos())
out = svc.get_prompt("recipes.parse-recipe-ingredients")
assert out == "CUSTOM PROMPT"
@@ -62,7 +67,7 @@ def test_get_prompt_custom_empty_falls_back_to_default(settings_stub, tmp_path):
(custom_dir / "recipes" / "parse-recipe-ingredients.txt").write_text("")
settings_stub.OPENAI_CUSTOM_PROMPT_DIR = str(custom_dir)
svc = OpenAIService()
svc = OpenAIService(_make_mock_repos())
out = svc.get_prompt("recipes.parse-recipe-ingredients")
assert out == "DEFAULT PROMPT"
@@ -73,7 +78,7 @@ def test_get_prompt_raises_when_no_files(settings_stub, monkeypatch):
for p in prompts_dir.rglob("*.txt"):
p.unlink()
svc = OpenAIService()
svc = OpenAIService(_make_mock_repos())
with pytest.raises(OSError) as ei:
svc.get_prompt("recipes.parse-recipe-ingredients")
assert "Unable to load prompt" in str(ei.value)

1
tests/unit_tests/test_config.py
View File

@@ -352,7 +352,6 @@ def test_oidc_settings_validation(data: OIDCValidationCase, monkeypatch: pytest.
def test_sensitive_settings_mask(monkeypatch: pytest.MonkeyPatch):
sensitive_settings = [
"LDAP_QUERY_PASSWORD",
"OPENAI_API_KEY",
"SMTP_USER",
"SMTP_PASSWORD",
"OIDC_CLIENT_SECRET",

26
tests/utils/api_routes/__init__.py
View File

@@ -11,8 +11,6 @@ admin_backups = "/api/admin/backups"
"""`/api/admin/backups`"""
admin_backups_upload = "/api/admin/backups/upload"
"""`/api/admin/backups/upload`"""
admin_debug_openai = "/api/admin/debug/openai"
"""`/api/admin/debug/openai`"""
admin_email = "/api/admin/email"
"""`/api/admin/email`"""
admin_groups = "/api/admin/groups"
@@ -57,6 +55,10 @@ foods = "/api/foods"
"""`/api/foods`"""
foods_merge = "/api/foods/merge"
"""`/api/foods/merge`"""
groups_ai_providers_providers = "/api/groups/ai-providers/providers"
"""`/api/groups/ai-providers/providers`"""
groups_ai_providers_settings = "/api/groups/ai-providers/settings"
"""`/api/groups/ai-providers/settings`"""
groups_households = "/api/groups/households"
"""`/api/groups/households`"""
groups_labels = "/api/groups/labels"
@@ -215,6 +217,21 @@ def admin_backups_file_name_restore(file_name):
return f"{prefix}/admin/backups/{file_name}/restore"
def admin_debug_openai_provider_id(provider_id):
"""`/api/admin/debug/openai/{provider_id}`"""
return f"{prefix}/admin/debug/openai/{provider_id}"
def admin_groups_group_id_ai_providers_providers(group_id):
"""`/api/admin/groups/{group_id}/ai-providers/providers`"""
return f"{prefix}/admin/groups/{group_id}/ai-providers/providers"
def admin_groups_group_id_ai_providers_providers_provider_id(group_id, provider_id):
"""`/api/admin/groups/{group_id}/ai-providers/providers/{provider_id}`"""
return f"{prefix}/admin/groups/{group_id}/ai-providers/providers/{provider_id}"
def admin_groups_item_id(item_id):
"""`/api/admin/groups/{item_id}`"""
return f"{prefix}/admin/groups/{item_id}"
@@ -315,6 +332,11 @@ def foods_item_id(item_id):
return f"{prefix}/foods/{item_id}"
def groups_ai_providers_providers_provider_id(provider_id):
"""`/api/groups/ai-providers/providers/{provider_id}`"""
return f"{prefix}/groups/ai-providers/providers/{provider_id}"
def groups_households_household_slug(household_slug):
"""`/api/groups/households/{household_slug}`"""
return f"{prefix}/groups/households/{household_slug}"