8000 Add tests for taskrouter tokens · ShowLowTechnology/twilio-python@32ea994 · GitHub
[go: up one dir, main page]
Skip to content

Commit 32ea994

Browse files
skimbrelskimbrel
committed
Add tests for taskrouter tokens
1 parent 294fa63 commit 32ea994

File tree

2 files changed

+138
-1
lines changed

2 files changed

+138
-1
lines changed

‎tests/task_router/test_capability.py

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
import time
2+
import unittest
3+
4+
from twilio import jwt
5+
6+
from twilio.task_router import TaskRouterCapability
7+
8+
9+
class TaskRouterCapabilityTest(unittest.TestCase):
10+
11+
def setUp(self):
12+
self.account_sid = "AC123"
13+
self.auth_token = "foobar"
14+
self.workspace_sid = "WS456"
15+
self.worker_sid = "WK789"
16+
self.cap = TaskRouterCapability(self.account_sid, self.auth_token,
17+
self.workspace_sid, self.worker_sid)
18+
19+
def test_generate_token(self):
20+
token = self.cap.generate_token()
21+
decoded = jwt.decode(token, self.auth_token)
22+
23+
self.assertIsNotNone(decoded)
24+
del decoded['exp']
25+
del decoded['policies']
26+
27+
expected = {
28+
'iss': self.account_sid,
29+
'account_sid': self.account_sid,
30+
'workspace_sid': self.workspace_sid,
31+
'worker_sid': self.worker_sid,
32+
'channel': self.worker_sid,
33+
'version': 'v1',
34+
'friendly_name': self.worker_sid,
35+
}
36+
self.assertDictEqual(expected, decoded)
37+
38+
def test_generate_token_default_ttl(self):
39+
token = self.cap.generate_token()
40+
decoded = jwt.decode(token, self.auth_token)
41+
42+
self.assertIsNotNone(decoded)
43+
self.assertEqual(int(time.time()) + 3600, decoded['exp'])
44+
45+
def test_generate_token_custom_ttl(self):
46+
token = self.cap.generate_token(10000)
47+
decoded = jwt.decode(token, self.auth_token)
48+
49+
self.assertIsNotNone(decoded)
50+
self.assertEqual(int(time.time()) + 10000, decoded['exp'])
51+
52+
def test_websockets_allowed(self):
53+
token = self.cap.generate_token()
54+
decoded = jwt.decode(token, self.auth_token)
55+
56+
self.assertIsNotNone(decoded)
57+
websocket_url = 'https://event-bridge.twilio.com/v1/wschannels/%s/%s' % (self.account_sid, self.worker_sid)
58+
expected = [
59+
{
60+
'url': websocket_url,
61+
'method': 'GET',
62+
'allow': True,
63+
'query_filter': {},
64+
'post_filter': {},
65+
},
66+
{
67+
'url': websocket_url,
68+
'method': 'POST',
69+
'allow': True,
70+
'query_filter': {},
71+
'post_filter': {},
72+
},
73+
]
74+
self.assertEqual(expected, decoded['policies'])
75+
76+
def test_allow_worker_activity_updates(self):
77+
self.cap.allow_worker_activity_updates()
78+
token = self.cap.generate_token()
79+
decoded = jwt.decode(token, self.auth_token)
80+
81+
self.assertIsNotNone(decoded)
82+
url = 'https://taskrouter.twilio.com/v1/Accounts/%s/Workspaces/%s/Workers/%s' % (
83+
self.account_sid,
84+
self.workspace_sid,
85+
self.worker_sid,
86+
)
87+
88+
expected = {
89+
'url': url,
90+
'method': 'POST',
91+
'allow': True,
92+
'query_filter': {},
93+
'post_filter': {'ActivitySid': {'required': True}},
94+
}
95+
self.assertEqual(expected, decoded['policies'][2])
96+
97+
def test_allow_worker_fetch_attributes(self):
98+
self.cap.allow_worker_fetch_attributes()
99+
token = self.cap.generate_token()
100+
decoded = jwt.decode(token, self.auth_token)
101+
102+
self.assertIsNotNone(decoded)
103+
url = 'https://taskrouter.twilio.com/v1/Accounts/%s/Workspaces/%s/Workers/%s' % (
104+
self.account_sid,
105+
self.workspace_sid,
106+
self.worker_sid,
107+
)
108+
109+
expected = {
110+
'url': url,
111+
'method': 'GET',
112+
'allow': True,
113+
'query_filter': {},
114+
'post_filter': {},
115+
}
116+
117+
self.assertEqual(expected, decoded['policies'][2])
118+
119+
def test_allow_task_reservation_updates(self):
120+
self.cap.allow_task_reservation_updates()
121+
token = self.cap.generate_token()
122+
decoded = jwt.decode(token, self.auth_token)
123+
124+
self.assertIsNotNone(decoded)
125+
url = 'https://taskrouter.twilio.com/v1/Accounts/%s/Workspaces/%s/Tasks/**' % (
126+
self.account_sid,
127+
self.workspace_sid,
128+
)
129+
130+
expected = {
131+
'url': url,
132+
'method': 'POST',
133+
'allow': True,
134+
'query_filter': {},
135+
'post_filter': {'ReservationStatus': {'required': True}},
136+
}
137+
self.assertEqual(expected, decoded['policies'][2])

‎twilio/task_router/__init__.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ def worker_url(self):
4646
return '%s/Workers/%s' % (self.workspace_url, self.worker_sid)
4747

4848
def _allow_worker_websocket_urls(self):
49-
worker_event_url = '%s/%s/%s' % (self.base_ws_url, self.account_sid, self.worker_sid),
49+
worker_event_url = '%s/%s/%s' % (self.base_ws_url, self.account_sid, self.worker_sid)
5050
self.policies.append(make_policy(
5151
worker_event_url,
5252
'GET',

0 commit comments

Comments
 (0)
0