8000 porting to python · twilio/twilio-python@099c06d · GitHub
[go: up one dir, main page]
Skip to content

Commit 099c06d

Browse files
wanjunsliwanjunsli
committed
porting to python
1 parent 1fdff6d commit 099c06d

8 files changed

+940
-144
lines changed

‎tests/task_router/test_task_router_capability.py

Lines changed: 253 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,253 @@
1+
import sys
2+
3+
import time
4+
import unittest
5+
import warnings
6+
7+
from twilio import jwt
8+
from twilio.task_router.capability import TaskRouterCapability
9+
10+
class TaskRouterCapabilityTest(unittest.TestCase):
11+
12+
def test_workspace_default(self):
13+
account_sid = "AC123"
14+
auth_token = "foobar"
15+
workspace_sid = "WS456"
16+
channel_id = "WS456"
17+
capability = TaskRouterCapability(account_sid, auth_token, workspace_sid, channel_id)
18+
19+
capability.generate_token()
20+
21+
token = capability.generate_token()
22+
self.assertIsNotNone(token)
23+
24+
decoded = jwt.decode(token, auth_token)
25+
self.assertIsNotNone(decoded)
26+
27+
self.assertEqual(decoded["iss"], account_sid)
28+
self.assertEqual(decoded["account_sid"], account_sid)
29+
self.assertEqual(decoded["workspace_sid"], workspace_sid)
30+
self.assertEqual(decoded["channel"], channel_id)
31+
self.assertEqual(decoded["version"], "v1")
32+
self.assertEqual(decoded["friendly_name"], channel_id)
33+
34+
policies = decoded['policies']
35+
self.assertEqual(len(policies), 3)
36+
37+
# websocket GET
38+
get_policy = policies[0]
39+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WS456", get_policy['url'])
40+
self.assertEqual("GET", get_policy['method'])
41+
self.assertTrue(get_policy['allowed'])
42+
self.assertEqual({}, get_policy['query_filter'])
43+
self.assertEqual({}, get_policy['post_filter'])
44+
45+
# websocket POST
46+
post_policy = policies[1]
47+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WS456", post_policy['url'])
48+
self.assertEqual("POST", post_policy['method'])
49+
self.assertTrue(post_policy['allowed'])
50+
self.assertEqual({}, post_policy['query_filter'])
51+
self.assertEqual({}, post_policy['post_filter'])
52+
53+
# fetch GET
54+
fetch_policy = policies[2]
55+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456", fetch_policy['url'])
56+
self.assertEqual("GET", fetch_policy['method'])
57+
self.assertTrue(fetch_policy['allowed'])
58+
self.assertEqual({}, fetch_policy['query_filter'])
59+
self.assertEqual({}, fetch_policy['post_filter'])
60+
61+
def test_worker_default(self):
62+
account_sid = "AC123"
63+
auth_token = "foobar"
64+
workspace_sid = "WS456"
65+
worker_sid = "WK789"
66+
capability = TaskRouterCapability(account_sid, auth_token, workspace_sid, worker_sid)
67+
68+
capability.generate_token()
69+
70+
token = capability.generate_token()
71+
self.assertIsNotNone(token)
72+
73+
decoded = jwt.decode(token, auth_token)
74+
self.assertIsNotNone(decoded)
75+
76+
self.a F438 ssertEqual(decoded["iss"], account_sid)
77+
self.assertEqual(decoded["account_sid"], account_sid)
78+
self.assertEqual(decoded["workspace_sid"], workspace_sid)
79+
self.assertEqual(decoded["worker_sid"], worker_sid)
80+
self.assertEqual(decoded["channel"], worker_sid)
81+
self.assertEqual(decoded["version"], "v1")
82+
self.assertEqual(decoded["friendly_name"], worker_sid)
83+
84+
policies = decoded['policies']
85+
self.assertEqual(len(policies), 4)
86+
87+
# activity GET
88+
fetch_activity = policies[0]
89+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456/Activities", fetch_activity['url'])
90+
self.assertEqual("GET", fetch_activity['method'])
91+
self.assertTrue(fetch_activity['allowed'])
92+
self.assertEqual({}, fetch_activity['query_filter'])
93+
self.assertEqual({}, fetch_activity['post_filter'])
94+
95+
# websocket GET
96+
get_policy = policies[1]
97+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WK789", get_policy['url'])
98+
self.assertEqual("GET", get_policy['method'])
99+
self.assertTrue(get_policy['allowed'])
100+
self.assertEqual({}, get_policy['query_filter'])
101+
self.assertEqual({}, get_policy['post_filter'])
102+
103+
# websocket POST
104+
post_policy = policies[2]
105+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WK789", post_policy['url'])
106+
self.assertEqual("POST", post_policy['method'])
107+
self.assertTrue(post_policy['allowed'])
108+
self.assertEqual({}, post_policy['query_filter'])
109+
self.assertEqual({}, post_policy['post_filter'])
110+
111+
# fetch GET
112+
fetch_policy = policies[3]
113+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456/Workers/WK789", fetch_policy['url'])
114+
self.assertEqual("GET", fetch_policy['method'])
115+
self.assertTrue(fetch_policy['allowed'])
116+
self.assertEqual({}, fetch_policy['query_filter'])
117+
self.assertEqual({}, fetch_policy['post_filter'])
118+
119+
def test_task_queue_default(self):
120+
account_sid = "AC123"
121+
auth_token = "foobar"
122+
workspace_sid = "WS456"
123+
taskqueue_sid = "WQ789"
124+
capability = TaskRouterCapability(account_sid, auth_token, workspace_sid, taskqueue_sid)
125+
126+
capability.generate_token()
127+
128+
token = capability.generate_token()
129+
self.assertIsNotNone(token)
130+
131+
decoded = jwt.decode(token, auth_token)
132+
self.assertIsNotNone(decoded)
133+
134+
self.assertEqual(decoded["iss"], account_sid)
135+
self.assertEqual(decoded["account_sid"], account_sid)
136+
self.assertEqual(decoded["workspace_sid"], workspace_sid)
137+
self.assertEqual(decoded["taskqueue_sid"], taskqueue_sid)
138+
self.assertEqual(decoded["channel"], taskqueue_sid)
139+
self.assertEqual(decoded["version"], "v1")
140+
self.assertEqual(decoded["friendly_name"], taskqueue_sid)
141+
142+
policies = decoded['policies']
143+
self.assertEqual(len(policies), 3)
144+
145+
# websocket GET
146+
get_policy = policies[0]
147+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WQ789", get_policy['url'])
148+
self.assertEqual("GET", get_policy['method'])
149+
self.assertTrue(get_policy['allowed'])
150+
self.assertEqual({}, get_policy['query_filter'])
151+
self.assertEqual({}, get_policy['post_filter'])
152+
153+
# websocket POST
154+
post_policy = policies[1]
155+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WQ789", post_policy['url'])
156+
self.assertEqual("POST", post_policy['method'])
157+
self.assertTrue(post_policy['allowed'])
158+
self.assertEqual({}, post_policy['query_filter'])
159+
self.assertEqual({}, post_policy['post_filter'])
160+
161+
# fetch GET
162+
fetch_policy = policies[2]
163+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789", fetch_policy['url'])
164+
self.assertEqual("GET", fetch_policy['method'])
165+
self.assertTrue(fetch_policy['allowed'])
166+
self.assertEqual({}, fetch_policy['query_filter'])
167+
self.assertEqual({}, fetch_policy['post_filter'])
168+
169+
def test_deprecated_worker(self):
170+
account_sid = "AC123"
171+
auth_token = "foobar"
172+
workspace_sid = "WS456"
173+
worker_sid = "WK789"
174+
capability = TaskRouterCapability(account_sid, auth_token, workspace_sid, worker_sid)
175+
176+
capability.generate_token()
177+
178+
token = capability.generate_token()
179+
self.assertIsNotNone(token)
180+
181+
decoded = jwt.decode(token, auth_token)
182+
self.assertIsNotNone(decoded)
183+
184+
self.assertEqual(decoded["iss"], account_sid)
185+
self.assertEqual(decoded["account_sid"], account_sid)
186+
self.assertEqual(decoded["workspace_sid"], workspace_sid)
187+
self.assertEqual(decoded["worker_sid"], worker_sid)
188+
self.assertEqual(decoded["channel"], worker_sid)
189+
self.assertEqual(decoded["version"], "v1")
190+
self.assertEqual(decoded["friendly_name"], worker_sid)
191+
192+
policies = decoded['policies']
193+
self.assertEqual(len(policies), 4)
194+
195+
# should expect 4 policies
196+
197+
# activity GET
198+
fetch_activity = policies[0]
199+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456/Activities", fetch_activity['url'])
200+
self.assertEqual("GET", fetch_activity['method'])
201+
self.assertTrue(fetch_activity['allowed'])
202+
self.assertEqual({}, fetch_activity['query_filter'])
203+
self.assertEqual({}, fetch_activity['post_filter'])
204+
205 F438 +
# websocket GET
206+
get_policy = policies[1]
207+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WK789", get_policy['url'])
208+
self.assertEqual("GET", get_policy['method'])
209+
self.assertTrue(get_policy['allowed'])
210+
self.assertEqual({}, get_policy['query_filter'])
211+
self.assertEqual({}, get_policy['post_filter'])
212+
213+
# websocket POST
214+
post_policy = policies[2]
215+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WK789", post_policy['url'])
216+
self.assertEqual("POST", post_policy['method'])
217+
self.assertTrue(post_policy['allowed'])
218+
self.assertEqual({}, post_policy['query_filter'])
219+
self.assertEqual({}, post_policy['post_filter'])
220+
221+
# fetch GET
222+
fetch_policy = policies[3]
223+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456/Workers/WK789", fetch_policy['url'])
224+
self.assertEqual("GET", fetch_policy['method'])
225+
self.assertTrue(fetch_policy['allowed'])
226+
self.assertEqual({}, fetch_policy['query_filter'])
227+
self.assertEqual({}, fetch_policy['post_filter'])
228+
229+
# check deprecated warnings
230+
with warnings.catch_warnings(record=True) as w:
231+
warnings.simplefilter("always")
232+
capability.allow_worker_fetch_attributes()
233+
assert len(w) == 1
234+
assert issubclass(w[-1].category, DeprecationWarning)
235+
assert "deprecated" in str(w[-1].message)
236+
237+
with warnings.catch_warnings(record=True) as w:
238+
warnings.simplefilter("always")
239+
capability.allow_worker_activity_updates()
240+
assert len(w) == 1
241+
assert issubclass(w[-1].category, DeprecationWarning)
242+
assert "deprecated" in str(w[-1].message)
243+
244+
with warnings.catch_warnings(record=True) as w:
245+
warnings.simplefilter("always")
246+
capability.allow_task_reservation_updates()
247+
assert len(w) == 1
248+
assert issubclass(w[-1].category, DeprecationWarning)
249+
assert "deprecated" in str(w[-1].message)
250+
251+
252+
if __name__ == "__main__":
253+
unittest.main()

