11444579728

Committed 21 Oct 2024 04:32PM UTC coverage: 85.325% (+0.002%) from 85.323%

Build # 11444579728

Build Type

Pull #1390

github

Committed by

web-flow

Commit Message

Merge 37b081ff4 into 6f44ce817

Pull Request Pull Request #1390: filter external events for future

Run Details

2944 of 3448 branches covered (85.38%)

Branch coverage included in aggregate %.

4 of 4 new or added lines in 2 files covered. (100.0%)

24 existing lines in 1 file now uncovered.

11714 of 13731 relevant lines covered (85.31%)

0.85 hits per line

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

74.1

/ephios/plugins/federation/views/frontend.py

from datetime import datetime
from urllib.parse import urljoin

import requests
from django.contrib import messages
from django.contrib.messages.views import SuccessMessageMixin
from django.db import transaction
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse, reverse_lazy
from django.utils.translation import gettext_lazy as _
from django.views.generic import CreateView, DeleteView, DetailView, FormView, TemplateView
from dynamic_preferences.registries import global_preferences_registry
from guardian.mixins import LoginRequiredMixin
from requests import HTTPError, JSONDecodeError, ReadTimeout
from rest_framework.exceptions import PermissionDenied

from ephios.api.models import Application
from ephios.core.models import Event
from ephios.core.views.signup import BaseShiftActionView
from ephios.extra.mixins import StaffRequiredMixin
from ephios.plugins.federation.forms import InviteCodeForm, RedeemInviteCodeForm
from ephios.plugins.federation.models import (
    FederatedEventShare,
    FederatedGuest,
    FederatedHost,
    FederatedUser,
    InviteCode,
)
from ephios.plugins.federation.views.api import FederationOAuthView


class ExternalEventListView(LoginRequiredMixin, TemplateView):
    """
    View that lists all events that are shared with this instance.
    """

    template_name = "federation/external_event_list.html"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        events = []
        for host in FederatedHost.objects.all():
            try:
                r = requests.get(
                    urljoin(host.url, reverse("federation:shared_event_list_view")),
                    headers={"Authorization": f"Bearer {host.access_token}"},
                    params={"shifts__end_time__gte": datetime.now().isoformat()},
                    timeout=5,
                )
                r.raise_for_status()
                results = r.json()["results"]
                for event in results:
                    event["type"] = event["type"]["title"]
                    event["start_time"] = datetime.fromisoformat(event["start_time"])
                    event["end_time"] = datetime.fromisoformat(event["end_time"])
                    event["host"] = host.name
                events += results
            except HTTPError as exc:
                if exc.response.status_code == 403:
                    # the host does not accept our token anymore, so the share needs to be set up again
                    host.oauth_application.delete()
                    host.delete()
            except (JSONDecodeError, ReadTimeout):
                continue
        events.sort(key=lambda e: e["start_time"])
        context["events"] = events
        return context


class CheckFederatedAccessTokenMixin:
    def dispatch(self, request, *args, **kwargs):
        if "federation_access_token" not in request.session.keys():
            return FederationOAuthView.as_view()(request, *args, **kwargs)
        return super().dispatch(request, *args, **kwargs)

    def get_context_data(self, object):
        context = super().get_context_data()
        context["federation_guest"] = FederatedGuest.objects.get(
            pk=self.request.session["federation_guest_pk"]
        )
        return context


class FederatedEventDetailView(CheckFederatedAccessTokenMixin, DetailView):
    """
    View that displays a shared event to a federated user from another instance
    """

    model = Event
    template_name = "federation/event_detail.html"

    def get_object(self, queryset=None):
        obj = super().get_object(queryset)
        try:
            guest = self.request.session["federation_guest_pk"]
            FederatedEventShare.objects.get(
                event=obj,
                shared_with__in=[FederatedGuest.objects.get(pk=guest)],
            )
        except (KeyError, FederatedEventShare.DoesNotExist, FederatedGuest.DoesNotExist) as exc:
            self.request.session.flush()
            raise PermissionDenied from exc
        return obj


