diff --git a/mealie/routes/admin/admin_management_users.py b/mealie/routes/admin/admin_management_users.py index ac47ba93d791..a9e5a1f5be39 100644 --- a/mealie/routes/admin/admin_management_users.py +++ b/mealie/routes/admin/admin_management_users.py @@ -1,12 +1,14 @@ from functools import cached_property -from fastapi import APIRouter, Depends +from fastapi import APIRouter, Depends, HTTPException from pydantic import UUID4 +from mealie.core import security from mealie.routes._base import BaseAdminController, controller from mealie.routes._base.dependencies import SharedDependencies from mealie.routes._base.mixins import CrudMixins from mealie.schema.query import GetAll +from mealie.schema.response.responses import ErrorResponse from mealie.schema.user.user import UserIn, UserOut router = APIRouter(prefix="/users", tags=["Admin: Users"]) @@ -34,8 +36,9 @@ class AdminUserManagementRoutes(BaseAdminController): def get_all(self, q: GetAll = Depends(GetAll)): return self.repo.get_all(start=q.start, limit=q.limit, override_schema=UserOut) - @router.post("", response_model=UserOut) + @router.post("", response_model=UserOut, status_code=201) def create_one(self, data: UserIn): + data.password = security.hash_password(data.password) return self.mixins.create_one(data) @router.get("/{item_id}", response_model=UserOut) @@ -44,6 +47,10 @@ class AdminUserManagementRoutes(BaseAdminController): @router.put("/{item_id}", response_model=UserOut) def update_one(self, item_id: UUID4, data: UserOut): + # Prevent self demotion + if self.deps.acting_user.id == item_id and self.deps.acting_user.admin != data.admin: + raise HTTPException(status_code=403, detail=ErrorResponse.respond("you cannot demote yourself")) + return self.mixins.update_one(data, item_id) @router.delete("/{item_id}", response_model=UserOut) diff --git a/tests/integration_tests/admin_tests/test_admin_user_actions.py b/tests/integration_tests/admin_tests/test_admin_user_actions.py index 1802bc11bbc1..5e85e95fa327 100644 --- a/tests/integration_tests/admin_tests/test_admin_user_actions.py +++ b/tests/integration_tests/admin_tests/test_admin_user_actions.py @@ -1,13 +1,30 @@ -import json - from fastapi.testclient import TestClient +from tests import utils +from tests.utils import routes from tests.utils.app_routes import AppRoutes +from tests.utils.factories import random_email, random_string from tests.utils.fixture_schemas import TestUser -def test_init_superuser(api_client: TestClient, api_routes: AppRoutes, admin_user: TestUser): - response = api_client.get(api_routes.users_id(admin_user.user_id), headers=admin_user.token) +def generate_create_data() -> dict: + return { + "username": random_string(), + "fullName": random_string(), + "email": random_email(), + "admin": False, + "group": "Home", + "advanced": False, + "favoriteRecipes": [], + "canInvite": False, + "canManage": False, + "canOrganize": False, + "password": random_string(), + } + + +def test_init_superuser(api_client: TestClient, admin_user: TestUser): + response = api_client.get(routes.RoutesAdminUsers.item(admin_user.user_id), headers=admin_user.token) assert response.status_code == 200 admin_data = response.json() @@ -20,19 +37,16 @@ def test_init_superuser(api_client: TestClient, api_routes: AppRoutes, admin_use def test_create_user(api_client: TestClient, api_routes: AppRoutes, admin_token): - create_data = { - "fullName": "My New User", - "email": "newuser@email.com", - "password": "MyStrongPassword", - "group": "Home", - "admin": False, - "tokens": [], - } - - response = api_client.post(api_routes.users, json=create_data, headers=admin_token) - + create_data = generate_create_data() + response = api_client.post(routes.RoutesAdminUsers.base, json=create_data, headers=admin_token) assert response.status_code == 201 + form_data = {"username": create_data["email"], "password": create_data["password"]} + header = utils.login(form_data=form_data, api_client=api_client, api_routes=api_routes) + + response = api_client.get(routes.RoutesUsers.self, headers=header) + assert response.status_code == 200 + user_data = response.json() assert user_data["fullName"] == create_data["fullName"] @@ -41,38 +55,31 @@ def test_create_user(api_client: TestClient, api_routes: AppRoutes, admin_token) assert user_data["admin"] == create_data["admin"] -def test_create_user_as_non_admin(api_client: TestClient, api_routes: AppRoutes, user_token): - create_data = { - "fullName": "My New User", - "email": "newuser@email.com", - "password": "MyStrongPassword", - "group": "Home", - "admin": False, - "tokens": [], - } - - response = api_client.post(api_routes.users, json=create_data, headers=user_token) - +def test_create_user_as_non_admin(api_client: TestClient, user_token): + create_data = generate_create_data() + response = api_client.post(routes.RoutesAdminUsers.base, json=create_data, headers=user_token) assert response.status_code == 403 -def test_update_user(api_client: TestClient, api_routes: AppRoutes, admin_user: TestUser): - update_data = { - "id": admin_user.user_id, - "fullName": "Updated Name", - "email": "changeme@email.com", - "group": "Home", - "admin": True, - } - response = api_client.put(api_routes.users_id(admin_user.user_id), headers=admin_user.token, json=update_data) +def test_update_user(api_client: TestClient, admin_user: TestUser): + # Create a new user + create_data = generate_create_data() + response = api_client.post(routes.RoutesAdminUsers.base, json=create_data, headers=admin_user.token) + assert response.status_code == 201 + update_data = response.json() + + # Change data + update_data["fullName"] = random_string() + update_data["email"] = random_email() + + response = api_client.put( + routes.RoutesAdminUsers.item(update_data["id"]), headers=admin_user.token, json=update_data + ) assert response.status_code == 200 - assert json.loads(response.text).get("access_token") -def test_update_other_user_as_not_admin( - api_client: TestClient, api_routes: AppRoutes, unique_user: TestUser, g2_user: TestUser -): +def test_update_other_user_as_not_admin(api_client: TestClient, unique_user: TestUser, g2_user: TestUser): update_data = { "id": unique_user.user_id, "fullName": "Updated Name", @@ -80,19 +87,28 @@ def test_update_other_user_as_not_admin( "group": "Home", "admin": True, } - response = api_client.put(api_routes.users_id(g2_user.user_id), headers=unique_user.token, json=update_data) + response = api_client.put( + routes.RoutesAdminUsers.item(g2_user.user_id), headers=unique_user.token, json=update_data + ) assert response.status_code == 403 -def test_self_demote_admin(api_client: TestClient, api_routes: AppRoutes, admin_user: TestUser): - update_data = {"fullName": "Updated Name", "email": "changeme@email.com", "group": "Home", "admin": False} - response = api_client.put(api_routes.users_id(admin_user.user_id), headers=admin_user.token, json=update_data) +def test_self_demote_admin(api_client: TestClient, admin_user: TestUser): + response = api_client.get(routes.RoutesUsers.self, headers=admin_user.token) + assert response.status_code == 200 + + user_data = response.json() + user_data["admin"] = False + + response = api_client.put( + routes.RoutesAdminUsers.item(admin_user.user_id), headers=admin_user.token, json=user_data + ) assert response.status_code == 403 -def test_self_promote_admin(api_client: TestClient, api_routes: AppRoutes, unique_user: TestUser): +def test_self_promote_admin(api_client: TestClient, unique_user: TestUser): update_data = { "id": unique_user.user_id, "fullName": "Updated Name", @@ -100,12 +116,13 @@ def test_self_promote_admin(api_client: TestClient, api_routes: AppRoutes, uniqu "group": "Home", "admin": True, } - response = api_client.put(api_routes.users_id(unique_user.user_id), headers=unique_user.token, json=update_data) + response = api_client.put( + routes.RoutesAdminUsers.item(unique_user.user_id), headers=unique_user.token, json=update_data + ) assert response.status_code == 403 -def test_delete_user(api_client: TestClient, api_routes: AppRoutes, admin_token, unique_user: TestUser): - response = api_client.delete(api_routes.users_id(unique_user.user_id), headers=admin_token) - +def test_delete_user(api_client: TestClient, admin_token, unique_user: TestUser): + response = api_client.delete(routes.RoutesAdminUsers.item(unique_user.user_id), headers=admin_token) assert response.status_code == 200 diff --git a/tests/integration_tests/recipe_migration_tests/test_recipe_migrations.py b/tests/integration_tests/recipe_migration_tests/test_recipe_migrations.py index 848114e98fd2..6f6f0f5f40f6 100644 --- a/tests/integration_tests/recipe_migration_tests/test_recipe_migrations.py +++ b/tests/integration_tests/recipe_migration_tests/test_recipe_migrations.py @@ -29,8 +29,15 @@ test_cases = [ MigrationTestData(typ=SupportedMigrations.mealie_alpha, archive=test_data.migrations_mealie), ] +test_ids = [ + "nextcloud_archive", + "paprika_archive", + "chowdown_archive", + "mealie_alpha_archive", +] -@pytest.mark.parametrize("mig", test_cases) + +@pytest.mark.parametrize("mig", test_cases, ids=test_ids) def test_recipe_migration(api_client: TestClient, unique_user: TestUser, mig: MigrationTestData) -> None: payload = { "migration_type": mig.typ.value, diff --git a/tests/utils/routes.py b/tests/utils/routes.py index 963b836fafb9..d3c78b8f53c1 100644 --- a/tests/utils/routes.py +++ b/tests/utils/routes.py @@ -41,3 +41,12 @@ class RoutesCategory(RoutesOrganizerBase): class RoutesRecipe(RoutesBase): base = "/api/recipes" + + +class RoutesAdminUsers(RoutesBase): + base = "/api/admin/users" + + +class RoutesUsers(RoutesBase): + base = "/api/users" + self = f"{base}/self" diff --git a/tests/utils/user_login.py b/tests/utils/user_login.py index 38778f2501e6..3b4024bbe81e 100644 --- a/tests/utils/user_login.py +++ b/tests/utils/user_login.py @@ -1,11 +1,11 @@ import json -import requests +from fastapi.testclient import TestClient from .app_routes import AppRoutes -def login(form_data, api_client: requests, api_routes: AppRoutes): +def login(form_data, api_client: TestClient, api_routes: AppRoutes): response = api_client.post(api_routes.auth_token, form_data) assert response.status_code == 200 token = json.loads(response.text).get("access_token")