Add Webhook update activation feature

alice/webhooks/webhooks.py

@@ -114,6 +114,41 @@ def update_webhook(
                 )
             )
 
+    def update_webhook_activation(
+        self, webhook_id: str, active: bool, verbose: bool = False
+    ) -> Result[Dict, OnboardingError]:
+        """
+
+        Update the activation of a Webhook
+
+        Parameters
+        ----------
+        webhook_id
+            Webhook identifier
+        active
+            Activation boolean value
+        verbose
+            Used for print service response as well as the time elapsed
+
+
+        Returns
+        -------
+            A Result where if the operation is successful it returns a webhook_id.
+            Otherwise, it returns an OnboardingError.
+        """
+        response = self.webhooks_client.update_webhook_activation(
+            webhook_id=webhook_id, active=active, verbose=verbose
+        )
+
+        if response.status_code == 200:
+            return isSuccess
+        else:
+            return Failure(
+                OnboardingError.from_response(
+                    operation="update_webhook_activation", response=response
+                )
+            )
+
     def ping_webhook(
         self, webhook_id: str, verbose: bool = False
     ) -> Result[bool, OnboardingError]:

alice/webhooks/webhooks_client.py

@@ -137,6 +137,45 @@ def update_webhook(self, webhook: Webhook, verbose: bool = False) -> Response:
 
         return response
 
+    @timeit
+    def update_webhook_activation(
+        self, webhook_id: str, active: bool, verbose: bool = False
+    ) -> Response:
+        """
+
+       Update Webhook activation
+
+       Parameters
+       ----------
+       webhook_id
+           Webhook identifier
+       active
+           Activation boolean value
+       verbose
+           Used for print service response as well as the time elapsed
+
+        Returns
+        -------
+            A Response object [requests library]
+        """
+        print_intro("update_webhook_activation", verbose=verbose)
+
+        backend_token = self.auth.create_backend_token().unwrap()
+        print_token("backend_token_with_user", backend_token, verbose=verbose)
+
+        headers = self._auth_headers(backend_token)
+        headers["Content-Type"] = "application/json"
+
+        response = requests.patch(
+            self.url + f"/webhook/{webhook_id}",
+            headers=headers,
+            json={"active": active},
+        )
+
+        print_response(response=response, verbose=verbose)
+
+        return response
+
     @timeit
     def ping_webhook(self, webhook_id: str, verbose: bool = False) -> Response:
         """

examples/onboarding_with_webhooks.py

@@ -21,7 +21,7 @@ def configure_webhooks(api_key: str, verbose: bool = False):
 
     # Create a new Webhook
     webhook = Webhook(
-        active=False,
+        active=True,
         post_url="http://google.com",
         api_key="b0b905d6-228f-44bf-a130-c85d7aecd765",
         event_name=selected_event.get("name"),
@@ -33,7 +33,7 @@ def configure_webhooks(api_key: str, verbose: bool = False):
     # Update an existent Webhook
     webhook_to_update = Webhook(
         webhook_id=webhook_id,  # Needed if we want to update
-        active=True,
+        active=False,
         post_url="http://alicebiometrics.com",
         api_key="b0b905d6-228f-44bf-a130-c85d7aecd765",
         event_name="user_created",
@@ -43,6 +43,11 @@ def configure_webhooks(api_key: str, verbose: bool = False):
     )
     webhooks_client.update_webhook(webhook_to_update, verbose)
 
+    # Update the activation of a Webhook
+    webhooks_client.update_webhook_activation(webhook_id, True, verbose)
+    retrieved_webhook = webhooks_client.get_webhook(webhook_id, verbose).unwrap()
+    assert retrieved_webhook.active
+
     # Send a ping using configured webhook
     result = webhooks_client.ping_webhook(webhook_id, verbose)
 

tests/test_integration_webhooks.py

@@ -26,7 +26,7 @@ def test_should_execute_all_webhook_lifecycle(given_valid_api_key):
 
     # Create a new Webhook
     webhook = Webhook(
-        active=False,
+        active=True,
         post_url="http://google.com",
         api_key="b0b905d6-228f-44bf-a130-c85d7aecd765",
         event_name=selected_event.get("name"),
@@ -36,10 +36,11 @@ def test_should_execute_all_webhook_lifecycle(given_valid_api_key):
     webhook_id = webhooks_client.create_webhook(webhook=webhook).unwrap_or_return()
 
     # Update an existent Webhook
+    new_post_url = "http://alicebiometrics.com"
     webhook_to_update = Webhook(
         webhook_id=webhook_id,  # Needed if we want to update
-        active=True,
-        post_url="http://alicebiometrics.com",
+        active=False,
+        post_url=new_post_url,
         api_key="b0b905d6-228f-44bf-a130-c85d7aecd765",
         event_name="user_created",
         event_version="1",
@@ -48,6 +49,15 @@ def test_should_execute_all_webhook_lifecycle(given_valid_api_key):
     )
     result = webhooks_client.update_webhook(webhook_to_update)
     assert_success(result)
+    retrieved_webhook = webhooks_client.get_webhook(webhook_id).unwrap()
+    assert retrieved_webhook.post_url == new_post_url
+    assert not retrieved_webhook.active
+
+    # Update Webhook activation
+    result = webhooks_client.update_webhook_activation(webhook_id, True)
+    assert_success(result)
+    retrieved_webhook = webhooks_client.get_webhook(webhook_id).unwrap()
+    assert not retrieved_webhook.active
 
     # Send a ping using configured webhook
     result = webhooks_client.ping_webhook(webhook_id)