1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
import os
from http.server import BaseHTTPRequestHandler
from twilio.request_validator import RequestValidator
class RequestHandler(BaseHTTPRequestHandler):
is_request_valid = False
validator = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])
def do_GET(self):
self.process_request()
def do_POST(self):
self.process_request()
def process_request(self):
self.signature_header = self.headers.get("x-twilio-signature")
self.url = (
self.headers.get("x-forwarded-proto")
+ "://"
+ self.headers.get("host")
+ self.path
)
self.send_response(200)
self.end_headers()
RequestHandler.is_request_valid = RequestHandler.validator.validate(
uri=self.url, params=None, signature=self.signature_header
)
# class WebhookTest(unittest.TestCase):
# def setUp(self):
# api_key = os.environ["TWILIO_API_KEY"]
# api_secret = os.environ["TWILIO_API_SECRET"]
# account_sid = os.environ["TWILIO_ACCOUNT_SID"]
# self.client = Client(api_key, api_secret, account_sid)
#
# portNumber = 7777
# self.validation_server = HTTPServer(("", portNumber), RequestHandler)
# self.tunnel = ngrok.connect(portNumber)
# self.flow_sid = ""
# _thread.start_new_thread(self.start_http_server, ())
#
# def start_http_server(self):
# self.validation_server.serve_forever()
#
# def tearDown(self):
# self.client.studio.v2.flows(self.flow_sid).delete()
# ngrok.kill()
# self.validation_server.shutdown()
# self.validation_server.server_close()
#
# def create_studio_flow(self, url, method):
# flow = self.client.studio.v2.flows.create(
# friendly_name="Python Cluster Test Flow",
# status="published",
# definition={
# "description": "Studio Flow",
# "states": [
# {
# "name": "Trigger",
# "type": "trigger",
# "transitions": [
# {
# "next": "httpRequest",
# "event": "incomingRequest",
# },
# ],
# "properties": {},
# },
# {
# "name": "httpRequest",
# "type": "make-http-request",
# "transitions": [],
# "properties": {
# "method": method,
# "content_type": "application/x-www-form-urlencoded;charset=utf-8",
# "url": url,
# },
# },
# ],
# "initial_state": "Trigger",
# "flags": {
# "allow_concurrent_calls": True,
# },
# },
# )
# return flow
#
# def validate(self, method):
# flow = self.create_studio_flow(url=self.tunnel.public_url, method=method)
# self.flow_sid = flow.sid
# time.sleep(5)
# self.client.studio.v2.flows(self.flow_sid).executions.create(
# to="to", from_="from"
# )
#
# def test_get(self):
# time.sleep(5)
# self.validate("GET")
# time.sleep(5)
# self.assertEqual(RequestHandler.is_request_valid, True)
#
# def test_post(self):
# time.sleep(5)
# self.validate("POST")
# time.sleep(5)
# self.assertEqual(RequestHandler.is_request_valid, True)
|
