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'
|
