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 110 111 112 113 114 115 116 117 118 119
|
import time
import unittest
from twilio.jwt import Jwt
from twilio.jwt.client import ClientCapabilityToken, ScopeURI
class ClientCapabilityTokenTest(unittest.TestCase):
def assertIn(self, foo, bar, msg=None):
"""backport for 2.6"""
assert foo in bar, msg or "%s not found in %s" % (foo, bar)
def now(self):
return int(time.time())
def test_no_permissions(self):
token = ClientCapabilityToken("AC123", "XXXXX")
assert len(token._generate_payload()) == 1
assert token._generate_payload()["scope"] == ""
def test_inbound_permissions(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_client_incoming("andy")
eurl = "scope:client:incoming?clientName=andy"
assert len(token._generate_payload()) == 1
assert token._generate_payload()["scope"] == eurl
def test_outbound_permissions(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_client_outgoing("AP123")
eurl = "scope:client:outgoing?appSid=AP123"
assert len(token._generate_payload()) == 1
self.assertIn(eurl, token._generate_payload()["scope"])
def test_outbound_permissions_params(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_client_outgoing("AP123", foobar=3)
eurl = "scope:client:outgoing?appParams=foobar%3D3&appSid=AP123"
assert token.payload["scope"] == eurl
def test_events(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_event_stream()
event_uri = "scope:stream:subscribe?path=%2F2010-04-01%2FEvents"
assert token.payload["scope"] == event_uri
def test_events_with_filters(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_event_stream(foobar="hey")
event_uri = (
"scope:stream:subscribe?params=foobar%3Dhey&path=%2F2010-04-01%2FEvents"
)
assert token.payload["scope"] == event_uri
def test_decode(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_client_outgoing("AP123", foobar=3)
token.allow_client_incoming("andy")
token.allow_event_stream()
outgoing_uri = (
"scope:client:outgoing?appParams=foobar%3D3&appSid=AP123&clientName=andy"
)
incoming_uri = "scope:client:incoming?clientName=andy"
event_uri = "scope:stream:subscribe?path=%2F2010-04-01%2FEvents"
result = Jwt.from_jwt(token.to_jwt(), "XXXXX")
scope = result.payload["scope"].split(" ")
self.assertIn(outgoing_uri, scope)
self.assertIn(incoming_uri, scope)
self.assertIn(event_uri, scope)
def test_encode_full_payload(self):
token = ClientCapabilityToken("AC123", "XXXXX")
token.allow_event_stream(foobar="hey")
token.allow_client_incoming("andy")
event_uri = (
"scope:stream:subscribe?params=foobar%3Dhey&path=%2F2010-04-01%2FEvents"
)
incoming_uri = "scope:client:incoming?clientName=andy"
self.assertIn(event_uri, token.payload["scope"])
self.assertIn(incoming_uri, token.payload["scope"])
self.assertEqual(token.payload["iss"], "AC123")
self.assertGreaterEqual(token.payload["exp"], self.now())
def test_pass_scopes_in_constructor(self):
token = ClientCapabilityToken(
"AC123",
"XXXXX",
allow_client_outgoing={"application_sid": "AP123", "param1": "val1"},
)
outgoing_uri = "scope:client:outgoing?appParams=param1%3Dval1&appSid=AP123"
result = Jwt.from_jwt(token.to_jwt(), "XXXXX")
self.assertEqual(outgoing_uri, result.payload["scope"])
class ScopeURITest(unittest.TestCase):
def test_to_payload_no_params(self):
scope_uri = ScopeURI("service", "godmode")
self.assertEqual("scope:service:godmode", scope_uri.to_payload())
def test_to_payload_with_params(self):
scope_uri = ScopeURI("service", "godmode", {"key": "val"})
self.assertEqual("scope:service:godmode?key=val", scope_uri.to_payload())
def test_to_payload_with_params_encoded(self):
scope_uri = ScopeURI("service", "godmode", {"key with space": "val"})
self.assertEqual(
"scope:service:godmode?key+with+space=val", scope_uri.to_payload()
)
|