11506351701

Committed 24 Oct 2024 07:44PM UTC coverage: 85.219% (+0.001%) from 85.218%

Build # 11506351701

Build Type

push

github

Committed by

felixrindt

Commit Message

fixup shift viewset

Run Details

2958 of 3486 branches covered (84.85%)

Branch coverage included in aggregate %.

10 of 12 new or added lines in 2 files covered. (83.33%)

52 existing lines in 3 files now uncovered.

11784 of 13813 relevant lines covered (85.31%)

0.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

57.01

/ephios/plugins/federation/views/api.py

from urllib.parse import urljoin

import django_filters
import requests
from django.conf import settings
from django.core.exceptions import MultipleObjectsReturned
from django.db.models import Max, Min
from django.shortcuts import redirect
from django.urls import reverse
from django.views import View
from oauth2_provider.contrib.rest_framework import TokenHasScope
from oauthlib.oauth2 import WebApplicationClient
from requests_oauthlib import OAuth2Session
from rest_framework.exceptions import PermissionDenied
from rest_framework.generics import CreateAPIView, DestroyAPIView, ListAPIView
from rest_framework.permissions import AllowAny

from ephios.api.filters import EventFilterSet
from ephios.core.models import Event, Qualification
from ephios.plugins.federation.models import FederatedGuest, FederatedUser
from ephios.plugins.federation.serializers import (
    FederatedGuestCreateSerializer,
    SharedEventSerializer,
)


class RedeemInviteCodeView(CreateAPIView):
    """
    API view that accepts an InviteCode and creates a FederatedGuest (to start sharing events with that instance).
    """

    serializer_class = FederatedGuestCreateSerializer
    queryset = FederatedGuest.objects.all()
    authentication_classes = []
    permission_classes = [AllowAny]


class FederatedGuestDeleteView(DestroyAPIView):
    """
    API view that deletes a FederatedGuest (to stop sharing events with that instance).
    """

    queryset = FederatedGuest.objects.all()
    permission_classes = [TokenHasScope]
    required_scopes = []

    def get_object(self):
        try:
            # request.auth is an auth token, federatedguest is the reverse relation
            return self.request.auth.federatedguest
        except FederatedGuest.DoesNotExist as exc:
            raise PermissionDenied from exc


class SharedEventListView(ListAPIView):
    """
    API view that lists all events that are shared with the instance corresponding to the access token.
    """

    serializer_class = SharedEventSerializer
    permission_classes = [TokenHasScope]
    required_scopes = []
    filter_backends = [django_filters.rest_framework.DjangoFilterBackend]
    filterset_class = EventFilterSet

    def get_queryset(self):
        try:
            # request.auth is an auth token, federatedguest is the reverse relation
            guest = self.request.auth.federatedguest
        except FederatedGuest.DoesNotExist as exc:
            raise PermissionDenied from exc
        return (
            Event.objects.filter(federatedeventshare__shared_with=guest)
            .annotate(
                start_time=Min("shifts__start_time"),
                end_time=Max("shifts__end_time"),
            )
            .select_related("type")
            .prefetch_related("shifts")
            .order_by("start_time")
        )

    def get_serializer_context(self):
        context = super().get_serializer_context()
        context["federated_guest"] = self.request.auth.federatedguest
        return context


