17704661380

Committed 14 Sep 2025 01:33AM UTC coverage: 89.585% (-0.1%) from 89.707%

Build # 17704661380

Build Type

Pull #387

github

Committed by

web-flow

Commit Message

Merge 651f7cddd into 51ee0e718

Pull Request Pull Request #387: Implement allauth

Run Details

6752 of 7537 relevant lines covered (89.58%)

0.9 hits per line

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

98.68

back/admin/integrations/views.py

import json
from datetime import timedelta
from urllib.parse import urlparse

import requests
from django.contrib import messages
from django.contrib.messages.views import SuccessMessageMixin
from django.http import Http404, HttpResponseRedirect
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse_lazy
from django.utils import timezone
from django.utils.translation import gettext as _
from django.views.generic import View
from django.views.generic.base import RedirectView
from django.views.generic.detail import DetailView
from django.views.generic.edit import CreateView, DeleteView, UpdateView
from django.views.generic.list import ListView

from users.mixins import AdminPermMixin, ManagerPermMixin

from .forms import IntegrationExtraArgsForm, IntegrationForm
from .models import Integration, IntegrationTracker


class IntegrationCreateView(AdminPermMixin, CreateView, SuccessMessageMixin):
    template_name = "token_create.html"
    form_class = IntegrationForm
    success_message = _("Integration has been created!")
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Add new integration")
        context["subtitle"] = _("settings")
        context["button_text"] = _("Create")
        return context

    def form_valid(self, form):
        form.instance.integration = Integration.Type.CUSTOM
        return super().form_valid(form)


class IntegrationUpdateView(AdminPermMixin, UpdateView, SuccessMessageMixin):
    template_name = "token_create.html"
    form_class = IntegrationForm
    queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)
    success_message = _("Integration has been updated!")
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Update existing integration")
        context["subtitle"] = _("settings")
        context["button_text"] = _("Update")
        return context

    def form_valid(self, form):
        new_initial_data = form.cleaned_data["manifest"].get("initial_data_form", [])
        old_initial_data = self.get_object().manifest.get("initial_data_form", [])

        # remove keys that don't exist anymore from saved secrets
        new_initial_data_keys = [item["id"] for item in new_initial_data]
        for item in old_initial_data:
            if item["id"] not in new_initial_data_keys:
                form.instance.extra_args.pop(item["id"], None)

        return super().form_valid(form)


class IntegrationDeleteView(AdminPermMixin, DeleteView):
    """This is a general delete function for all integrations"""

    template_name = "integration-delete.html"
    model = Integration
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Delete integration")
        context["subtitle"] = _("settings")
        return context


class IntegrationUpdateExtraArgsView(AdminPermMixin, UpdateView, SuccessMessageMixin):
    template_name = "update_initial_data_form.html"
    form_class = IntegrationExtraArgsForm
    queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)
    success_message = _("Your config values have been updated!")
    success_url = reverse_lazy("settings:integrations")

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("Integration settings")
        context["subtitle"] = _("settings")
        context["button_text"] = _("Update")
        return context


class IntegrationDeleteExtraArgsView(AdminPermMixin, DeleteView, SuccessMessageMixin):
    template_name = "update_initial_data_form.html"
    queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)
    success_message = _("Secret value has been removed")
    success_url = reverse_lazy("settings:integrations")

    def form_valid(self, form):
        self.object = self.get_object()

        secret_value = self.kwargs.get("secret")
        if secret_value not in [
            item["id"] for item in self.object.filled_secret_values
        ]:
            raise Http404

        self.object.extra_args.pop(secret_value)
        self.object.save()
        success_url = reverse_lazy("integrations:update-creds", args=[self.object.pk])
        return HttpResponseRedirect(success_url)


class IntegrationOauthRedirectView(RedirectView):
    permanent = False

    def get_redirect_url(self, pk, *args, **kwargs):
        integration = get_object_or_404(
            Integration,
            pk=pk,
            manifest__oauth__isnull=False,
            enabled_oauth=False,
        )
        return integration._replace_vars(
            integration.manifest["oauth"]["authenticate_url"]
        )


