Support webhook setting APIs

linebot/api.py

@@ -1129,6 +1129,77 @@ def get_insight_message_event(self, request_id, timeout=None):
 
         return InsightMessageEventResponse.new_from_json_dict(response.json)
 
+    def set_webhook_endpoint(self, webhook_endpoint, timeout=None):
+        """Set the webhook endpoint URL.
+
+        https://developers.line.biz/en/reference/messaging-api/#set-webhook-endpoint-url
+
+        :param str webhook_endpoint: A valid webhook URL to be set.
+        :param timeout: (optional) How long to wait for the server
+            to send data before giving up, as a float,
+            or a (connect timeout, read timeout) float tuple.
+            Default is self.http_client.timeout
+        :type timeout: float | tuple(float, float)
+        """
+        data = {
+            'endpoint': webhook_endpoint
+        }
+
+        self._put(
+            '/v2/bot/channel/webhook/endpoint',
+            data=json.dumps(data),
+            timeout=timeout,
+        )
+
+    def get_webhook_endpoint(self, timeout=None):
+        """Get information on a webhook endpoint.
+
+        https://developers.line.biz/en/reference/messaging-api/#get-webhook-endpoint-information
+
+        :param timeout: (optional) How long to wait for the server
+            to send data before giving up, as a float,
+            or a (connect timeout, read timeout) float tuple.
+            Default is self.http_client.timeout
+        :type timeout: float | tuple(float, float)
+        :rtype: dict
+        :return: Webhook information, including `endpoint` for webhook
+            URL and `active` for webhook usage status.
+        """
+        response = self._get(
+            '/v2/bot/channel/webhook/endpoint',
+            timeout=timeout,
+        )
+
+        return response.json
+
+    def test_webhook_endpoint(self, webhook_endpoint=None, timeout=None):
+        """Checks if the configured webhook endpoint can receive a test webhook event.
+
+        https://developers.line.biz/en/reference/messaging-api/#test-webhook-endpoint
+
+        :param webhook_endpoint: (optional) Set this parameter to
+            specific the webhook endpoint of the webhook. Default is the webhook
+            endpoint that is already set to the channel.
+        :param timeout: (optional) How long to wait for the server
+            to send data before giving up, as a float,
+            or a (connect timeout, read timeout) float tuple.
+            Default is self.http_client.timeout
+        :type timeout: float | tuple(float, float)
+        :rtype: dict
+        """
+        data = {}
+
+        if webhook_endpoint is not None:
+            data['endpoint'] = webhook_endpoint
+
+        response = self._post(
+            '/v2/bot/channel/webhook/test',
+            data=json.dumps(data),
+            timeout=timeout,
+        )
+
+        return response.json
+
     def _get(self, path, endpoint=None, params=None, headers=None, stream=False, timeout=None):
         url = (endpoint or self.endpoint) + path
 
@@ -1171,6 +1242,20 @@ def _delete(self, path, endpoint=None, data=None, headers=None, timeout=None):
         self.__check_error(response)
         return response
 
+    def _put(self, path, endpoint=None, data=None, headers=None, timeout=None):
+        url = (endpoint or self.endpoint) + path
+
+        if headers is None:
+            headers = {'Content-Type': 'application/json'}
+        headers.update(self.headers)
+
+        response = self.http_client.put(
+            url, headers=headers, data=data, timeout=timeout
+        )
+
+        self.__check_error(response)
+        return response
+
     @staticmethod
     def __check_error(response):
         if 200 <= response.status_code < 300:

linebot/http_client.py

