Move plivo connect view to channel type

temba/channels/models.py

@@ -253,9 +253,6 @@ class Channel(LegacyUUIDMixin, TembaModel, DependencyMixin):
     CONFIG_USE_NATIONAL = "use_national"
     CONFIG_ENCODING = "encoding"
     CONFIG_PAGE_NAME = "page_name"
-    CONFIG_PLIVO_AUTH_ID = "PLIVO_AUTH_ID"
-    CONFIG_PLIVO_AUTH_TOKEN = "PLIVO_AUTH_TOKEN"
-    CONFIG_PLIVO_APP_ID = "PLIVO_APP_ID"
 
     CONFIG_AUTH_TOKEN = "auth_token"
     CONFIG_SECRET = "secret"

temba/channels/types/plivo/tests.py

@@ -6,6 +6,8 @@
 from temba.tests import MockResponse, TembaTest
 from temba.utils import json
 
+from .type import PlivoType
+
 
 class PlivoTypeTest(TembaTest):
     def test_claim(self):
@@ -14,7 +16,7 @@ def test_claim(self):
         # remove any existing channels
         self.org.channels.update(is_active=False)
 
-        connect_plivo_url = reverse("orgs.org_plivo_connect")
+        connect_plivo_url = reverse("channels.types.plivo.connect")
         claim_plivo_url = reverse("channels.types.plivo.claim")
 
         # make sure plivo is on the claim page
@@ -30,15 +32,15 @@ def test_claim(self):
 
             response = self.client.get(claim_plivo_url, follow=True)
 
-            self.assertEqual(response.request["PATH_INFO"], reverse("orgs.org_plivo_connect"))
+            self.assertEqual(response.request["PATH_INFO"], connect_plivo_url)
 
         with patch("requests.get") as plivo_get:
             plivo_get.return_value = MockResponse(400, json.dumps(dict()))
 
             # try hit the claim page, should be redirected; no credentials in session
             response = self.client.get(claim_plivo_url, follow=True)
             self.assertNotIn("account_trial", response.context)
-            self.assertContains(response, connect_plivo_url)
+            self.assertEqual(response.request["PATH_INFO"], connect_plivo_url)
 
         # let's add a number already connected to the account
         with patch("requests.get") as plivo_get:
@@ -65,12 +67,12 @@ def test_claim(self):
 
                 # claim it the US number
                 session = self.client.session
-                session[Channel.CONFIG_PLIVO_AUTH_ID] = "auth-id"
-                session[Channel.CONFIG_PLIVO_AUTH_TOKEN] = "auth-token"
+                session[PlivoType.CONFIG_AUTH_ID] = "auth-id"
+                session[PlivoType.CONFIG_AUTH_TOKEN] = "auth-token"
                 session.save()
 
-                self.assertTrue(Channel.CONFIG_PLIVO_AUTH_ID in self.client.session)
-                self.assertTrue(Channel.CONFIG_PLIVO_AUTH_TOKEN in self.client.session)
+                self.assertTrue(PlivoType.CONFIG_AUTH_ID in self.client.session)
+                self.assertTrue(PlivoType.CONFIG_AUTH_TOKEN in self.client.session)
 
                 response = self.client.post(claim_plivo_url, dict(phone_number="+1 606-268-1435", country="US"))
                 self.assertRedirects(response, reverse("public.public_welcome") + "?success")
@@ -81,16 +83,24 @@ def test_claim(self):
                 self.assertEqual(
                     channel.config,
                     {
-                        Channel.CONFIG_PLIVO_AUTH_ID: "auth-id",
-                        Channel.CONFIG_PLIVO_AUTH_TOKEN: "auth-token",
-                        Channel.CONFIG_PLIVO_APP_ID: "app-id",
+                        PlivoType.CONFIG_AUTH_ID: "auth-id",
+                        PlivoType.CONFIG_AUTH_TOKEN: "auth-token",
+                        PlivoType.CONFIG_APP_ID: "app-id",
                         Channel.CONFIG_CALLBACK_DOMAIN: "app.rapidpro.io",
                     },
                 )
                 self.assertEqual(channel.address, "+16062681435")
                 # no more credential in the session
-                self.assertFalse(Channel.CONFIG_PLIVO_AUTH_ID in self.client.session)
-                self.assertFalse(Channel.CONFIG_PLIVO_AUTH_TOKEN in self.client.session)
+                self.assertFalse(PlivoType.CONFIG_AUTH_ID in self.client.session)
+                self.assertFalse(PlivoType.CONFIG_AUTH_TOKEN in self.client.session)
+
+                response = self.client.get(reverse("channels.types.plivo.connect"))
+                self.assertEqual(302, response.status_code)
+                self.assertRedirects(response, reverse("channels.types.plivo.claim"))
+
+                response = self.client.get(reverse("channels.types.plivo.connect") + "?reset_creds=reset")
+                self.assertEqual(200, response.status_code)
+                self.assertEqual(list(response.context["form"].fields.keys()), ["auth_id", "auth_token", "loc"])
 
         # delete existing channels
         Channel.objects.all().delete()