class IntegrationOauthCallbackView(RedirectView):
    permanent = False

    def get_redirect_url(self, pk, *args, **kwargs):
        integration = get_object_or_404(
            Integration,
            pk=pk,
            manifest__oauth__isnull=False,
            enabled_oauth=False,
        )
        code = self.request.GET.get("code", "")
        if code == "" and not integration.manifest["oauth"].get("without_code", False):
            messages.error(self.request, "Code was not provided")
            return reverse_lazy("settings:integrations")

        # Check if url has parameters already
        access_obj = integration.manifest["oauth"]["access_token"]
        if not integration.manifest["oauth"].get("without_code", False):
            parsed_url = urlparse(access_obj["url"])
            if len(parsed_url.query):
                access_obj["url"] += "&code=" + code
            else:
                access_obj["url"] += "?code=" + code

        success, response = integration.run_request(access_obj)

        if not success:
            messages.error(self.request, f"Couldn't save token: {response}")
            return reverse_lazy("settings:integrations")

        integration.extra_args["oauth"] = response.json()
        if "expires_in" in response.json():
            integration.expiring = timezone.now() + timedelta(
                seconds=response.json()["expires_in"]
            )

        integration.enabled_oauth = True
        integration.save(update_fields=["enabled_oauth", "extra_args", "expiring"])

        return reverse_lazy("settings:integrations")


class SlackOAuthView(View):
    def get(self, request):
        access_token, _dummy = Integration.objects.get_or_create(
            integration=Integration.Type.SLACK_BOT
        )
        if "code" not in request.GET:
            messages.error(
                request,
                _("Could not optain slack authentication code."),
            )
            return redirect("settings:integrations")
        code = request.GET["code"]
        params = {
            "code": code,
            "client_id": access_token.client_id,
            "client_secret": access_token.client_secret,
            "redirect_uri": access_token.redirect_url,
        }
        url = "https://slack.com/api/oauth.v2.access"
        json_response = requests.get(url, params)
        data = json.loads(json_response.text)
        if data["ok"]:
            access_token.bot_token = data["access_token"]
            access_token.bot_id = data["bot_user_id"]
            access_token.token = data["access_token"]
            access_token.save()
            messages.success(
                request,
                _(
                    "Slack has successfully been connected. You have a new bot in your "
                    "workspace."
                ),
            )
        else:
            messages.error(request, _("Could not get tokens from Slack"))
        return redirect("settings:integrations")


class IntegrationTrackerListView(ManagerPermMixin, ListView):
    queryset = (
        IntegrationTracker.objects.all()
        .select_related("integration", "for_user")
        .filter(integration__is_active=True)
        .order_by("-ran_at")
    )
    template_name = "tracker_list.html"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("All integration runs")
        context["subtitle"] = _("integrations")
        return context


class IntegrationTrackerDetailView(ManagerPermMixin, DetailView):
    model = IntegrationTracker
    template_name = "tracker.html"

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["title"] = _("%(integration)s for %(user)s") % {
            "integration": (
                self.object.integration.name
                if self.object.integration is not None
                else "Test integration"
            ),
            "user": self.object.for_user,
        }
        context["subtitle"] = _("integrations")
        return context

