FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
google-api-python-client/tests/test_channel.py at master · arycloud/google-api-python-client · GitHub
arycloud
/
google-api-python-client
Public
forked from
googleapis/google-api-python-client
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
google-api-python-client
/
tests
/
test_channel.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
125 lines (104 loc) · 4.55 KB
Breadcrumbs
google-api-python-client
/
tests
/
test_channel.py
Copy path
File metadata and controls
125 lines (104 loc) · 4.55 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
120
121
122
123
124
125
"""Notification channels tests."""
from
__future__
import
absolute_import
__author__
=
'jcgregorio@google.com (Joe Gregorio)'
import
unittest2
as
unittest
import
datetime
from
googleapiclient
import
channel
from
googleapiclient
import
errors
class
TestChannel
(
unittest
.
TestCase
):
def
test_basic
(
self
):
ch
=
channel
.
Channel
(
'web_hook'
,
'myid'
,
'mytoken'
,
'http://example.org/callback'
,
expiration
=
0
,
params
=
{
'extra'
:
'info'
},
resource_id
=
'the_resource_id'
,
resource_uri
=
'http://example.com/resource_1'
)
# Converting to a body.
body
=
ch
.
body
()
self
.
assertEqual
(
'http://example.org/callback'
,
body
[
'address'
])
self
.
assertEqual
(
'myid'
,
body
[
'id'
])
self
.
assertEqual
(
'missing'
,
body
.
get
(
'expiration'
,
'missing'
))
self
.
assertEqual
(
'info'
,
body
[
'params'
][
'extra'
])
self
.
assertEqual
(
'the_resource_id'
,
body
[
'resourceId'
])
self
.
assertEqual
(
'http://example.com/resource_1'
,
body
[
'resourceUri'
])
self
.
assertEqual
(
'web_hook'
,
body
[
'type'
])
# Converting to a body with expiration set.
ch
.
expiration
=
1
body
=
ch
.
body
()
self
.
assertEqual
(
1
,
body
.
get
(
'expiration'
,
'missing'
))
# Converting to a body after updating with a response body.
ch
.
update
({
'resourceId'
:
'updated_res_id'
,
'resourceUri'
:
'updated_res_uri'
,
'some_random_parameter'
:
2
,
})
body
=
ch
.
body
()
self
.
assertEqual
(
'http://example.org/callback'
,
body
[
'address'
])
self
.
assertEqual
(
'myid'
,
body
[
'id'
])
self
.
assertEqual
(
1
,
body
.
get
(
'expiration'
,
'missing'
))
self
.
assertEqual
(
'info'
,
body
[
'params'
][
'extra'
])
self
.
assertEqual
(
'updated_res_id'
,
body
[
'resourceId'
])
self
.
assertEqual
(
'updated_res_uri'
,
body
[
'resourceUri'
])
self
.
assertEqual
(
'web_hook'
,
body
[
'type'
])
def
test_new_webhook_channel
(
self
):
ch
=
channel
.
new_webhook_channel
(
'http://example.com/callback'
)
self
.
assertEqual
(
0
,
ch
.
expiration
)
self
.
assertEqual
(
'http://example.com/callback'
,
ch
.
address
)
self
.
assertEqual
(
None
,
ch
.
params
)
# New channel with an obviously wrong expiration time.
ch
=
channel
.
new_webhook_channel
(
'http://example.com/callback'
,
expiration
=
datetime
.
datetime
(
1965
,
1
,
1
))
self
.
assertEqual
(
0
,
ch
.
expiration
)
# New channel with an expiration time.
ch
=
channel
.
new_webhook_channel
(
'http://example.com/callback'
,
expiration
=
datetime
.
datetime
(
1970
,
1
,
1
,
second
=
5
))
self
.
assertEqual
(
5000
,
ch
.
expiration
)
self
.
assertEqual
(
'http://example.com/callback'
,
ch
.
address
)
self
.
assertEqual
(
None
,
ch
.
params
)
# New channel with an expiration time and params.
ch
=
channel
.
new_webhook_channel
(
'http://example.com/callback'
,
expiration
=
datetime
.
datetime
(
1970
,
1
,
1
,
second
=
5
,
microsecond
=
1000
),
params
=
{
'some'
:
'stuff'
})
self
.
assertEqual
(
5001
,
ch
.
expiration
)
self
.
assertEqual
(
'http://example.com/callback'
,
ch
.
address
)
self
.
assertEqual
({
'some'
:
'stuff'
},
ch
.
params
)
class
TestNotification
(
unittest
.
TestCase
):
def
test_basic
(
self
):
n
=
channel
.
Notification
(
12
,
'sync'
,
'http://example.org'
,
'http://example.org/v1'
)
self
.
assertEqual
(
12
,
n
.
message_number
)
self
.
assertEqual
(
'sync'
,
n
.
state
)
self
.
assertEqual
(
'http://example.org'
,
n
.
resource_uri
)
self
.
assertEqual
(
'http://example.org/v1'
,
n
.
resource_id
)
def
test_notification_from_headers
(
self
):
headers
=
{
'X-GoOG-CHANNEL-ID'
:
'myid'
,
'X-Goog-MESSAGE-NUMBER'
:
'1'
,
'X-Goog-rESOURCE-STATE'
:
'sync'
,
'X-Goog-reSOURCE-URI'
:
'http://example.com/'
,
'X-Goog-resOURCE-ID'
:
'http://example.com/resource_1'
,
}
ch
=
channel
.
Channel
(
'web_hook'
,
'myid'
,
'mytoken'
,
'http://example.org/callback'
,
expiration
=
0
,
params
=
{
'extra'
:
'info'
},
resource_id
=
'the_resource_id'
,
resource_uri
=
'http://example.com/resource_1'
)
# Good test case.
n
=
channel
.
notification_from_headers
(
ch
,
headers
)
self
.
assertEqual
(
'http://example.com/resource_1'
,
n
.
resource_id
)
self
.
assertEqual
(
'http://example.com/'
,
n
.
resource_uri
)
self
.
assertEqual
(
'sync'
,
n
.
state
)
self
.
assertEqual
(
1
,
n
.
message_number
)
# Detect id mismatch.
ch
.
id
=
'different_id'
try
:
n
=
channel
.
notification_from_headers
(
ch
,
headers
)
self
.
fail
(
'Should have raised exception'
)
except
errors
.
InvalidNotificationError
:
pass
# Set the id back to a correct value.
ch
.
id
=
'myid'
Back
|
FazBrowse Home
|
New Git URL