@@ -118,12 +128,12 @@ def test_claim(self):
 
                 # claim it the US number
                 session = self.client.session
-                session[Channel.CONFIG_PLIVO_AUTH_ID] = "auth-id"
-                session[Channel.CONFIG_PLIVO_AUTH_TOKEN] = "auth-token"
+                session[PlivoType.CONFIG_AUTH_ID] = "auth-id"
+                session[PlivoType.CONFIG_AUTH_TOKEN] = "auth-token"
                 session.save()
 
-                self.assertTrue(Channel.CONFIG_PLIVO_AUTH_ID in self.client.session)
-                self.assertTrue(Channel.CONFIG_PLIVO_AUTH_TOKEN in self.client.session)
+                self.assertTrue(PlivoType.CONFIG_AUTH_ID in self.client.session)
+                self.assertTrue(PlivoType.CONFIG_AUTH_TOKEN in self.client.session)
 
                 response = self.client.post(claim_plivo_url, dict(phone_number="+1 606-268-1440", country="US"))
                 self.assertRedirects(response, reverse("public.public_welcome") + "?success")
@@ -133,17 +143,17 @@ def test_claim(self):
                 self.assertEqual(
                     channel.config,
                     {
-                        Channel.CONFIG_PLIVO_AUTH_ID: "auth-id",
-                        Channel.CONFIG_PLIVO_AUTH_TOKEN: "auth-token",
-                        Channel.CONFIG_PLIVO_APP_ID: "app-id",
+                        PlivoType.CONFIG_AUTH_ID: "auth-id",
+                        PlivoType.CONFIG_AUTH_TOKEN: "auth-token",
+                        PlivoType.CONFIG_APP_ID: "app-id",
                         Channel.CONFIG_CALLBACK_DOMAIN: "app.rapidpro.io",
                     },
                 )
 
                 self.assertEqual(channel.address, "+16062681440")
                 # no more credential in the session
