File: test_client.py

package info (click to toggle)
python-twilio 6.51.0%2Bds-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 12,260 kB
  • sloc: python: 128,982; makefile: 51
file content (114 lines) | stat: -rw-r--r-- 4,413 bytes parent folder | download | duplicates (2)
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
import unittest

import time
from nose.tools import assert_true, assert_equal

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"""
        return assert_true(foo in bar, msg=(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_equal(len(token._generate_payload()), 1)
        assert_equal(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_equal(len(token._generate_payload()), 1)
        assert_equal(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_equal(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_equal(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_equal(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_equal(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())