class FederatedUserShiftActionView(CheckFederatedAccessTokenMixin, BaseShiftActionView):
    """
    View that allows a federated user from another instanceto sign up for a shift
    """

    def get_participant(self):
        try:
            return FederatedUser.objects.get(
                pk=self.request.session["federated_user"]
            ).as_participant()
        except FederatedUser.DoesNotExist as e:
            raise PermissionDenied from e


class FederationSettingsView(StaffRequiredMixin, TemplateView):
    """
    View that displays the federation settings page where new instances can be connected
    """

    template_name = "federation/federation_settings.html"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["federation_guests"] = FederatedGuest.objects.all()
        context["federation_hosts"] = FederatedHost.objects.all()
        context["federation_invites"] = InviteCode.objects.all()
        return context


class CreateInviteCodeView(StaffRequiredMixin, CreateView):
    """
    View that allows staff users to create new invite codes (to share events with another instance)
    """

    model = InviteCode
    form_class = InviteCodeForm

    def get_success_url(self):
        return reverse("federation:reveal_invite_code", kwargs={"pk": self.object.pk})


class InviteCodeRevealView(StaffRequiredMixin, TemplateView):
    """
    View that displays an invite code to a staff user
    """

    template_name = "federation/invitecode_reveal.html"

    def get(self, request, *args, **kwargs):
        invite = get_object_or_404(InviteCode, pk=kwargs["pk"])
        if invite.is_expired:
            messages.error(request, _("Invite code has expired."))
            return redirect("federation:settings")
        context = self.get_context_data(invite=invite, **kwargs)
        return self.render_to_response(context)


class RedeemInviteCodeView(StaffRequiredMixin, FormView):
    """
    View that allows staff users to redeem an invite code (to receive events from another instance)
    """

    form_class = RedeemInviteCodeForm
    template_name = "federation/redeem_invite_code.html"

    def get_form_kwargs(self):
        kwargs = super().get_form_kwargs()
        if "code" in self.request.GET:
            kwargs["initial"] = {"code": self.request.GET["code"]}
        return kwargs

    @transaction.atomic
    def _setup_federated_host(self, data):
        oauth_application = Application(
            user=self.request.user,
            client_type=Application.CLIENT_CONFIDENTIAL,
            authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
            redirect_uris=urljoin(data["host_url"], reverse("federation:oauth_callback")),
        )
        response = requests.post(
            urljoin(data["host_url"], reverse("federation:redeem_invite_code")),
            data={
                "name": global_preferences_registry.manager()["general__organization_name"],
                "url": data["guest_url"],
                "client_id": oauth_application.client_id,
                "client_secret": oauth_application.client_secret,
                "code": data["code"],
            },
            timeout=10,
        )
        response.raise_for_status()
        response_data = response.json()
        oauth_application.name = response_data["host_name"]
        oauth_application.save()
        FederatedHost.objects.create(
            name=response_data["host_name"],
            url=data["host_url"],
            access_token=response_data["access_token"],
            oauth_application=oauth_application,
        )

    def form_valid(self, form):
        try:
            self._setup_federated_host(form.code_data)
        except (KeyError, JSONDecodeError):
            messages.error(
                self.request,
                _("Invalid code data. Please try again with a new invite code."),
            )
        except (HTTPError, ReadTimeout):
            messages.error(
                self.request,
                _("Could not connect to host instance. Please try again later."),
            )
        else:
            messages.success(
                self.request,
                _(
                    "Invite code redeemded successfully. You are now receiving events from this instance."
                ),
            )
            return redirect(reverse("federation:settings"))
        return self.form_invalid(form)


class FederatedGuestDeleteView(StaffRequiredMixin, SuccessMessageMixin, DeleteView):
    """
    View that allows staff users to remove a guest instance (to stop sharing events with another instance)
    """

    model = FederatedGuest
    success_url = reverse_lazy("federation:settings")
    success_message = _("You are no longer sharing events with this instance.")

    def form_valid(self, form):
        access_token = self.object.access_token
        response = super().form_valid(form)
        access_token.delete()
        return response