-                self.assertFalse(Channel.CONFIG_PLIVO_AUTH_ID in self.client.session)
-                self.assertFalse(Channel.CONFIG_PLIVO_AUTH_TOKEN in self.client.session)
+                self.assertFalse(PlivoType.CONFIG_AUTH_ID in self.client.session)
+                self.assertFalse(PlivoType.CONFIG_AUTH_TOKEN in self.client.session)
 
                 self.assertEqual(mock_get.call_args_list[0][0][0], "https://api.plivo.com/v1/Account/auth-id/")
                 self.assertEqual(
@@ -193,3 +203,32 @@ def test_search(self, mock_get):
         response = self.client.post(search_url, {"country": "US", "pattern": ""})
 
         self.assertContains(response, "Bad request")
+
+    def test_connect_plivo(self):
+        self.login(self.admin)
+
+        # connect plivo
+        connect_url = reverse("channels.types.plivo.connect")
+
+        # simulate invalid credentials
+        with patch("requests.get") as mock_get:
+            mock_get.return_value = MockResponse(
+                401, "Could not verify your access level for that URL." "\nYou have to login with proper credentials"
+            )
+            response = self.client.post(connect_url, dict(auth_id="auth-id", auth_token="auth-token"))
+            self.assertContains(
+                response, "Your Plivo auth ID and auth token seem invalid. Please check them again and retry."
+            )
+            self.assertFalse(PlivoType.CONFIG_AUTH_ID in self.client.session)
+            self.assertFalse(PlivoType.CONFIG_AUTH_TOKEN in self.client.session)
+
+        # ok, now with a success
+        with patch("requests.get") as mock_get:
+            mock_get.return_value = MockResponse(200, json.dumps(dict()))
+            response = self.client.post(connect_url, dict(auth_id="auth-id", auth_token="auth-token"))
+
+            # plivo should be added to the session
+            self.assertEqual(self.client.session[PlivoType.CONFIG_AUTH_ID], "auth-id")
+            self.assertEqual(self.client.session[PlivoType.CONFIG_AUTH_TOKEN], "auth-token")
+
+            self.assertRedirect(response, reverse("channels.types.plivo.claim"))

temba/channels/types/plivo/type.py

@@ -3,18 +3,22 @@
 from django.urls import re_path
 from django.utils.translation import gettext_lazy as _
 
-from temba.channels.models import Channel, ChannelType
+from temba.channels.models import ChannelType
 from temba.contacts.models import URN
 from temba.utils.http import http_headers
 
-from .views import ClaimView, SearchView
+from .views import ClaimView, Connect, SearchView
 
 
 class PlivoType(ChannelType):
     """
     An Plivo channel (https://www.plivo.com/)
     """
 
+    CONFIG_AUTH_ID = "PLIVO_AUTH_ID"
+    CONFIG_AUTH_TOKEN = "PLIVO_AUTH_TOKEN"
+    CONFIG_APP_ID = "PLIVO_APP_ID"
+
     code = "PL"
     category = ChannelType.Category.PHONE
 
@@ -36,10 +40,14 @@ def deactivate(self, channel):
         config = channel.config
         requests.delete(
             "https://api.plivo.com/v1/Account/%s/Application/%s/"
-            % (config[Channel.CONFIG_PLIVO_AUTH_ID], config[Channel.CONFIG_PLIVO_APP_ID]),
-            auth=(config[Channel.CONFIG_PLIVO_AUTH_ID], config[Channel.CONFIG_PLIVO_AUTH_TOKEN]),
+            % (config[self.CONFIG_AUTH_ID], config[self.CONFIG_APP_ID]),
+            auth=(config[self.CONFIG_AUTH_ID], config[self.CONFIG_AUTH_TOKEN]),
             headers=http_headers(extra={"Content-Type": "application/json"}),
         )
 
     def get_urls(self):
-        return [self.get_claim_url(), re_path(r"^search$", SearchView.as_view(channel_type=self), name="search")]
+        return [
+            self.get_claim_url(),
+            re_path(r"^search$", SearchView.as_view(channel_type=self), name="search"),
+            re_path(r"^connect$", Connect.as_view(channel_type=self), name="connect"),
+        ]

temba/channels/types/plivo/views.py

@@ -63,8 +63,8 @@ def clean_phone_number(self):
     form_class = Form
 
     def pre_process(self, *args, **kwargs):
-        auth_id = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_ID, None)
-        auth_token = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_TOKEN, None)
+        auth_id = self.request.session.get(self.channel_type.CONFIG_AUTH_ID, None)
+        auth_token = self.request.session.get(self.channel_type.CONFIG_AUTH_TOKEN, None)
 
         headers = http_headers(extra={"Content-Type": "application/json"})
         response = requests.get(
@@ -74,7 +74,7 @@ def pre_process(self, *args, **kwargs):
         if response.status_code == 200:
             return None
         else:
-            return HttpResponseRedirect(reverse("orgs.org_plivo_connect"))
+            return HttpResponseRedirect(reverse("channels.types.plivo.connect"))
 
     def is_valid_country(self, calling_code: int) -> bool:
         return calling_code in CALLING_CODES
@@ -95,8 +95,8 @@ def get_search_countries_tuple(self):
         return COUNTRY_CHOICES
 
     def get_existing_numbers(self, org):
-        auth_id = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_ID, None)
-        auth_token = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_TOKEN, None)
+        auth_id = self.request.session.get(self.channel_type.CONFIG_AUTH_ID, None)
+        auth_token = self.request.session.get(self.channel_type.CONFIG_AUTH_TOKEN, None)
 
         headers = http_headers(extra={"Content-Type": "application/json"})
         response = requests.get(
@@ -120,8 +120,8 @@ def get_existing_numbers(self, org):
         return account_numbers
 
     def claim_number(self, user, phone_number, country, role):
-        auth_id = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_ID, None)
-        auth_token = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_TOKEN, None)
+        auth_id = self.request.session.get(self.channel_type.CONFIG_AUTH_ID, None)
+        auth_token = self.request.session.get(self.channel_type.CONFIG_AUTH_TOKEN, None)
 
         org = self.request.org
 
@@ -148,9 +148,9 @@ def claim_number(self, user, phone_number, country, role):
             plivo_app_id = None
 
         plivo_config = {
-            Channel.CONFIG_PLIVO_AUTH_ID: auth_id,
-            Channel.CONFIG_PLIVO_AUTH_TOKEN: auth_token,
-            Channel.CONFIG_PLIVO_APP_ID: plivo_app_id,
+            self.channel_type.CONFIG_AUTH_ID: auth_id,
+            self.channel_type.CONFIG_AUTH_TOKEN: auth_token,
+            self.channel_type.CONFIG_APP_ID: plivo_app_id,
             Channel.CONFIG_CALLBACK_DOMAIN: org.get_brand_domain(),
         }
 
