18481185920

Committed 13 Oct 2025 11:53PM UTC coverage: 89.613% (-0.006%) from 89.619%

Build # 18481185920

Build Type

Pull #572

github

Committed by

web-flow

Commit Message

Merge f5b8970dc into f01e5ecbf

Pull Request Pull Request #572: Add permissions based on department

Run Details

7057 of 7875 relevant lines covered (89.61%)

0.9 hits per line

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

98.69

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 misc.mixins import FormWithUserContextMixin
from users.mixins import AdminOrManagerPermMixin, AdminPermMixin

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


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

chiefonboarding / ChiefOnboarding / 18481185920

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