1
0
Fork 0
mirror of https://github.com/mealie-recipes/mealie.git synced 2025-08-06 22:15:22 +02:00

API security hardening (#571)

* Enhance security and safety around user update API

- Prevent a regular user from promoting themself to admin
- Prevent an admin from demoting themself
- Refactor token fixture to admin + regular user tokens

* Restrict user CRUD API to admins

* Secure admin API routes

* Refactor APIrouter into Admin/UserAPIRouter

* Secure theme routes

* Make 'all recipes' routes public

* Secure favorite routes

* Remove redundant checks

* Fix public routes mistakenly flagged user routes

* Make webhooks changeable only by admin

* Allow users to create categories and tags

* Address lint issues
This commit is contained in:
sephrat 2021-06-22 20:22:15 +02:00 committed by GitHub
parent f5faff66d3
commit 6320ba7ec5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
43 changed files with 456 additions and 347 deletions

View file

@ -10,35 +10,35 @@ recipe_test_data = get_recipe_test_cases()
@pytest.mark.parametrize("recipe_data", recipe_test_data)
def test_create_by_url(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, token):
api_client.delete(api_routes.recipes_recipe_slug(recipe_data.expected_slug), headers=token)
def test_create_by_url(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, user_token):
api_client.delete(api_routes.recipes_recipe_slug(recipe_data.expected_slug), headers=user_token)
response = api_client.post(api_routes.recipes_create_url, json={"url": recipe_data.url}, headers=token)
response = api_client.post(api_routes.recipes_create_url, json={"url": recipe_data.url}, headers=user_token)
assert response.status_code == 201
assert json.loads(response.text) == recipe_data.expected_slug
def test_create_by_json(api_client: TestClient, api_routes: AppRoutes, token, raw_recipe):
def test_create_by_json(api_client: TestClient, api_routes: AppRoutes, user_token, raw_recipe):
recipe_url = api_routes.recipes_recipe_slug("banana-bread")
api_client.delete(recipe_url, headers=token)
response = api_client.post(api_routes.recipes_create, json=raw_recipe, headers=token)
api_client.delete(recipe_url, headers=user_token)
response = api_client.post(api_routes.recipes_create, json=raw_recipe, headers=user_token)
assert response.status_code == 201
assert json.loads(response.text) == "banana-bread"
def test_create_no_image(api_client: TestClient, api_routes: AppRoutes, token, raw_recipe_no_image):
response = api_client.post(api_routes.recipes_create, json=raw_recipe_no_image, headers=token)
def test_create_no_image(api_client: TestClient, api_routes: AppRoutes, user_token, raw_recipe_no_image):
response = api_client.post(api_routes.recipes_create, json=raw_recipe_no_image, headers=user_token)
assert response.status_code == 201
assert json.loads(response.text) == "banana-bread-no-image"
@pytest.mark.parametrize("recipe_data", recipe_test_data)
def test_read_update(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, token):
def test_read_update(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, user_token):
recipe_url = api_routes.recipes_recipe_slug(recipe_data.expected_slug)
response = api_client.get(recipe_url, headers=token)
response = api_client.get(recipe_url, headers=user_token)
assert response.status_code == 200
recipe = json.loads(response.text)
@ -54,7 +54,7 @@ def test_read_update(api_client: TestClient, api_routes: AppRoutes, recipe_data:
test_categories = ["one", "two", "three"]
recipe["recipeCategory"] = test_categories
response = api_client.put(recipe_url, json=recipe, headers=token)
response = api_client.put(recipe_url, json=recipe, headers=user_token)
assert response.status_code == 200
assert json.loads(response.text).get("slug") == recipe_data.expected_slug
@ -69,9 +69,9 @@ def test_read_update(api_client: TestClient, api_routes: AppRoutes, recipe_data:
@pytest.mark.parametrize("recipe_data", recipe_test_data)
def test_rename(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, token):
def test_rename(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, user_token):
recipe_url = api_routes.recipes_recipe_slug(recipe_data.expected_slug)
response = api_client.get(recipe_url, headers=token)
response = api_client.get(recipe_url, headers=user_token)
assert response.status_code == 200
recipe = json.loads(response.text)
@ -79,7 +79,7 @@ def test_rename(api_client: TestClient, api_routes: AppRoutes, recipe_data: Reci
new_slug = slugify(new_name)
recipe["name"] = new_name
response = api_client.put(recipe_url, json=recipe, headers=token)
response = api_client.put(recipe_url, json=recipe, headers=user_token)
assert response.status_code == 200
assert json.loads(response.text).get("slug") == new_slug
@ -88,7 +88,7 @@ def test_rename(api_client: TestClient, api_routes: AppRoutes, recipe_data: Reci
@pytest.mark.parametrize("recipe_data", recipe_test_data)
def test_delete(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, token):
def test_delete(api_client: TestClient, api_routes: AppRoutes, recipe_data: RecipeSiteTestCase, user_token):
recipe_url = api_routes.recipes_recipe_slug(recipe_data.expected_slug)
response = api_client.delete(recipe_url, headers=token)
response = api_client.delete(recipe_url, headers=user_token)
assert response.status_code == 200

View file

@ -10,8 +10,8 @@ def page_data():
return {"name": "My New Page", "position": 0, "categories": []}
def test_create_page(api_client: TestClient, api_routes: AppRoutes, token, page_data):
response = api_client.post(api_routes.site_settings_custom_pages, json=page_data, headers=token)
def test_create_page(api_client: TestClient, api_routes: AppRoutes, admin_token, page_data):
response = api_client.post(api_routes.site_settings_custom_pages, json=page_data, headers=admin_token)
assert response.status_code == 200
@ -25,16 +25,16 @@ def test_read_page(api_client: TestClient, api_routes: AppRoutes, page_data):
assert json.loads(response.text) == page_data
def test_update_page(api_client: TestClient, api_routes: AppRoutes, page_data, token):
def test_update_page(api_client: TestClient, api_routes: AppRoutes, page_data, admin_token):
page_data["id"] = 1
page_data["name"] = "My New Name"
response = api_client.put(api_routes.site_settings_custom_pages_id(1), json=page_data, headers=token)
response = api_client.put(api_routes.site_settings_custom_pages_id(1), json=page_data, headers=admin_token)
assert response.status_code == 200
def test_delete_page(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.delete(api_routes.site_settings_custom_pages_id(1), headers=token)
def test_delete_page(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.delete(api_routes.site_settings_custom_pages_id(1), headers=admin_token)
assert response.status_code == 200

View file

@ -10,20 +10,20 @@ def group_data():
return {"name": "Test Group"}
def test_create_group(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.post(api_routes.groups, json={"name": "Test Group"}, headers=token)
def test_create_group(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.post(api_routes.groups, json={"name": "Test Group"}, headers=admin_token)
assert response.status_code == 201
def test_get_self_group(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.get(api_routes.groups, headers=token)
def test_get_self_group(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.get(api_routes.groups, headers=admin_token)
assert response.status_code == 200
assert len(json.loads(response.text)) >= 2
def test_update_group(api_client: TestClient, api_routes: AppRoutes, token):
def test_update_group(api_client: TestClient, api_routes: AppRoutes, admin_token):
new_data = {
"name": "New Group Name",
"id": 2,
@ -36,23 +36,23 @@ def test_update_group(api_client: TestClient, api_routes: AppRoutes, token):
"shoppingLists": [],
}
# Test Update
response = api_client.put(api_routes.groups_id(2), json=new_data, headers=token)
response = api_client.put(api_routes.groups_id(2), json=new_data, headers=admin_token)
assert response.status_code == 200
# Validate Changes
response = api_client.get(api_routes.groups, headers=token)
response = api_client.get(api_routes.groups, headers=admin_token)
all_groups = json.loads(response.text)
id_2 = filter(lambda x: x["id"] == 2, all_groups)
assert next(id_2) == new_data
def test_home_group_not_deletable(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.delete(api_routes.groups_id(1), headers=token)
def test_home_group_not_deletable(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.delete(api_routes.groups_id(1), headers=admin_token)
assert response.status_code == 400
def test_delete_group(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.delete(api_routes.groups_id(2), headers=token)
def test_delete_group(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.delete(api_routes.groups_id(2), headers=admin_token)
assert response.status_code == 200

View file

@ -19,9 +19,9 @@ def backup_data():
}
def test_import(api_client: TestClient, api_routes: AppRoutes, backup_data, token):
def test_import(api_client: TestClient, api_routes: AppRoutes, backup_data, admin_token):
import_route = api_routes.backups_file_name_import("test_backup_2021-Apr-27.zip")
response = api_client.post(import_route, json=backup_data, headers=token)
response = api_client.post(import_route, json=backup_data, headers=admin_token)
assert response.status_code == 200
for _, value in json.loads(response.content).items():
for v in value:

View file

@ -6,15 +6,15 @@ from tests.app_routes import AppRoutes
@fixture
def long_live_token(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.post(api_routes.users_api_tokens, json={"name": "Test Fixture Token"}, headers=token)
def long_live_token(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.post(api_routes.users_api_tokens, json={"name": "Test Fixture Token"}, headers=admin_token)
assert response.status_code == 201
return {"Authorization": f"Bearer {json.loads(response.text).get('token')}"}
def test_api_token_creation(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.post(api_routes.users_api_tokens, json={"name": "Test API Token"}, headers=token)
def test_api_token_creation(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.post(api_routes.users_api_tokens, json={"name": "Test API Token"}, headers=admin_token)
assert response.status_code == 201
@ -24,9 +24,9 @@ def test_use_token(api_client: TestClient, api_routes: AppRoutes, long_live_toke
assert response.status_code == 200
def test_delete_token(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.delete(api_routes.users_api_tokens_token_id(1), headers=token)
def test_delete_token(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.delete(api_routes.users_api_tokens_token_id(1), headers=admin_token)
assert response.status_code == 200
response = api_client.delete(api_routes.users_api_tokens_token_id(2), headers=token)
response = api_client.delete(api_routes.users_api_tokens_token_id(2), headers=admin_token)
assert response.status_code == 200

View file

@ -25,8 +25,8 @@ def get_meal_plan_template(first=None, second=None):
@pytest.fixture(scope="session")
def slug_1(api_client: TestClient, api_routes: AppRoutes, token, recipe_store: list[RecipeSiteTestCase]):
slug_1 = api_client.post(api_routes.recipes_create_url, json={"url": recipe_store[0].url}, headers=token)
def slug_1(api_client: TestClient, api_routes: AppRoutes, admin_token, recipe_store: list[RecipeSiteTestCase]):
slug_1 = api_client.post(api_routes.recipes_create_url, json={"url": recipe_store[0].url}, headers=admin_token)
slug_1 = json.loads(slug_1.content)
yield slug_1
@ -35,8 +35,8 @@ def slug_1(api_client: TestClient, api_routes: AppRoutes, token, recipe_store: l
@pytest.fixture(scope="session")
def slug_2(api_client: TestClient, api_routes: AppRoutes, token, recipe_store: list[RecipeSiteTestCase]):
slug_2 = api_client.post(api_routes.recipes_create_url, json={"url": recipe_store[1].url}, headers=token)
def slug_2(api_client: TestClient, api_routes: AppRoutes, admin_token, recipe_store: list[RecipeSiteTestCase]):
slug_2 = api_client.post(api_routes.recipes_create_url, json={"url": recipe_store[1].url}, headers=admin_token)
slug_2 = json.loads(slug_2.content)
yield slug_2
@ -44,15 +44,15 @@ def slug_2(api_client: TestClient, api_routes: AppRoutes, token, recipe_store: l
api_client.delete(api_routes.recipes_recipe_slug(slug_2))
def test_create_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, slug_2, token):
def test_create_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, slug_2, admin_token):
meal_plan = get_meal_plan_template(slug_1, slug_2)
response = api_client.post(api_routes.meal_plans_create, json=meal_plan, headers=token)
response = api_client.post(api_routes.meal_plans_create, json=meal_plan, headers=admin_token)
assert response.status_code == 201
def test_read_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, slug_2, token):
response = api_client.get(api_routes.meal_plans_all, headers=token)
def test_read_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, slug_2, admin_token):
response = api_client.get(api_routes.meal_plans_all, headers=admin_token)
assert response.status_code == 200
@ -65,9 +65,9 @@ def test_read_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, sl
assert meals[1]["meals"][0]["slug"] == meal_plan_template["planDays"][1]["meals"][0]["slug"]
def test_update_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, slug_2, token):
def test_update_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1, slug_2, admin_token):
response = api_client.get(api_routes.meal_plans_all, headers=token)
response = api_client.get(api_routes.meal_plans_all, headers=admin_token)
existing_mealplan = json.loads(response.text)
existing_mealplan = existing_mealplan[0]
@ -77,11 +77,11 @@ def test_update_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1,
existing_mealplan["planDays"][0]["meals"][0]["slug"] = slug_2
existing_mealplan["planDays"][1]["meals"][0]["slug"] = slug_1
response = api_client.put(api_routes.meal_plans_plan_id(plan_uid), json=existing_mealplan, headers=token)
response = api_client.put(api_routes.meal_plans_plan_id(plan_uid), json=existing_mealplan, headers=admin_token)
assert response.status_code == 200
response = api_client.get(api_routes.meal_plans_all, headers=token)
response = api_client.get(api_routes.meal_plans_all, headers=admin_token)
existing_mealplan = json.loads(response.text)
existing_mealplan = existing_mealplan[0]
@ -89,14 +89,14 @@ def test_update_mealplan(api_client: TestClient, api_routes: AppRoutes, slug_1,
assert existing_mealplan["planDays"][1]["meals"][0]["slug"] == slug_1
def test_delete_mealplan(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.get(api_routes.meal_plans_all, headers=token)
def test_delete_mealplan(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.get(api_routes.meal_plans_all, headers=admin_token)
assert response.status_code == 200
existing_mealplan = json.loads(response.text)
existing_mealplan = existing_mealplan[0]
plan_uid = existing_mealplan.get("uid")
response = api_client.delete(api_routes.meal_plans_plan_id(plan_uid), headers=token)
response = api_client.delete(api_routes.meal_plans_plan_id(plan_uid), headers=admin_token)
assert response.status_code == 200

View file

@ -22,22 +22,22 @@ def chowdown_zip():
zip_copy.unlink()
def test_upload_chowdown_zip(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, token):
def test_upload_chowdown_zip(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, admin_token):
upload_url = api_routes.migrations_import_type_upload("chowdown")
response = api_client.post(upload_url, files={"archive": chowdown_zip.open("rb")}, headers=token)
response = api_client.post(upload_url, files={"archive": chowdown_zip.open("rb")}, headers=admin_token)
assert response.status_code == 200
assert app_dirs.MIGRATION_DIR.joinpath("chowdown", chowdown_zip.name).is_file()
def test_import_chowdown_directory(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, token):
def test_import_chowdown_directory(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, admin_token):
delete_url = api_routes.recipes_recipe_slug("roasted-okra")
api_client.delete(delete_url, headers=token) # TODO: Manage Test Data better
api_client.delete(delete_url, headers=admin_token) # TODO: Manage Test Data better
selection = chowdown_zip.name
import_url = api_routes.migrations_import_type_file_name_import("chowdown", selection)
response = api_client.post(import_url, headers=token)
response = api_client.post(import_url, headers=admin_token)
assert response.status_code == 200
@ -47,10 +47,10 @@ def test_import_chowdown_directory(api_client: TestClient, api_routes: AppRoutes
assert report.get("status") is True
def test_delete_chowdown_migration_data(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, token):
def test_delete_chowdown_migration_data(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, admin_token):
selection = chowdown_zip.name
delete_url = api_routes.migrations_import_type_file_name_delete("chowdown", selection)
response = api_client.delete(delete_url, headers=token)
response = api_client.delete(delete_url, headers=admin_token)
assert response.status_code == 200
assert not app_dirs.MIGRATION_DIR.joinpath(chowdown_zip.name).is_file()
@ -70,19 +70,19 @@ def nextcloud_zip():
zip_copy.unlink()
def test_upload_nextcloud_zip(api_client: TestClient, api_routes: AppRoutes, nextcloud_zip, token):
def test_upload_nextcloud_zip(api_client: TestClient, api_routes: AppRoutes, nextcloud_zip, admin_token):
upload_url = api_routes.migrations_import_type_upload("nextcloud")
response = api_client.post(upload_url, files={"archive": nextcloud_zip.open("rb")}, headers=token)
response = api_client.post(upload_url, files={"archive": nextcloud_zip.open("rb")}, headers=admin_token)
assert response.status_code == 200
assert app_dirs.MIGRATION_DIR.joinpath("nextcloud", nextcloud_zip.name).is_file()
def test_import_nextcloud_directory(api_client: TestClient, api_routes: AppRoutes, nextcloud_zip, token):
def test_import_nextcloud_directory(api_client: TestClient, api_routes: AppRoutes, nextcloud_zip, admin_token):
selection = nextcloud_zip.name
import_url = api_routes.migrations_import_type_file_name_import("nextcloud", selection)
response = api_client.post(import_url, headers=token)
response = api_client.post(import_url, headers=admin_token)
assert response.status_code == 200
@ -91,10 +91,12 @@ def test_import_nextcloud_directory(api_client: TestClient, api_routes: AppRoute
assert report.get("status") is True
def test_delete__nextcloud_migration_data(api_client: TestClient, api_routes: AppRoutes, nextcloud_zip: Path, token):
def test_delete__nextcloud_migration_data(
api_client: TestClient, api_routes: AppRoutes, nextcloud_zip: Path, admin_token
):
selection = nextcloud_zip.name
delete_url = api_routes.migrations_import_type_file_name_delete("nextcloud", selection)
response = api_client.delete(delete_url, headers=token)
response = api_client.delete(delete_url, headers=admin_token)
assert response.status_code == 200
assert not app_dirs.MIGRATION_DIR.joinpath(nextcloud_zip.name).is_file()

View file

@ -19,11 +19,11 @@ def test_default_settings(api_client: TestClient, api_routes: AppRoutes, default
assert json.loads(response.content) == default_settings
def test_update_settings(api_client: TestClient, api_routes: AppRoutes, default_settings, token):
def test_update_settings(api_client: TestClient, api_routes: AppRoutes, default_settings, admin_token):
default_settings["language"] = "fr"
default_settings["showRecent"] = False
response = api_client.put(api_routes.site_settings, json=default_settings, headers=token)
response = api_client.put(api_routes.site_settings, json=default_settings, headers=admin_token)
assert response.status_code == 200

View file

@ -7,10 +7,10 @@ from tests.app_routes import AppRoutes
@pytest.fixture()
def active_link(api_client: TestClient, api_routes: AppRoutes, token):
def active_link(api_client: TestClient, api_routes: AppRoutes, admin_token):
data = {"name": "Fixture Token", "admin": True}
response = api_client.post(api_routes.users_sign_ups, json=data, headers=token)
response = api_client.post(api_routes.users_sign_ups, json=data, headers=admin_token)
return SignUpToken(**json.loads(response.text))
@ -26,10 +26,10 @@ def sign_up_user():
}
def test_create_sign_up_link(api_client: TestClient, api_routes: AppRoutes, token):
def test_create_sign_up_link(api_client: TestClient, api_routes: AppRoutes, admin_token):
data = {"name": "Test Token", "admin": False}
response = api_client.post(api_routes.users_sign_ups, json=data, headers=token)
response = api_client.post(api_routes.users_sign_ups, json=data, headers=admin_token)
assert response.status_code == 200
@ -47,11 +47,11 @@ def test_new_user_signup(api_client: TestClient, api_routes: AppRoutes, active_l
def test_delete_sign_up_link(
api_client: TestClient, api_routes: AppRoutes, token, active_link: SignUpToken, sign_up_user
api_client: TestClient, api_routes: AppRoutes, admin_token, active_link: SignUpToken, sign_up_user
):
response = api_client.delete(api_routes.users_sign_ups_token(active_link.token), headers=token)
response = api_client.delete(api_routes.users_sign_ups_token(active_link.token), headers=admin_token)
assert response.status_code == 200
# Validate Token is Gone
response = api_client.get(api_routes.users_sign_ups, headers=token)
# Validate admin_token is Gone
response = api_client.get(api_routes.users_sign_ups, headers=admin_token)
assert sign_up_user not in json.loads(response.content)

View file

@ -28,38 +28,38 @@ def new_theme():
}
def test_default_theme(api_client: TestClient, api_routes: AppRoutes, default_theme):
response = api_client.get(api_routes.themes_id(1))
def test_default_theme(api_client: TestClient, api_routes: AppRoutes, default_theme, user_token):
response = api_client.get(api_routes.themes_id(1), headers=user_token)
assert response.status_code == 200
assert json.loads(response.content) == default_theme
def test_create_theme(api_client: TestClient, api_routes: AppRoutes, new_theme, token):
def test_create_theme(api_client: TestClient, api_routes: AppRoutes, new_theme, user_token):
response = api_client.post(api_routes.themes_create, json=new_theme, headers=token)
response = api_client.post(api_routes.themes_create, json=new_theme, headers=user_token)
assert response.status_code == 201
response = api_client.get(api_routes.themes_id(new_theme.get("id")), headers=token)
response = api_client.get(api_routes.themes_id(new_theme.get("id")), headers=user_token)
assert response.status_code == 200
assert json.loads(response.content) == new_theme
def test_read_all_themes(api_client: TestClient, api_routes: AppRoutes, default_theme, new_theme):
response = api_client.get(api_routes.themes)
def test_read_all_themes(api_client: TestClient, api_routes: AppRoutes, default_theme, new_theme, user_token):
response = api_client.get(api_routes.themes, headers=user_token)
assert response.status_code == 200
response_dict = json.loads(response.content)
assert default_theme in response_dict
assert new_theme in response_dict
def test_read_theme(api_client: TestClient, api_routes: AppRoutes, default_theme, new_theme):
def test_read_theme(api_client: TestClient, api_routes: AppRoutes, default_theme, new_theme, user_token):
for theme in [default_theme, new_theme]:
response = api_client.get(api_routes.themes_id(theme.get("id")))
response = api_client.get(api_routes.themes_id(theme.get("id")), headers=user_token)
assert response.status_code == 200
assert json.loads(response.content) == theme
def test_update_theme(api_client: TestClient, api_routes: AppRoutes, token, new_theme):
def test_update_theme(api_client: TestClient, api_routes: AppRoutes, user_token, new_theme):
theme_colors = {
"primary": "#E12345",
"accent": "#012345",
@ -72,14 +72,14 @@ def test_update_theme(api_client: TestClient, api_routes: AppRoutes, token, new_
new_theme["colors"] = theme_colors
new_theme["name"] = "New Theme Name"
response = api_client.put(api_routes.themes_id(new_theme.get("id")), json=new_theme, headers=token)
response = api_client.put(api_routes.themes_id(new_theme.get("id")), json=new_theme, headers=user_token)
assert response.status_code == 200
response = api_client.get(api_routes.themes_id(new_theme.get("id")))
response = api_client.get(api_routes.themes_id(new_theme.get("id")), headers=user_token)
assert json.loads(response.content) == new_theme
def test_delete_theme(api_client: TestClient, api_routes: AppRoutes, default_theme, new_theme, token):
def test_delete_theme(api_client: TestClient, api_routes: AppRoutes, default_theme, new_theme, user_token):
for theme in [default_theme, new_theme]:
response = api_client.delete(api_routes.themes_id(theme.get("id")), headers=token)
response = api_client.delete(api_routes.themes_id(theme.get("id")), headers=user_token)
assert response.status_code == 200

View file

@ -9,7 +9,7 @@ from tests.app_routes import AppRoutes
@fixture(scope="session")
def default_user():
def admin_user():
return UserOut(
id=1,
fullName="Change Me",
@ -24,7 +24,7 @@ def default_user():
@fixture(scope="session")
def new_user():
return UserOut(
id=3,
id=4,
fullName="My New User",
username="My New User",
email="newuser@email.com",
@ -41,27 +41,27 @@ def test_failed_login(api_client: TestClient, api_routes: AppRoutes):
assert response.status_code == 401
def test_superuser_login(api_client: TestClient, api_routes: AppRoutes, token):
def test_superuser_login(api_client: TestClient, api_routes: AppRoutes, admin_token):
form_data = {"username": "changeme@email.com", "password": "MyPassword"}
response = api_client.post(api_routes.auth_token, form_data)
assert response.status_code == 200
new_token = json.loads(response.text).get("access_token")
response = api_client.get(api_routes.users_self, headers=token)
response = api_client.get(api_routes.users_self, headers=admin_token)
assert response.status_code == 200
return {"Authorization": f"Bearer {new_token}"}
def test_init_superuser(api_client: TestClient, api_routes: AppRoutes, token, default_user: UserOut):
response = api_client.get(api_routes.users_id(1), headers=token)
def test_init_superuser(api_client: TestClient, api_routes: AppRoutes, admin_token, admin_user: UserOut):
response = api_client.get(api_routes.users_id(1), headers=admin_token)
assert response.status_code == 200
assert json.loads(response.text) == default_user.dict(by_alias=True)
assert json.loads(response.text) == admin_user.dict(by_alias=True)
def test_create_user(api_client: TestClient, api_routes: AppRoutes, token, new_user):
def test_create_user(api_client: TestClient, api_routes: AppRoutes, admin_token, new_user):
create_data = {
"fullName": "My New User",
"email": "newuser@email.com",
@ -71,32 +71,74 @@ def test_create_user(api_client: TestClient, api_routes: AppRoutes, token, new_u
"tokens": [],
}
response = api_client.post(api_routes.users, json=create_data, headers=token)
response = api_client.post(api_routes.users, json=create_data, headers=admin_token)
assert response.status_code == 201
assert json.loads(response.text) == new_user.dict(by_alias=True)
assert True
def test_get_all_users(api_client: TestClient, api_routes: AppRoutes, token, new_user, default_user):
response = api_client.get(api_routes.users, headers=token)
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)
assert response.status_code == 403
def test_get_all_users(api_client: TestClient, api_routes: AppRoutes, admin_token, new_user, admin_user):
response = api_client.get(api_routes.users, headers=admin_token)
assert response.status_code == 200
all_users = json.loads(response.text)
assert default_user.dict(by_alias=True) in all_users
assert admin_user.dict(by_alias=True) in all_users
assert new_user.dict(by_alias=True) in all_users
def test_update_user(api_client: TestClient, api_routes: AppRoutes, token):
def test_update_user(api_client: TestClient, api_routes: AppRoutes, admin_token):
update_data = {"id": 1, "fullName": "Updated Name", "email": "changeme@email.com", "group": "Home", "admin": True}
response = api_client.put(api_routes.users_id(1), headers=token, json=update_data)
response = api_client.put(api_routes.users_id(1), headers=admin_token, json=update_data)
assert response.status_code == 200
assert json.loads(response.text).get("access_token")
def test_reset_user_password(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.put(api_routes.users_id_reset_password(3), headers=token)
def test_update_other_user_as_not_admin(api_client: TestClient, api_routes: AppRoutes, user_token):
update_data = {"id": 1, "fullName": "Updated Name", "email": "changeme@email.com", "group": "Home", "admin": True}
response = api_client.put(api_routes.users_id(1), headers=user_token, json=update_data)
assert response.status_code == 403
def test_update_self_as_not_admin(api_client: TestClient, api_routes: AppRoutes, user_token):
update_data = {"id": 3, "fullName": "User fullname", "email": "user@email.com", "group": "Home", "admin": False}
response = api_client.put(api_routes.users_id(3), headers=user_token, json=update_data)
assert response.status_code == 200
def test_self_demote_admin(api_client: TestClient, api_routes: AppRoutes, admin_token):
update_data = {"id": 1, "fullName": "Updated Name", "email": "changeme@email.com", "group": "Home", "admin": False}
response = api_client.put(api_routes.users_id(1), headers=admin_token, json=update_data)
assert response.status_code == 403
def test_self_promote_admin(api_client: TestClient, api_routes: AppRoutes, user_token):
update_data = {"id": 3, "fullName": "Updated Name", "email": "user@email.com", "group": "Home", "admin": True}
response = api_client.put(api_routes.users_id(3), headers=user_token, json=update_data)
assert response.status_code == 403
def test_reset_user_password(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.put(api_routes.users_id_reset_password(4), headers=admin_token)
assert response.status_code == 200
@ -106,23 +148,23 @@ def test_reset_user_password(api_client: TestClient, api_routes: AppRoutes, toke
assert response.status_code == 200
def test_delete_user(api_client: TestClient, api_routes: AppRoutes, token):
response = api_client.delete(api_routes.users_id(2), headers=token)
def test_delete_user(api_client: TestClient, api_routes: AppRoutes, admin_token):
response = api_client.delete(api_routes.users_id(2), headers=admin_token)
assert response.status_code == 200
def test_update_user_image(
api_client: TestClient, api_routes: AppRoutes, test_image_jpg: Path, test_image_png: Path, token
api_client: TestClient, api_routes: AppRoutes, test_image_jpg: Path, test_image_png: Path, admin_token
):
response = api_client.post(
api_routes.users_id_image(2), files={"profile_image": test_image_jpg.open("rb")}, headers=token
api_routes.users_id_image(2), files={"profile_image": test_image_jpg.open("rb")}, headers=admin_token
)
assert response.status_code == 200
response = api_client.post(
api_routes.users_id_image(2), files={"profile_image": test_image_png.open("rb")}, headers=token
api_routes.users_id_image(2), files={"profile_image": test_image_png.open("rb")}, headers=admin_token
)
assert response.status_code == 200