‎tests/task_router/test_task_router_taskqueue_capability.py

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
import sys
2+
import time
3+
import unittest
4+
5+
from twilio import jwt
6+
from twilio.task_router.capability import TaskRouterTaskQueueCapability
7+
8+
class TaskRouterTaskQueueCapabilityTest(unittest.TestCase):
9+
10+
def setUp(self):
11+
self.account_sid = "AC123"
12+
self.auth_token = "foobar"
13+
self.workspace_sid = "WS456"
14+
self.taskqueue_sid = "WQ789"
15+
self.capability = TaskRouterTaskQueueCapability(self.account_sid, self.auth_token, self.workspace_sid, self.taskqueue_sid)
16+
17+
def test_generate_token(self):
18+
19+
token = self.capability.generate_token()
20+
self.assertIsNotNone(token)
21+
22+
decoded = jwt.decode(token, self.auth_token)
23+
self.assertIsNotNone(decoded)
24+
25+
self.assertEqual(decoded["iss"], self.account_sid)
26+
self.assertEqual(decoded["account_sid"], self.account_sid)
27+
self.assertEqual(decoded["workspace_sid"], self.workspace_sid)
28+
self.assertEqual(decoded["taskqueue_sid"], self.taskqueue_sid)
29+
self.assertEqual(decoded["channel"], self.taskqueue_sid)
30+
self.assertEqual(decoded["version"], "v1")
31+
self.assertEqual(decoded["friendly_name"], self.taskqueue_sid)
32+
33+
def test_generate_token_with_default_ttl(self):
34+
token = self.capability.generate_token()
35+
self.assertIsNotNone(token)
36+
37+
decoded = jwt.decode(token, self.auth_token)
38+
self.assertIsNotNone(decoded)
39+
40+
self.assertEqual(int(time.time()) + 3600, decoded["exp"])
41+
42+
def test_generate_token_with_custom_ttl(self):
43+
ttl = 10000
44+
45+
token = self.capability.generate_token(ttl)
46+
self.assertIsNotNone(token)
47+
48+
decoded = jwt.decode(token, self.auth_token)
49+
self.assertIsNotNone(decoded)
50+
51+
self.assertEqual(int(time.time()) + 10000, decoded["exp"])
52+
53+
def test_default(self):
54+
token = self.capability.generate_token()
55+
self.assertIsNotNone(token)
56+
57+
decoded = jwt.decode(token, self.auth_token)
58+
self.assertIsNotNone(decoded)
59+
60+
policies = decoded['policies']
61+
self.assertEqual(len(policies), 3)
62+
63+
# websocket GET
64+
get_policy = policies[0]
65+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WQ789", get_policy['url'])
66+
self.assertEqual("GET", get_policy['method'])
67+
self.assertTrue(get_policy['allowed'])
68+
self.assertEqual({}, get_policy['query_filter'])
69+
self.assertEqual({}, get_policy['post_filter'])
70+
71+
# websocket POST
72+
post_policy = policies[1]
73+
self.assertEqual("https://event-bridge.twilio.com/v1/wschannels/AC123/WQ789", post_policy['url'])
74+
self.assertEqual("POST", post_policy['method'])
75+
self.assertTrue(post_policy['allowed'])
76+
self.assertEqual({}, post_policy['query_filter'])
77+
self.assertEqual({}, post_policy['post_filter'])
78+
79+
# fetch GET
80+
fetch_policy = policies[2]
81+
self.assertEqual("https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789", fetch_policy['url'])
82+
self.assertEqual("GET", fetch_policy['method'])
83+
self.assertTrue(fetch_policy['allowed'])
84+
self.assertEqual({}, fetch_policy['query_filter'])
85+
self.assertEqual({}, fetch_policy['post_filter'])
86+
87+
def test_allow_fetch_subresources(self):
88+
self.capability.allow_fetch_subresources()
89+
90+
token = self.capability.generate_token()
91+
self.assertIsNotNone(token)
92+
93+
decoded = jwt.decode(token, self.auth_token)
94+
self.assertIsNotNone(decoded)
95+
96+
policies = decoded['policies']
97+
self.assertEqual(len(policies), 4)
98+
99+
# confirm the additional policy generated with allow_fetch_subresources()
100+
101+
policy = policies[3]
102+
103+
self.assertEqual(policy['url'], "https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789/**")
104+
self.assertEqual(policy['method'], "GET")
105+
self.assertTrue(policy['allowed'])
106+
self.assertEqual({}, policy['query_filter'])
107+
self.assertEqual({}, policy['post_filter'])
108+
109+
def test_allow_updates_subresources(self):
110+
self.capability.allow_updates_subresources()
111+
112+
token = self.capability.generate_token()
113+
self.assertIsNotNone(token)
114+
115+
decoded = jwt.decode(token, self.auth_token)
116+
self.assertIsNotNone(decoded)
117+
118+
policies = decoded['policies']
119+
self.assertEqual(len(policies), 4)
120+
121+
# confirm the additional policy generated with allow_updates_subresources()
122+
123+
policy = policies[3]
124+
125+
self.assertEqual(policy['url'], "https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789/**")
126+
self.assertEqual(policy['method'], "POST")
127+
self.assertTrue(policy['allowed'])
128+
self.assertEqual({}, policy['query_filter'])
129+
self.assertEqual({}, policy['post_filter'])
130+
131+
if __name__ == "__main__":
132+
unittest.main()

0 commit comments

Comments
 (0)
0