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
|
from tests import IntegrationTestCase
from tests.holodeck import Request
from twilio.base.page import Page
from twilio.http.response import Response
class TestPage(Page):
def get_instance(self, payload):
return payload
class StreamTestCase(IntegrationTestCase):
def setUp(self):
super(StreamTestCase, self).setUp()
self.holodeck.mock(
Response(
200,
"""
{
"next_page_uri": "/2010-04-01/Accounts/AC123/Messages.json?Page=1",
"messages": [{"body": "payload0"}, {"body": "payload1"}]
}
""",
),
Request(
url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages.json"
),
)
self.holodeck.mock(
Response(
200,
"""
{
"next_page_uri": "/2010-04-01/Accounts/AC123/Messages.json?Page=2",
"messages": [{"body": "payload2"}, {"body": "payload3"}]
}
""",
),
Request(
url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages.json?Page=1"
),
)
self.holodeck.mock(
Response(
200,
"""
{
"next_page_uri": null,
"messages": [{"body": "payload4"}]
}
""",
),
Request(
url="https://api.twilio.com/2010-04-01/Accounts/AC123/Messages.json?Page=2"
),
)
self.version = self.client.api.v2010
self.response = self.version.page(
method="GET", uri="/Accounts/AC123/Messages.json"
)
self.page = TestPage(self.version, self.response, {})
def test_stream(self):
messages = list(self.version.stream(self.page))
self.assertEqual(len(messages), 5)
def test_stream_limit(self):
messages = list(self.version.stream(self.page, limit=3))
self.assertEqual(len(messages), 3)
def test_stream_page_limit(self):
messages = list(self.version.stream(self.page, page_limit=1))
self.assertEqual(len(messages), 2)
class VersionTestCase(IntegrationTestCase):
def test_fetch_redirect(self):
self.holodeck.mock(
Response(307, '{"redirect_to": "some_place"}'),
Request(url="https://messaging.twilio.com/v1/Deactivations"),
)
response = self.client.messaging.v1.fetch(method="GET", uri="/Deactivations")
self.assertIsNotNone(response)
|