class FederationOAuthView(View):
    """
    View that handles the OAuth2 flow for federated users from another instance.
    """

    def get(self, request, *args, **kwargs):
        try:
            guest_pk = self.request.session.get("federation_guest_pk", kwargs.get("guest"))
            self.guest = FederatedGuest.objects.get(pk=guest_pk)
        except (KeyError, FederatedGuest.DoesNotExist, MultipleObjectsReturned) as exc:
            raise PermissionDenied from exc
        if "error" in request.GET.keys():
            return redirect(
                urljoin(
                    urljoin(self.guest.url, reverse("federation:external_event_list")),
                    "?error=oauth_error",
                )
            )
        elif "code" in request.GET.keys():
            self._oauth_callback()
            try:
                return redirect(
                    "federation:event_detail",
                    pk=self.request.session["federation_event"],
                    guest=guest_pk,
                )
            except KeyError:
                return redirect(
                    urljoin(
                        urljoin(self.guest.url, reverse("federation:external_event_list")),
                        "?error=event_error",
                    )
                )
        else:
            return redirect(self._get_authorization_url())

    def _get_authorization_url(self):
        oauth_client = WebApplicationClient(client_id=self.guest.client_id)
        oauth = OAuth2Session(
            client=oauth_client,
            redirect_uri=urljoin(settings.GET_SITE_URL(), reverse("federation:oauth_callback")),
            scope=["ME_READ"],
        )
        verifier = oauth_client.create_code_verifier(64)
        self.request.session["code_verifier"] = verifier
        self.request.session["federation_event"] = self.kwargs["pk"]
        self.request.session["federation_guest_pk"] = self.guest.pk
        challenge = oauth_client.create_code_challenge(verifier, "S256")
        authorization_url, _ = oauth.authorization_url(
            urljoin(self.guest.url, "api/oauth/authorize/"),
            code_challenge=challenge,
            code_challenge_method="S256",
        )
        return authorization_url

    def _oauth_callback(self):
        oauth_client = WebApplicationClient(
            client_id=self.guest.client_id, code_verifier=self.request.session["code_verifier"]
        )
        oauth = OAuth2Session(client=oauth_client)
        token = oauth.fetch_token(
            urljoin(self.guest.url, "api/oauth/token/"),
            authorization_response=urljoin(settings.GET_SITE_URL(), self.request.get_full_path()),
            client_secret=self.guest.client_secret,
            code_verifier=self.request.session["code_verifier"],
        )
        self.request.session["federation_access_token"] = token["access_token"]
        self.request.session.set_expiry(token["expires_in"])
        user_data = requests.get(
            urljoin(self.guest.url, "api/users/me/"),
            headers={"Authorization": f"Bearer {token['access_token']}"},
            timeout=5,
        )
        try:
            user = FederatedUser.objects.get(
                federated_instance=self.guest, email=user_data.json()["email"]
            )
        except FederatedUser.DoesNotExist:
            user = self._create_user(user_data.json())
        self.request.session["federated_user"] = user.pk

    def _create_user(self, user_data):
        user = FederatedUser.objects.create(
            federated_instance=self.guest,
            email=user_data["email"],
            display_name=user_data["display_name"],
            date_of_birth=user_data["date_of_birth"],
        )
        for qualification in user_data["qualifications"]:
            try:
                # Note that we assign the qualification on the host instance without further checks.
                # This may lead to incorrect qualifications if the inclusions for the qualification
                # are defined differently on the guest instance. We are accepting this as it should
                # not happen with the pre-defined qualifications as we are displaying a warning if
                # the user adapt these and custom qualifications will have different uuids anyway.
                user.qualifications.add(Qualification.objects.get(uuid=qualification["uuid"]))
            except Qualification.DoesNotExist:
                for included_qualification in qualification["includes"]:
                    try:
                        user.qualifications.add(
                            Qualification.objects.get(uuid=included_qualification)
                        )
                    except Qualification.DoesNotExist:
                        continue
        return user

