import unittest import time from twilio.jwt.taskrouter.capabilities import ( WorkerCapabilityToken, TaskQueueCapabilityToken, WorkspaceCapabilityToken, ) class TaskQueueCapabilityTokenTest(unittest.TestCase): def setUp(self): self.account_sid = "AC123" self.auth_token = "foobar" self.workspace_sid = "WS456" self.taskqueue_sid = "WQ789" self.capability = TaskQueueCapabilityToken( self.account_sid, self.auth_token, self.workspace_sid, self.taskqueue_sid ) def test_generate_token(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertEqual(decoded.payload["iss"], self.account_sid) self.assertEqual(decoded.payload["account_sid"], self.account_sid) self.assertEqual(decoded.payload["workspace_sid"], self.workspace_sid) self.assertEqual(decoded.payload["taskqueue_sid"], self.taskqueue_sid) self.assertEqual(decoded.payload["channel"], self.taskqueue_sid) self.assertEqual(decoded.payload["version"], "v1") self.assertEqual(decoded.payload["friendly_name"], self.taskqueue_sid) def test_generate_token_with_default_ttl(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertAlmostEqual(int(time.time()) + 3600, decoded.valid_until, delta=5) def test_generate_token_with_custom_ttl(self): ttl = 10000 token = self.capability.to_jwt(ttl=ttl) self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertAlmostEqual(int(time.time()) + 10000, decoded.valid_until, delta=5) def test_default(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 3) # websocket GET get_policy = policies[0] self.assertEqual( "https://event-bridge.twilio.com/v1/wschannels/AC123/WQ789", get_policy["url"], ) self.assertEqual("GET", get_policy["method"]) self.assertTrue(get_policy["allow"]) self.assertEqual({}, get_policy["query_filter"]) self.assertEqual({}, get_policy["post_filter"]) # websocket POST post_policy = policies[1] self.assertEqual( "https://event-bridge.twilio.com/v1/wschannels/AC123/WQ789", post_policy["url"], ) self.assertEqual("POST", post_policy["method"]) self.assertTrue(post_policy["allow"]) self.assertEqual({}, post_policy["query_filter"]) self.assertEqual({}, post_policy["post_filter"]) # fetch GET fetch_policy = policies[2] self.assertEqual( "https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789", fetch_policy["url"], ) self.assertEqual("GET", fetch_policy["method"]) self.assertTrue(fetch_policy["allow"]) self.assertEqual({}, fetch_policy["query_filter"]) self.assertEqual({}, fetch_policy["post_filter"]) def test_allow_fetch_subresources(self): self.capability.allow_fetch_subresources() token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 4) # confirm the additional policy generated with allow_fetch_subresources() policy = policies[3] self.assertEqual( policy["url"], "https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789/**", ) self.assertEqual(policy["method"], "GET") self.assertTrue(policy["allow"]) self.assertEqual({}, policy["query_filter"]) self.assertEqual({}, policy["post_filter"]) def test_allow_updates_subresources(self): self.capability.allow_update_subresources() token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 4) # confirm the additional policy generated with allow_updates_subresources() policy = policies[3] self.assertEqual( policy["url"], "https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789/**", ) self.assertEqual(policy["method"], "POST") self.assertTrue(policy["allow"]) self.assertEqual({}, policy["query_filter"]) self.assertEqual({}, policy["post_filter"]) def test_pass_policy_in_constructor(self): self.capability = TaskQueueCapabilityToken( self.account_sid, self.auth_token, self.workspace_sid, self.taskqueue_sid, allow_update_subresources=True, ) token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = TaskQueueCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 4) # confirm the additional policy generated with allow_updates_subresources() policy = policies[3] self.assertEqual( policy["url"], "https://taskrouter.twilio.com/v1/Workspaces/WS456/TaskQueues/WQ789/**", ) self.assertEqual(policy["method"], "POST") self.assertTrue(policy["allow"]) self.assertEqual({}, policy["query_filter"]) self.assertEqual({}, policy["post_filter"]) class WorkerCapabilityTokenTest(unittest.TestCase): def check_policy(self, method, url, policy): self.assertEqual(url, policy["url"]) self.assertEqual(method, policy["method"]) self.assertTrue(policy["allow"]) self.assertEqual({}, policy["query_filter"]) self.assertEqual({}, policy["post_filter"]) def check_decoded( self, decoded, account_sid, workspace_sid, channel_id, channel_sid=None ): self.assertEqual(decoded["iss"], account_sid) self.assertEqual(decoded["account_sid"], account_sid) self.assertEqual(decoded["workspace_sid"], workspace_sid) self.assertEqual(decoded["channel"], channel_id) self.assertEqual(decoded["version"], "v1") self.assertEqual(decoded["friendly_name"], channel_id) if "worker_sid" in decoded.keys(): self.assertEqual(decoded["worker_sid"], channel_sid) if "taskqueue_sid" in decoded.keys(): self.assertEqual(decoded["taskqueue_sid"], channel_sid) def setUp(self): self.account_sid = "AC123" self.auth_token = "foobar" self.workspace_sid = "WS456" self.worker_sid = "WK789" self.capability = WorkerCapabilityToken( self.account_sid, self.auth_token, self.workspace_sid, self.worker_sid ) def test_generate_token(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.check_decoded( decoded.payload, self.account_sid, self.workspace_sid, self.worker_sid, self.worker_sid, ) def test_generate_token_with_default_ttl(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertAlmostEqual(int(time.time()) + 3600, decoded.valid_until, delta=5) def test_generate_token_with_custom_ttl(self): ttl = 10000 token = self.capability.to_jwt(ttl=ttl) self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertAlmostEqual(int(time.time()) + 10000, decoded.valid_until, delta=5) def test_defaults(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.decode(token, self.auth_token) self.assertNotEqual(None, decoded) websocket_url = "https://event-bridge.twilio.com/v1/wschannels/{0}/{1}".format( self.account_sid, self.worker_sid ) # expect 6 policies policies = decoded.payload["policies"] self.assertEqual(len(policies), 6) # should expect 6 policies for method, url, policy in [ ("GET", websocket_url, policies[0]), ("POST", websocket_url, policies[1]), ( "GET", "https://taskrouter.twilio.com/v1/Workspaces/WS456/Workers/WK789", policies[2], ), ( "GET", "https://taskrouter.twilio.com/v1/Workspaces/WS456/Activities", policies[3], ), ( "GET", "https://taskrouter.twilio.com/v1/Workspaces/WS456/Tasks/**", policies[4], ), ( "GET", "https://taskrouter.twilio.com/v1/Workspaces/WS456/Workers/WK789/Reservations/**", policies[5], ), ]: yield self.check_policy, method, url, policy def test_allow_activity_updates(self): # allow activity updates to the worker self.capability.allow_update_activities() token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 7) policy = policies[6] url = "https://taskrouter.twilio.com/v1/Workspaces/{0}/Workers/{1}".format( self.workspace_sid, self.worker_sid ) self.assertEqual(url, policy["url"]) self.assertEqual("POST", policy["method"]) self.assertTrue(policy["allow"]) self.assertNotEqual(None, policy["post_filter"]) self.assertEqual({}, policy["query_filter"]) self.assertTrue(policy["post_filter"]["ActivitySid"]) def test_allow_reservation_updates(self): # allow reservation updates self.capability.allow_update_reservations() token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 8) taskPolicy = policies[6] tasksUrl = "https://taskrouter.twilio.com/v1/Workspaces/{0}/Tasks/**".format( self.workspace_sid ) self.check_policy("POST", tasksUrl, taskPolicy) workerReservationsPolicy = policies[7] reservationsUrl = "https://taskrouter.twilio.com/v1/Workspaces/{0}/Workers/{1}/Reservations/**".format( self.workspace_sid, self.worker_sid ) self.check_policy("POST", reservationsUrl, workerReservationsPolicy) def test_pass_policies_in_constructor(self): # allow reservation updates self.capability = WorkerCapabilityToken( self.account_sid, self.auth_token, self.workspace_sid, self.worker_sid, allow_update_reservations=True, ) token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkerCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 8) taskPolicy = policies[6] tasksUrl = "https://taskrouter.twilio.com/v1/Workspaces/{0}/Tasks/**".format( self.workspace_sid ) self.check_policy("POST", tasksUrl, taskPolicy) workerReservationsPolicy = policies[7] reservationsUrl = "https://taskrouter.twilio.com/v1/Workspaces/{0}/Workers/{1}/Reservations/**".format( self.workspace_sid, self.worker_sid ) self.check_policy("POST", reservationsUrl, workerReservationsPolicy) class WorkspaceCapabilityTokenTest(unittest.TestCase): def check_policy(self, method, url, policy): self.assertEqual(url, policy["url"]) self.assertEqual(method, policy["method"]) self.assertTrue(policy["allow"]) self.assertEqual({}, policy["query_filter"]) self.assertEqual({}, policy["post_filter"]) def check_decoded( self, decoded, account_sid, workspace_sid, channel_id, channel_sid=None ): self.assertEqual(decoded["iss"], account_sid) self.assertEqual(decoded["account_sid"], account_sid) self.assertEqual(decoded["workspace_sid"], workspace_sid) self.assertEqual(decoded["channel"], channel_id) self.assertEqual(decoded["version"], "v1") self.assertEqual(decoded["friendly_name"], channel_id) if "worker_sid" in decoded.keys(): self.assertEqual(decoded["worker_sid"], channel_sid) if "taskqueue_sid" in decoded.keys(): self.assertEqual(decoded["taskqueue_sid"], channel_sid) def setUp(self): self.account_sid = "AC123" self.auth_token = "foobar" self.workspace_sid = "WS456" self.capability = WorkspaceCapabilityToken( self.account_sid, self.auth_token, self.workspace_sid ) def test_generate_token(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.check_decoded( decoded.payload, self.account_sid, self.workspace_sid, self.workspace_sid ) def test_generate_token_with_default_ttl(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertAlmostEqual(int(time.time()) + 3600, decoded.valid_until, delta=5) def test_generate_token_with_custom_ttl(self): ttl = 10000 token = self.capability.to_jwt(ttl=ttl) self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) self.assertAlmostEqual(int(time.time()) + 10000, decoded.valid_until, delta=5) def test_default(self): token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 3) for method, url, policy in [ ( "GET", "https://event-bridge.twilio.com/v1/wschannels/AC123/WS456", policies[0], ), ( "POST", "https://event-bridge.twilio.com/v1/wschannels/AC123/WS456", policies[1], ), ("GET", "https://taskrouter.twilio.com/v1/Workspaces/WS456", policies[2]), ]: yield self.check_policy, method, url, policy def test_allow_fetch_subresources(self): self.capability.allow_fetch_subresources() token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 4) # confirm the additional policy generated with allow_fetch_subresources() policy = policies[3] self.check_policy( "GET", "https://taskrouter.twilio.com/v1/Workspaces/WS456/**", policy ) def test_allow_updates_subresources(self): self.capability.allow_update_subresources() token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 4) # confirm the additional policy generated with allow_update_subresources() policy = policies[3] self.check_policy( "POST", "https://taskrouter.twilio.com/v1/Workspaces/WS456/**", policy ) def test_pass_policy_in_constructor(self): self.capability = WorkspaceCapabilityToken( self.account_sid, self.auth_token, self.workspace_sid, allow_update_subresources=True, ) token = self.capability.to_jwt() self.assertNotEqual(None, token) decoded = WorkspaceCapabilityToken.from_jwt(token, self.auth_token) self.assertNotEqual(None, decoded) policies = decoded.payload["policies"] self.assertEqual(len(policies), 4) # confirm the additional policy generated with allow_update_subresources() policy = policies[3] self.check_policy( "POST", "https://taskrouter.twilio.com/v1/Workspaces/WS456/**", policy ) if __name__ == "__main__": unittest.main()