class FederatedHostDeleteView(StaffRequiredMixin, SuccessMessageMixin, DeleteView):
    """
    View that allows staff users to remove a host instance (to stop receiving events from another instance)
    """

    model = FederatedHost
    success_url = reverse_lazy("federation:settings")
    success_message = _("You are no longer receiving events from this instance.")

    def form_valid(self, form):
        oauth_app = self.object.oauth_application
        guest_response = requests.delete(
            urljoin(self.object.url, reverse("federation:api_delete_guest")),
            headers={"Authorization": f"Bearer {self.object.access_token}"},
            timeout=5,
        )
        if guest_response.status_code != 204:
            messages.error(
                self.request,
                _("Failed to remove this instance. Please try again later."),
            )
            return redirect("federation:settings")
        response = super().form_valid(form)
        oauth_app.delete()
        return response

1	from datetime import datetime	1✔
2	from urllib.parse import urljoin	1✔
3
4	import requests	1✔
5	from django.contrib import messages	1✔
6	from django.contrib.messages.views import SuccessMessageMixin	1✔
7	from django.db import transaction	1✔
8	from django.shortcuts import get_object_or_404, redirect	1✔
9	from django.urls import reverse, reverse_lazy	1✔
10	from django.utils.translation import gettext_lazy as _	1✔
11	from django.views.generic import CreateView, DeleteView, DetailView, FormView, TemplateView	1✔
12	from dynamic_preferences.registries import global_preferences_registry	1✔
13	from guardian.mixins import LoginRequiredMixin	1✔
14	from requests import HTTPError, JSONDecodeError, ReadTimeout	1✔
15	from rest_framework.exceptions import PermissionDenied	1✔
16
17	from ephios.api.models import Application	1✔
18	from ephios.core.models import Event	1✔
19	from ephios.core.views.signup import BaseShiftActionView	1✔
20	from ephios.extra.mixins import StaffRequiredMixin	1✔
21	from ephios.plugins.federation.forms import InviteCodeForm, RedeemInviteCodeForm	1✔
22	from ephios.plugins.federation.models import (	1✔
23	FederatedEventShare,
24	FederatedGuest,
25	FederatedHost,
26	FederatedUser,
27	InviteCode,
28	)
29	from ephios.plugins.federation.views.api import FederationOAuthView	1✔
30
31
32	class ExternalEventListView(LoginRequiredMixin, TemplateView):	1✔
33	"""
34	View that lists all events that are shared with this instance.
35	"""
36
37	template_name = "federation/external_event_list.html"	1✔
38
39	def get_context_data(self, **kwargs):	1✔
40	context = super().get_context_data(**kwargs)	1✔
41	events = []	1✔
42	for host in FederatedHost.objects.all():	1✔
43	try:	1✔
44	r = requests.get(	1✔
45	urljoin(host.url, reverse("federation:shared_event_list_view")),
46	headers={"Authorization": f"Bearer {host.access_token}"},
47	params={"shifts__end_time__gte": datetime.now().isoformat()},
48	timeout=5,
49	)
50	r.raise_for_status()	1✔
51	results = r.json()["results"]	1✔
52	for event in results:	1✔
53	event["type"] = event["type"]["title"]	1✔
54	event["start_time"] = datetime.fromisoformat(event["start_time"])	1✔
55	event["end_time"] = datetime.fromisoformat(event["end_time"])	1✔
56	event["host"] = host.name	1✔
57	events += results	1✔
UNCOV 58	except HTTPError as exc:	×
UNCOV 59	if exc.response.status_code == 403:	×
60	# the host does not accept our token anymore, so the share needs to be set up again
61	host.oauth_application.delete()	×
UNCOV 62	host.delete()	×
63	except (JSONDecodeError, ReadTimeout):	×
64	continue	×
65	events.sort(key=lambda e: e["start_time"])	1✔
66	context["events"] = events	1✔
67	return context	1✔
68
69
70	class CheckFederatedAccessTokenMixin:	1✔
71	def dispatch(self, request, args, *kwargs):	1✔
72	if "federation_access_token" not in request.session.keys():	1!
UNCOV 73	return FederationOAuthView.as_view()(request, args, *kwargs)	×
74	return super().dispatch(request, args, *kwargs)	1✔
75
76	def get_context_data(self, object):	1✔
77	context = super().get_context_data()	1✔
78	context["federation_guest"] = FederatedGuest.objects.get(	1✔
79	pk=self.request.session["federation_guest_pk"]
80	)
81	return context	1✔
82
83
84	class FederatedEventDetailView(CheckFederatedAccessTokenMixin, DetailView):	1✔
85	"""
86	View that displays a shared event to a federated user from another instance
87	"""
88
89	model = Event	1✔
90	template_name = "federation/event_detail.html"	1✔
91
92	def get_object(self, queryset=None):	1✔
93	obj = super().get_object(queryset)	1✔
94	try:	1✔
95	guest = self.request.session["federation_guest_pk"]	1✔
96	FederatedEventShare.objects.get(	1✔
97	event=obj,
98	shared_with__in=[FederatedGuest.objects.get(pk=guest)],
99	)
UNCOV 100	except (KeyError, FederatedEventShare.DoesNotExist, FederatedGuest.DoesNotExist) as exc:	×
UNCOV 101	self.request.session.flush()	×
102	raise PermissionDenied from exc	×
103	return obj	1✔
104
105
106	class FederatedUserShiftActionView(CheckFederatedAccessTokenMixin, BaseShiftActionView):	1✔
107	"""
108	View that allows a federated user from another instanceto sign up for a shift
109	"""
110
111	def get_participant(self):	1✔
112	try:	1✔
113	return FederatedUser.objects.get(	1✔
114	pk=self.request.session["federated_user"]
115	).as_participant()
UNCOV 116	except FederatedUser.DoesNotExist as e:	×
UNCOV 117	raise PermissionDenied from e	×
118
119
120	class FederationSettingsView(StaffRequiredMixin, TemplateView):	1✔
121	"""
122	View that displays the federation settings page where new instances can be connected
123	"""
124
125	template_name = "federation/federation_settings.html"	1✔
126
127	def get_context_data(self, **kwargs):	1✔
128	context = super().get_context_data(**kwargs)	1✔
129	context["federation_guests"] = FederatedGuest.objects.all()	1✔
130	context["federation_hosts"] = FederatedHost.objects.all()	1✔
131	context["federation_invites"] = InviteCode.objects.all()	1✔
132	return context	1✔
133
134
135	class CreateInviteCodeView(StaffRequiredMixin, CreateView):	1✔
136	"""
137	View that allows staff users to create new invite codes (to share events with another instance)
138	"""
139
140	model = InviteCode	1✔
141	form_class = InviteCodeForm	1✔
142
143	def get_success_url(self):	1✔
144	return reverse("federation:reveal_invite_code", kwargs={"pk": self.object.pk})	1✔
145
146
147	class InviteCodeRevealView(StaffRequiredMixin, TemplateView):	1✔
148	"""
149	View that displays an invite code to a staff user
150	"""
151
152	template_name = "federation/invitecode_reveal.html"	1✔
153
154	def get(self, request, args, *kwargs):	1✔
155	invite = get_object_or_404(InviteCode, pk=kwargs["pk"])	1✔
156	if invite.is_expired:	1!
UNCOV 157	messages.error(request, _("Invite code has expired."))	×
UNCOV 158	return redirect("federation:settings")	×
159	context = self.get_context_data(invite=invite, **kwargs)	1✔
160	return self.render_to_response(context)	1✔
161
162
163	class RedeemInviteCodeView(StaffRequiredMixin, FormView):	1✔
164	"""
165	View that allows staff users to redeem an invite code (to receive events from another instance)
166	"""
167
168	form_class = RedeemInviteCodeForm	1✔
169	template_name = "federation/redeem_invite_code.html"	1✔
170
171	def get_form_kwargs(self):	1✔
172	kwargs = super().get_form_kwargs()	1✔
173	if "code" in self.request.GET:	1!
UNCOV 174	kwargs["initial"] = {"code": self.request.GET["code"]}	×
175	return kwargs	1✔
176
177	@transaction.atomic	1✔
178	def _setup_federated_host(self, data):	1✔
179	oauth_application = Application(	1✔
180	user=self.request.user,
181	client_type=Application.CLIENT_CONFIDENTIAL,
182	authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE,
183	redirect_uris=urljoin(data["host_url"], reverse("federation:oauth_callback")),
184	)
185	response = requests.post(	1✔
186	urljoin(data["host_url"], reverse("federation:redeem_invite_code")),
187	data={
188	"name": global_preferences_registry.manager()["general__organization_name"],
189	"url": data["guest_url"],
190	"client_id": oauth_application.client_id,
191	"client_secret": oauth_application.client_secret,
192	"code": data["code"],
193	},
194	timeout=10,
195	)
196	response.raise_for_status()	1✔
197	response_data = response.json()	1✔
198	oauth_application.name = response_data["host_name"]	1✔
199	oauth_application.save()	1✔
200	FederatedHost.objects.create(	1✔
201	name=response_data["host_name"],
202	url=data["host_url"],
203	access_token=response_data["access_token"],
204	oauth_application=oauth_application,
205	)
206
207	def form_valid(self, form):	1✔
208	try:	1✔
209	self._setup_federated_host(form.code_data)	1✔
UNCOV 210	except (KeyError, JSONDecodeError):	×
UNCOV 211	messages.error(	×
212	self.request,
213	_("Invalid code data. Please try again with a new invite code."),
214	)
UNCOV 215	except (HTTPError, ReadTimeout):	×
UNCOV 216	messages.error(	×
217	self.request,
218	_("Could not connect to host instance. Please try again later."),
219	)
220	else:
221	messages.success(	1✔
222	self.request,
223	_(
224	"Invite code redeemded successfully. You are now receiving events from this instance."
225	),
226	)
227	return redirect(reverse("federation:settings"))	1✔
UNCOV 228	return self.form_invalid(form)	×
229
230
231	class FederatedGuestDeleteView(StaffRequiredMixin, SuccessMessageMixin, DeleteView):	1✔
232	"""
233	View that allows staff users to remove a guest instance (to stop sharing events with another instance)
234	"""
235
236	model = FederatedGuest	1✔
237	success_url = reverse_lazy("federation:settings")	1✔
238	success_message = _("You are no longer sharing events with this instance.")	1✔
239
240	def form_valid(self, form):	1✔
UNCOV 241	access_token = self.object.access_token	×
UNCOV 242	response = super().form_valid(form)	×
243	access_token.delete()	×
244	return response	×
245
246
247	class FederatedHostDeleteView(StaffRequiredMixin, SuccessMessageMixin, DeleteView):	1✔
248	"""
249	View that allows staff users to remove a host instance (to stop receiving events from another instance)
250	"""
251
252	model = FederatedHost	1✔
253	success_url = reverse_lazy("federation:settings")	1✔
254	success_message = _("You are no longer receiving events from this instance.")	1✔
255
256	def form_valid(self, form):	1✔
UNCOV 257	oauth_app = self.object.oauth_application	×
UNCOV 258	guest_response = requests.delete(	×
259	urljoin(self.object.url, reverse("federation:api_delete_guest")),
260	headers={"Authorization": f"Bearer {self.object.access_token}"},
261	timeout=5,
262	)
UNCOV 263	if guest_response.status_code != 204:	×
UNCOV 264	messages.error(	×
265	self.request,
266	_("Failed to remove this instance. Please try again later."),
267	)
UNCOV 268	return redirect("federation:settings")	×
UNCOV 269	response = super().form_valid(form)	×
270	oauth_app.delete()	×
271	return response	×

ephios-dev / ephios / 11444579728

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