@@ -201,11 +201,17 @@ def claim_number(self, user, phone_number, country, role):
 
         return channel
 
+    def get_context_data(self, **kwargs):
+        context = super().get_context_data(**kwargs)
+        context["current_creds_account"] = self.request.session.get(self.channel_type.CONFIG_AUTH_ID, None)
+
+        return context
+
     def remove_api_credentials_from_session(self):
-        if Channel.CONFIG_PLIVO_AUTH_ID in self.request.session:
-            del self.request.session[Channel.CONFIG_PLIVO_AUTH_ID]
-        if Channel.CONFIG_PLIVO_AUTH_TOKEN in self.request.session:
-            del self.request.session[Channel.CONFIG_PLIVO_AUTH_TOKEN]
+        if self.channel_type.CONFIG_AUTH_ID in self.request.session:
+            del self.request.session[self.channel_type.CONFIG_AUTH_ID]
+        if self.channel_type.CONFIG_AUTH_TOKEN in self.request.session:
+            del self.request.session[self.channel_type.CONFIG_AUTH_TOKEN]
 
 
 class SearchView(ChannelTypeMixin, OrgPermsMixin, SmartFormView):
@@ -218,8 +224,8 @@ class Form(forms.Form):
 
     def form_valid(self, form, *args, **kwargs):
         data = form.cleaned_data
-        auth_id = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_ID, None)
-        auth_token = self.request.session.get(Channel.CONFIG_PLIVO_AUTH_TOKEN, None)
+        auth_id = self.request.session.get(self.channel_type.CONFIG_AUTH_ID, None)
+        auth_token = self.request.session.get(self.channel_type.CONFIG_AUTH_TOKEN, None)
 
         try:
             params = dict(country_iso=data["country"], pattern=data.get("pattern"))
@@ -245,3 +251,68 @@ def form_valid(self, form, *args, **kwargs):
             return JsonResponse(numbers, safe=False)
         except Exception as e:
             return JsonResponse({"error": str(e)})
+
+
+class Connect(ChannelTypeMixin, OrgPermsMixin, SmartFormView):
+    class PlivoConnectForm(forms.Form):
+        auth_id = forms.CharField(help_text=_("Your Plivo auth ID"))
+        auth_token = forms.CharField(help_text=_("Your Plivo auth token"))
+
+        def clean(self):
+            super().clean()
+
+            auth_id = self.cleaned_data.get("auth_id", None)
+            auth_token = self.cleaned_data.get("auth_token", None)
+
+            headers = http_headers(extra={"Content-Type": "application/json"})
+
+            response = requests.get(
+                "https://api.plivo.com/v1/Account/%s/" % auth_id, headers=headers, auth=(auth_id, auth_token)
+            )
+
+            if response.status_code != 200:
+                raise ValidationError(
+                    _("Your Plivo auth ID and auth token seem invalid. Please check them again and retry.")
+                )
+
+            return self.cleaned_data
+
+    form_class = PlivoConnectForm
+    permission = "channels.channel_claim"
+    submit_button_name = "Save"
+    template_name = "channels/types/plivo/connect.html"
+    field_config = dict(auth_id=dict(label=""), auth_token=dict(label=""))
+    success_message = "Plivo credentials verified. You can now add a Plivo channel."
+    menu_path = "/settings/workspace"
+
+    def get_success_url(self):
+        return reverse("channels.types.plivo.claim")
+
+    def pre_process(self, *args, **kwargs):
+        reset_creds = self.request.GET.get("reset_creds", "")
+
+        org = self.request.org
+        last_plivo_channel = (
+            org.channels.filter(is_active=True, channel_type=self.channel_type.code).order_by("-created_on").first()
+        )
+
+        if last_plivo_channel and not reset_creds:
+            self.request.session[self.channel_type.CONFIG_AUTH_ID] = last_plivo_channel.config.get(
+                self.channel_type.CONFIG_AUTH_ID, ""
+            )
+            self.request.session[self.channel_type.CONFIG_AUTH_TOKEN] = last_plivo_channel.config.get(
+                self.channel_type.CONFIG_AUTH_TOKEN, ""
+            )
+            return HttpResponseRedirect(self.get_success_url())
+
+        return None
+
+    def form_valid(self, form):
+        auth_id = form.cleaned_data["auth_id"]
+        auth_token = form.cleaned_data["auth_token"]
+
+        # add the credentials to the session
+        self.request.session[self.channel_type.CONFIG_AUTH_ID] = auth_id
+        self.request.session[self.channel_type.CONFIG_AUTH_TOKEN] = auth_token
+
+        return HttpResponseRedirect(self.get_success_url())