1	from urllib.parse import urljoin	1✔
2
3	import django_filters	1✔
4	import requests	1✔
5	from django.conf import settings	1✔
6	from django.core.exceptions import MultipleObjectsReturned	1✔
7	from django.db.models import Max, Min	1✔
8	from django.shortcuts import redirect	1✔
9	from django.urls import reverse	1✔
10	from django.views import View	1✔
11	from oauth2_provider.contrib.rest_framework import TokenHasScope	1✔
12	from oauthlib.oauth2 import WebApplicationClient	1✔
13	from requests_oauthlib import OAuth2Session	1✔
14	from rest_framework.exceptions import PermissionDenied	1✔
15	from rest_framework.generics import CreateAPIView, DestroyAPIView, ListAPIView	1✔
16	from rest_framework.permissions import AllowAny	1✔
17
18	from ephios.api.filters import EventFilterSet	1✔
19	from ephios.core.models import Event, Qualification	1✔
20	from ephios.plugins.federation.models import FederatedGuest, FederatedUser	1✔
21	from ephios.plugins.federation.serializers import (	1✔
22	FederatedGuestCreateSerializer,
23	SharedEventSerializer,
24	)
25
26
27	class RedeemInviteCodeView(CreateAPIView):	1✔
28	"""
29	API view that accepts an InviteCode and creates a FederatedGuest (to start sharing events with that instance).
30	"""
31
32	serializer_class = FederatedGuestCreateSerializer	1✔
33	queryset = FederatedGuest.objects.all()	1✔
34	authentication_classes = []	1✔
35	permission_classes = [AllowAny]	1✔
36
37
38	class FederatedGuestDeleteView(DestroyAPIView):	1✔
39	"""
40	API view that deletes a FederatedGuest (to stop sharing events with that instance).
41	"""
42
43	queryset = FederatedGuest.objects.all()	1✔
44	permission_classes = [TokenHasScope]	1✔
45	required_scopes = []	1✔
46
47	def get_object(self):	1✔
48	try:	×
49	# request.auth is an auth token, federatedguest is the reverse relation
50	return self.request.auth.federatedguest	×
UNCOV 51	except FederatedGuest.DoesNotExist as exc:	×
UNCOV 52	raise PermissionDenied from exc	×
53
54
55	class SharedEventListView(ListAPIView):	1✔
56	"""
57	API view that lists all events that are shared with the instance corresponding to the access token.
58	"""
59
60	serializer_class = SharedEventSerializer	1✔
61	permission_classes = [TokenHasScope]	1✔
62	required_scopes = []	1✔
63	filter_backends = [django_filters.rest_framework.DjangoFilterBackend]	1✔
64	filterset_class = EventFilterSet	1✔
65
66	def get_queryset(self):	1✔
67	try:	1✔
68	# request.auth is an auth token, federatedguest is the reverse relation
69	guest = self.request.auth.federatedguest	1✔
UNCOV 70	except FederatedGuest.DoesNotExist as exc:	×
UNCOV 71	raise PermissionDenied from exc	×
72	return (	1✔
73	Event.objects.filter(federatedeventshare__shared_with=guest)
74	.annotate(
75	start_time=Min("shifts__start_time"),
76	end_time=Max("shifts__end_time"),
77	)
78	.select_related("type")
79	.prefetch_related("shifts")
80	.order_by("start_time")
81	)
82
83	def get_serializer_context(self):	1✔
84	context = super().get_serializer_context()	1✔
85	context["federated_guest"] = self.request.auth.federatedguest	1✔
86	return context	1✔
87
88
89	class FederationOAuthView(View):	1✔
90	"""
91	View that handles the OAuth2 flow for federated users from another instance.
92	"""
93
94	def get(self, request, args, *kwargs):	1✔
UNCOV 95	try:	×
UNCOV 96	guest_pk = self.request.session.get("federation_guest_pk", kwargs.get("guest"))	×
UNCOV 97	self.guest = FederatedGuest.objects.get(pk=guest_pk)	×
UNCOV 98	except (KeyError, FederatedGuest.DoesNotExist, MultipleObjectsReturned) as exc:	×
UNCOV 99	raise PermissionDenied from exc	×
100	if "error" in request.GET.keys():	×
101	return redirect(	×
102	urljoin(
103	urljoin(self.guest.url, reverse("federation:external_event_list")),
104	"?error=oauth_error",
105	)
106	)
UNCOV 107	elif "code" in request.GET.keys():	×
108	self._oauth_callback()	×
109	try:	×
UNCOV 110	return redirect(	×
111	"federation:event_detail",
112	pk=self.request.session["federation_event"],
113	guest=guest_pk,
114	)
UNCOV 115	except KeyError:	×
116	return redirect(	×
117	urljoin(
118	urljoin(self.guest.url, reverse("federation:external_event_list")),
119	"?error=event_error",
120	)
121	)
122	else:
UNCOV 123	return redirect(self._get_authorization_url())	×
124
125	def _get_authorization_url(self):	1✔
126	oauth_client = WebApplicationClient(client_id=self.guest.client_id)	×
127	oauth = OAuth2Session(	×
128	client=oauth_client,
129	redirect_uri=urljoin(settings.GET_SITE_URL(), reverse("federation:oauth_callback")),
130	scope=["ME_READ"],
131	)
UNCOV 132	verifier = oauth_client.create_code_verifier(64)	×
UNCOV 133	self.request.session["code_verifier"] = verifier	×
UNCOV 134	self.request.session["federation_event"] = self.kwargs["pk"]	×
135	self.request.session["federation_guest_pk"] = self.guest.pk	×
UNCOV 136	challenge = oauth_client.create_code_challenge(verifier, "S256")	×
UNCOV 137	authorization_url, _ = oauth.authorization_url(	×
138	urljoin(self.guest.url, "api/oauth/authorize/"),
139	code_challenge=challenge,
140	code_challenge_method="S256",
141	)
142	return authorization_url	×
143
144	def _oauth_callback(self):	1✔
UNCOV 145	oauth_client = WebApplicationClient(	×
146	client_id=self.guest.client_id, code_verifier=self.request.session["code_verifier"]
147	)
148	oauth = OAuth2Session(client=oauth_client)	×
149	token = oauth.fetch_token(	×
150	urljoin(self.guest.url, "api/oauth/token/"),
151	authorization_response=urljoin(settings.GET_SITE_URL(), self.request.get_full_path()),
152	client_secret=self.guest.client_secret,
153	code_verifier=self.request.session["code_verifier"],
154	)
155	self.request.session["federation_access_token"] = token["access_token"]	×
156	self.request.session.set_expiry(token["expires_in"])	×
UNCOV 157	user_data = requests.get(	×
158	urljoin(self.guest.url, "api/users/me/"),
159	headers={"Authorization": f"Bearer {token['access_token']}"},
160	timeout=5,
161	)
UNCOV 162	try:	×
UNCOV 163	user = FederatedUser.objects.get(	×
164	federated_instance=self.guest, email=user_data.json()["email"]
165	)
UNCOV 166	except FederatedUser.DoesNotExist:	×
UNCOV 167	user = self._create_user(user_data.json())	×
UNCOV 168	self.request.session["federated_user"] = user.pk	×
169
170	def _create_user(self, user_data):	1✔
171	user = FederatedUser.objects.create(	1✔
172	federated_instance=self.guest,
173	email=user_data["email"],
174	display_name=user_data["display_name"],
175	date_of_birth=user_data["date_of_birth"],
176	)
177	for qualification in user_data["qualifications"]:	1✔
178	try:	1✔
179	# Note that we assign the qualification on the host instance without further checks.
180	# This may lead to incorrect qualifications if the inclusions for the qualification
181	# are defined differently on the guest instance. We are accepting this as it should
182	# not happen with the pre-defined qualifications as we are displaying a warning if
183	# the user adapt these and custom qualifications will have different uuids anyway.
184	user.qualifications.add(Qualification.objects.get(uuid=qualification["uuid"]))	1✔
185	except Qualification.DoesNotExist:	1✔
186	for included_qualification in qualification["includes"]:	1✔
187	try:	1✔
188	user.qualifications.add(	1✔
189	Qualification.objects.get(uuid=included_qualification)
190	)
UNCOV 191	except Qualification.DoesNotExist:	×
UNCOV 192	continue	×
193	return user	1✔

ephios-dev / ephios / 11506351701

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous