etesync-server/etebase_fastapi/test_reset_view.py

39 lines
1.6 KiB
Python
Raw Normal View History

2020-12-25 12:06:35 +00:00
from django.conf import settings
from django.contrib.auth import get_user_model
from django.db import transaction
from django.shortcuts import get_object_or_404
2020-12-28 14:57:09 +00:00
from fastapi import APIRouter, Request, status
2020-12-25 12:06:35 +00:00
2020-12-27 21:06:25 +00:00
from django_etebase.utils import get_user_queryset, CallbackContext
2020-12-25 15:22:14 +00:00
from etebase_fastapi.authentication import SignupIn, signup_save
2020-12-25 12:06:35 +00:00
from etebase_fastapi.msgpack import MsgpackRoute
2020-12-28 14:57:09 +00:00
from etebase_fastapi.exceptions import HttpError
2020-12-25 12:06:35 +00:00
2020-12-28 07:04:45 +00:00
test_reset_view_router = APIRouter(route_class=MsgpackRoute, tags=["test helpers"])
2020-12-25 12:06:35 +00:00
User = get_user_model()
2020-12-28 14:57:09 +00:00
@test_reset_view_router.post("/reset/", status_code=status.HTTP_204_NO_CONTENT)
2020-12-27 18:36:11 +00:00
def reset(data: SignupIn, request: Request):
2020-12-25 12:06:35 +00:00
# Only run when in DEBUG mode! It's only used for tests
if not settings.DEBUG:
2020-12-28 14:57:09 +00:00
raise HttpError(code="generic", detail="Only allowed in debug mode.")
2020-12-25 12:06:35 +00:00
with transaction.atomic():
2020-12-27 21:06:25 +00:00
user_queryset = get_user_queryset(User.objects.all(), CallbackContext(request.path_params))
2020-12-25 12:06:35 +00:00
user = get_object_or_404(user_queryset, username=data.user.username)
# Only allow test users for extra safety
if not getattr(user, User.USERNAME_FIELD).startswith("test_user"):
2020-12-28 14:57:09 +00:00
raise HttpError(code="generic", detail="Endpoint not allowed for user.")
2020-12-25 12:06:35 +00:00
if hasattr(user, "userinfo"):
user.userinfo.delete()
2020-12-28 14:57:09 +00:00
2020-12-27 18:36:11 +00:00
signup_save(data, request)
2020-12-25 12:06:35 +00:00
# Delete all of the journal data for this user for a clear test env
user.collection_set.all().delete()
user.collectionmember_set.all().delete()
user.incoming_invitations.all().delete()
# FIXME: also delete chunk files!!!