1	import json	1✔
2	from datetime import timedelta	1✔
3	from urllib.parse import urlparse	1✔
4
5	import requests	1✔
6	from django.contrib import messages	1✔
7	from django.contrib.messages.views import SuccessMessageMixin	1✔
8	from django.http import Http404, HttpResponseRedirect	1✔
9	from django.shortcuts import get_object_or_404, redirect	1✔
10	from django.urls import reverse_lazy	1✔
11	from django.utils import timezone	1✔
12	from django.utils.translation import gettext as _	1✔
13	from django.views.generic import View	1✔
14	from django.views.generic.base import RedirectView	1✔
15	from django.views.generic.detail import DetailView	1✔
16	from django.views.generic.edit import CreateView, DeleteView, UpdateView	1✔
17	from django.views.generic.list import ListView	1✔
18
19	from users.mixins import AdminPermMixin, ManagerPermMixin	1✔
20
21	from .forms import IntegrationExtraArgsForm, IntegrationForm	1✔
22	from .models import Integration, IntegrationTracker	1✔
23
24
25	class IntegrationCreateView(AdminPermMixin, CreateView, SuccessMessageMixin):	1✔
26	template_name = "token_create.html"	1✔
27	form_class = IntegrationForm	1✔
28	success_message = _("Integration has been created!")	1✔
29	success_url = reverse_lazy("settings:integrations")	1✔
30
31	def get_context_data(self, **kwargs):	1✔
32	context = super().get_context_data(**kwargs)	1✔
33	context["title"] = _("Add new integration")	1✔
34	context["subtitle"] = _("settings")	1✔
35	context["button_text"] = _("Create")	1✔
36	return context	1✔
37
38	def form_valid(self, form):	1✔
39	form.instance.integration = Integration.Type.CUSTOM	1✔
40	return super().form_valid(form)	1✔
41
42
43	class IntegrationUpdateView(AdminPermMixin, UpdateView, SuccessMessageMixin):	1✔
44	template_name = "token_create.html"	1✔
45	form_class = IntegrationForm	1✔
46	queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)	1✔
47	success_message = _("Integration has been updated!")	1✔
48	success_url = reverse_lazy("settings:integrations")	1✔
49
50	def get_context_data(self, **kwargs):	1✔
51	context = super().get_context_data(**kwargs)	1✔
52	context["title"] = _("Update existing integration")	1✔
53	context["subtitle"] = _("settings")	1✔
54	context["button_text"] = _("Update")	1✔
55	return context	1✔
56
57	def form_valid(self, form):	1✔
58	new_initial_data = form.cleaned_data["manifest"].get("initial_data_form", [])	1✔
59	old_initial_data = self.get_object().manifest.get("initial_data_form", [])	1✔
60
61	# remove keys that don't exist anymore from saved secrets
62	new_initial_data_keys = [item["id"] for item in new_initial_data]	1✔
63	for item in old_initial_data:	1✔
64	if item["id"] not in new_initial_data_keys:	1✔
65	form.instance.extra_args.pop(item["id"], None)	1✔
66
67	return super().form_valid(form)	1✔
68
69
70	class IntegrationDeleteView(AdminPermMixin, DeleteView):	1✔
71	"""This is a general delete function for all integrations"""
72
73	template_name = "integration-delete.html"	1✔
74	model = Integration	1✔
75	success_url = reverse_lazy("settings:integrations")	1✔
76
77	def get_context_data(self, **kwargs):	1✔
78	context = super().get_context_data(**kwargs)	1✔
79	context["title"] = _("Delete integration")	1✔
80	context["subtitle"] = _("settings")	1✔
81	return context	1✔
82
83
84	class IntegrationUpdateExtraArgsView(AdminPermMixin, UpdateView, SuccessMessageMixin):	1✔
85	template_name = "update_initial_data_form.html"	1✔
86	form_class = IntegrationExtraArgsForm	1✔
87	queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)	1✔
88	success_message = _("Your config values have been updated!")	1✔
89	success_url = reverse_lazy("settings:integrations")	1✔
90
91	def get_context_data(self, **kwargs):	1✔
92	context = super().get_context_data(**kwargs)	1✔
93	context["title"] = _("Integration settings")	1✔
94	context["subtitle"] = _("settings")	1✔
95	context["button_text"] = _("Update")	1✔
96	return context	1✔
97
98
99	class IntegrationDeleteExtraArgsView(AdminPermMixin, DeleteView, SuccessMessageMixin):	1✔
100	template_name = "update_initial_data_form.html"	1✔
101	queryset = Integration.objects.filter(integration=Integration.Type.CUSTOM)	1✔
102	success_message = _("Secret value has been removed")	1✔
103	success_url = reverse_lazy("settings:integrations")	1✔
104
105	def form_valid(self, form):	1✔
106	self.object = self.get_object()	1✔
107
108	secret_value = self.kwargs.get("secret")	1✔
109	if secret_value not in [	1✔
110	item["id"] for item in self.object.filled_secret_values
111	]:
112	raise Http404	1✔
113
114	self.object.extra_args.pop(secret_value)	1✔
115	self.object.save()	1✔
116	success_url = reverse_lazy("integrations:update-creds", args=[self.object.pk])	1✔
117	return HttpResponseRedirect(success_url)	1✔
118
119
120	class IntegrationOauthRedirectView(RedirectView):	1✔
121	permanent = False	1✔
122
123	def get_redirect_url(self, pk, args, *kwargs):	1✔
124	integration = get_object_or_404(	1✔
125	Integration,
126	pk=pk,
127	manifest__oauth__isnull=False,
128	enabled_oauth=False,
129	)
130	return integration._replace_vars(	1✔
131	integration.manifest["oauth"]["authenticate_url"]
132	)
133
134
135	class IntegrationOauthCallbackView(RedirectView):	1✔
136	permanent = False	1✔
137
138	def get_redirect_url(self, pk, args, *kwargs):	1✔
139	integration = get_object_or_404(	1✔
140	Integration,
141	pk=pk,
142	manifest__oauth__isnull=False,
143	enabled_oauth=False,
144	)
145	code = self.request.GET.get("code", "")	1✔
146	if code == "" and not integration.manifest["oauth"].get("without_code", False):	1✔
147	messages.error(self.request, "Code was not provided")	1✔
148	return reverse_lazy("settings:integrations")	1✔
149
150	# Check if url has parameters already
151	access_obj = integration.manifest["oauth"]["access_token"]	1✔
152	if not integration.manifest["oauth"].get("without_code", False):	1✔
153	parsed_url = urlparse(access_obj["url"])	1✔
154	if len(parsed_url.query):	1✔
155	access_obj["url"] += "&code=" + code	1✔
156	else:
157	access_obj["url"] += "?code=" + code	1✔
158
159	success, response = integration.run_request(access_obj)	1✔
160
161	if not success:	1✔
162	messages.error(self.request, f"Couldn't save token: {response}")	1✔
163	return reverse_lazy("settings:integrations")	1✔
164
165	integration.extra_args["oauth"] = response.json()	1✔
166	if "expires_in" in response.json():	1✔
167	integration.expiring = timezone.now() + timedelta(	×
168	seconds=response.json()["expires_in"]
169	)
170
171	integration.enabled_oauth = True	1✔
172	integration.save(update_fields=["enabled_oauth", "extra_args", "expiring"])	1✔
173
174	return reverse_lazy("settings:integrations")	1✔
175
176
177	class SlackOAuthView(View):	1✔
178	def get(self, request):	1✔
179	access_token, _dummy = Integration.objects.get_or_create(	1✔
180	integration=Integration.Type.SLACK_BOT
181	)
182	if "code" not in request.GET:	1✔
183	messages.error(	1✔
184	request,
185	_("Could not optain slack authentication code."),
186	)
187	return redirect("settings:integrations")	1✔
188	code = request.GET["code"]	1✔
189	params = {	1✔
190	"code": code,
191	"client_id": access_token.client_id,
192	"client_secret": access_token.client_secret,
193	"redirect_uri": access_token.redirect_url,
194	}
195	url = "https://slack.com/api/oauth.v2.access"	1✔
196	json_response = requests.get(url, params)	1✔
197	data = json.loads(json_response.text)	1✔
198	if data["ok"]:	1✔
199	access_token.bot_token = data["access_token"]	1✔
200	access_token.bot_id = data["bot_user_id"]	1✔
201	access_token.token = data["access_token"]	1✔
202	access_token.save()	1✔
203	messages.success(	1✔
204	request,
205	_(
206	"Slack has successfully been connected. You have a new bot in your "
207	"workspace."
208	),
209	)
210	else:
211	messages.error(request, _("Could not get tokens from Slack"))	×
212	return redirect("settings:integrations")	1✔
213
214
215	class IntegrationTrackerListView(ManagerPermMixin, ListView):	1✔
216	queryset = (	1✔
217	IntegrationTracker.objects.all()
218	.select_related("integration", "for_user")
219	.filter(integration__is_active=True)
220	.order_by("-ran_at")
221	)
222	template_name = "tracker_list.html"	1✔
223
224	def get_context_data(self, **kwargs):	1✔
225	context = super().get_context_data(**kwargs)	1✔
226	context["title"] = _("All integration runs")	1✔
227	context["subtitle"] = _("integrations")	1✔
228	return context	1✔
229
230
231	class IntegrationTrackerDetailView(ManagerPermMixin, DetailView):	1✔
232	model = IntegrationTracker	1✔
233	template_name = "tracker.html"	1✔
234
235	def get_context_data(self, **kwargs):	1✔
236	context = super().get_context_data(**kwargs)	1✔
237	context["title"] = _("%(integration)s for %(user)s") % {	1✔
238	"integration": (
239	self.object.integration.name
240	if self.object.integration is not None
241	else "Test integration"
242	),
243	"user": self.object.for_user,
244	}
245	context["subtitle"] = _("integrations")	1✔
246	return context	1✔

chiefonboarding / ChiefOnboarding / 17704661380

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