blob: f6a6d55662714f5dd0f5f10101a279a0221169a5 [file] [log] [blame]
Joe Gregorio1a5e30e2013-06-25 15:35:47 -04001"""Notification channels tests."""
2
3__author__ = 'jcgregorio@google.com (Joe Gregorio)'
4
5import unittest
6import datetime
7
8from apiclient import channel
9from apiclient import errors
10
11
12class TestChannel(unittest.TestCase):
13 def test_basic(self):
14 ch = channel.Channel('web_hook', 'myid', 'mytoken',
15 'http://example.org/callback',
16 expiration=0,
17 params={'extra': 'info'},
18 resource_id='the_resource_id',
19 resource_uri='http://example.com/resource_1')
20
21 # Converting to a body.
22 body = ch.body()
23 self.assertEqual('http://example.org/callback', body['address'])
24 self.assertEqual('myid', body['id'])
25 self.assertEqual('missing', body.get('expiration', 'missing'))
26 self.assertEqual('info', body['params']['extra'])
27 self.assertEqual('the_resource_id', body['resourceId'])
28 self.assertEqual('http://example.com/resource_1', body['resourceUri'])
29 self.assertEqual('web_hook', body['type'])
30
31 # Converting to a body with expiration set.
32 ch.expiration = 1
33 body = ch.body()
34 self.assertEqual(1, body.get('expiration', 'missing'))
35
36 # Converting to a body after updating with a response body.
37 ch.update({
38 'resourceId': 'updated_res_id',
39 'resourceUri': 'updated_res_uri',
40 'some_random_parameter': 2,
41 })
42
43 body = ch.body()
44 self.assertEqual('http://example.org/callback', body['address'])
45 self.assertEqual('myid', body['id'])
46 self.assertEqual(1, body.get('expiration', 'missing'))
47 self.assertEqual('info', body['params']['extra'])
48 self.assertEqual('updated_res_id', body['resourceId'])
49 self.assertEqual('updated_res_uri', body['resourceUri'])
50 self.assertEqual('web_hook', body['type'])
51
52 def test_new_webhook_channel(self):
53 ch = channel.new_webhook_channel('http://example.com/callback')
54 self.assertEqual(0, ch.expiration)
55 self.assertEqual('http://example.com/callback', ch.address)
56 self.assertEqual(None, ch.params)
57
58 # New channel with an obviously wrong expiration time.
59 ch = channel.new_webhook_channel(
60 'http://example.com/callback',
61 expiration=datetime.datetime(1965, 1, 1))
62 self.assertEqual(0, ch.expiration)
63
64 # New channel with an expiration time.
65 ch = channel.new_webhook_channel(
66 'http://example.com/callback',
67 expiration=datetime.datetime(1970, 1, 1, second=5))
68 self.assertEqual(5000, ch.expiration)
69 self.assertEqual('http://example.com/callback', ch.address)
70 self.assertEqual(None, ch.params)
71
72 # New channel with an expiration time and params.
73 ch = channel.new_webhook_channel(
74 'http://example.com/callback',
75 expiration=datetime.datetime(1970, 1, 1, second=5, microsecond=1000),
76 params={'some':'stuff'})
77 self.assertEqual(5001, ch.expiration)
78 self.assertEqual('http://example.com/callback', ch.address)
79 self.assertEqual({'some': 'stuff'}, ch.params)
80
81
82class TestNotification(unittest.TestCase):
83 def test_basic(self):
84 n = channel.Notification(12, 'sync', 'http://example.org',
85 'http://example.org/v1')
86
87 self.assertEqual(12, n.message_number)
88 self.assertEqual('sync', n.state)
89 self.assertEqual('http://example.org', n.resource_uri)
90 self.assertEqual('http://example.org/v1', n.resource_id)
91
92 def test_notification_from_headers(self):
93 headers = {
94 'X-GoOG-CHANNEL-ID': 'myid',
95 'X-Goog-MESSAGE-NUMBER': '1',
96 'X-Goog-rESOURCE-STATE': 'sync',
97 'X-Goog-reSOURCE-URI': 'http://example.com/',
98 'X-Goog-resOURCE-ID': 'http://example.com/resource_1',
99 }
100
101 ch = channel.Channel('web_hook', 'myid', 'mytoken',
102 'http://example.org/callback',
103 expiration=0,
104 params={'extra': 'info'},
105 resource_id='the_resource_id',
106 resource_uri='http://example.com/resource_1')
107
108 # Good test case.
109 n = channel.notification_from_headers(ch, headers)
110 self.assertEqual('http://example.com/resource_1', n.resource_id)
111 self.assertEqual('http://example.com/', n.resource_uri)
112 self.assertEqual('sync', n.state)
113 self.assertEqual(1, n.message_number)
114
115 # Detect id mismatch.
116 ch.id = 'different_id'
117 try:
118 n = channel.notification_from_headers(ch, headers)
119 self.fail('Should have raised exception')
120 except errors.InvalidNotificationError:
121 pass
122
123 # Set the id back to a correct value.
124 ch.id = 'myid'