@@ -92,6 +92,23 @@ def delete(self, url, headers=None, data=None, timeout=None):
         """
         raise NotImplementedError
 
+    @abstractmethod
+    def put(self, url, headers=None, data=None, timeout=None):
+        """PUT request.
+
+        :param str url: Request url
+        :param dict headers: (optional) Request headers
+        :param data: (optional) Dictionary, bytes, or file-like object to send in the body
+        :param timeout: (optional), How long to wait for the server
+            to send data before giving up, as a float,
+            or a (connect timeout, read timeout) float tuple.
+            Default is :py:attr:`self.timeout`
+        :type timeout: float | tuple(float, float)
+        :rtype: :py:class:`RequestsHttpResponse`
+        :return: RequestsHttpResponse instance
+        """
+        raise NotImplementedError
+
 
 class RequestsHttpClient(HttpClient):
     """HttpClient implemented by requests."""
@@ -177,6 +194,29 @@ def delete(self, url, headers=None, data=None, timeout=None):
 
         return RequestsHttpResponse(response)
 
+    def put(self, url, headers=None, data=None, timeout=None):
+        """PUT request.
+
+        :param str url: Request url
+        :param dict headers: (optional) Request headers
+        :param data: (optional) Dictionary, bytes, or file-like object to send in the body
+        :param timeout: (optional), How long to wait for the server
+            to send data before giving up, as a float,
+            or a (connect timeout, read timeout) float tuple.
+            Default is :py:attr:`self.timeout`
+        :type timeout: float | tuple(float, float)
+        :rtype: :py:class:`RequestsHttpResponse`
+        :return: RequestsHttpResponse instance
+        """
+        if timeout is None:
+            timeout = self.timeout
+
+        response = requests.put(
+            url, headers=headers, data=data, timeout=timeout
+        )
+
+        return RequestsHttpResponse(response)
+
 
 class HttpResponse(with_metaclass(ABCMeta)):
     """HttpResponse."""

tests/api/test_get_webhook_endpoint.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License. You may obtain
+#  a copy of the License at
+#
+#       https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+
+from __future__ import unicode_literals, absolute_import
+
+import unittest
+
+import responses
+
+from linebot import (
+    LineBotApi
+)
+
+
+class TestLineBotApi(unittest.TestCase):
+    def setUp(self):
+        self.tested = LineBotApi('channel_secret')
+
+    @responses.activate
+    def test_get_webhook_endpoint(self):
+        responses.add(
+            responses.GET,
+            LineBotApi.DEFAULT_API_ENDPOINT + '/v2/bot/channel/webhook/endpoint',
+            json={
+                "endpoint": "https://example.herokuapp.com/test",
+                "active": "true",
+            },
+            status=200
+        )
+
+        webhook = self.tested.get_webhook_endpoint()
+
+        request = responses.calls[0].request
+        self.assertEqual(request.method, 'GET')
+        self.assertEqual(
+            request.url,
+            LineBotApi.DEFAULT_API_ENDPOINT + '/v2/bot/channel/webhook/endpoint')
+        self.assertEqual(webhook['endpoint'], 'https://example.herokuapp.com/test')
+        self.assertEqual(webhook['active'], 'true')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/api/test_set_webhook_endpoint.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License. You may obtain
+#  a copy of the License at
+#
+#       https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+
+from __future__ import unicode_literals, absolute_import
+
+import unittest
+
+import responses
+
+from linebot import (
+    LineBotApi
+)
+
+
+class TestLineBotApi(unittest.TestCase):
+    def setUp(self):
+        self.tested = LineBotApi('channel_secret')
+
+    @responses.activate
+    def test_set_webhook_endpoint(self):
+        responses.add(
+            responses.PUT,
+            LineBotApi.DEFAULT_API_ENDPOINT + '/v2/bot/channel/webhook/endpoint',
+            json={},
+            status=200
+        )
+
+        self.tested.set_webhook_endpoint('endpoint')
+
+        request = responses.calls[0].request
+        self.assertEqual(request.method, 'PUT')
+        self.assertEqual(
+            request.url,
+            LineBotApi.DEFAULT_API_ENDPOINT + '/v2/bot/channel/webhook/endpoint')
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/api/test_test_webhook_endpoint.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+
+#  Licensed under the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License. You may obtain
+#  a copy of the License at
+#
+#       https://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#  License for the specific language governing permissions and limitations
+#  under the License.
+
+from __future__ import unicode_literals, absolute_import
+
+import unittest
+
+import responses
+
+from linebot import (
+    LineBotApi
+)
+
+
+class TestLineBotApi(unittest.TestCase):
+    def setUp(self):
+        self.tested = LineBotApi('channel_secret')
+
+    @responses.activate
+    def test_test_webhook_endpoint_with_endpoint(self):
+        responses.add(
+            responses.POST,
+            LineBotApi.DEFAULT_API_ENDPOINT + '/v2/bot/channel/webhook/test',
+            json={
+                "success": "true",
+                "timestamp": "2020-09-30T05:38:20.031Z",
+                "statusCode": 200,
+                "reason": "OK",
+                "detail": "200"
+            },
+            status=200
+        )
+
+        result = self.tested.test_webhook_endpoint('endpoint')
+
+        request = responses.calls[0].request
+        self.assertEqual(request.method, 'POST')
+        self.assertEqual(
+            request.url,
+            LineBotApi.DEFAULT_API_ENDPOINT + '/v2/bot/channel/webhook/test')
+        self.assertEqual(result['success'], 'true')
+        self.assertEqual(result['timestamp'], '2020-09-30T05:38:20.031Z')
+        self.assertEqual(result['statusCode'], 200)
+        self.assertEqual(result['reason'], 'OK')
+        self.assertEqual(result['detail'], '200')
+
+
+if __name__ == '__main__':
+    unittest.main()