mirror of
https://github.com/mealie-recipes/mealie.git
synced 2025-07-09 03:04:54 -04:00
* fix #967 and test error to catch * add admin tests
This commit is contained in:
parent
c617251f4c
commit
14cc541f7a
@ -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)
|
||||
|
@ -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
|
||||
|
@ -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,
|
||||
|
@ -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"
|
||||
|
@ -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")
|
||||
|
Loading…
x
Reference in New Issue
Block a user