Add push notification subscriptions.
Reviewed in https://codereview.appspot.com/6488087/.
diff --git a/tests/test_push.py b/tests/test_push.py
new file mode 100644
index 0000000..5a42835
--- /dev/null
+++ b/tests/test_push.py
@@ -0,0 +1,272 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Push notifications tests."""
+
+__author__ = 'afshar@google.com (Ali Afshar)'
+
+import unittest
+
+from apiclient import push
+from apiclient import model
+from apiclient import http
+
+
+class ClientTokenGeneratorTest(unittest.TestCase):
+
+ def test_next(self):
+ t = push.new_token()
+ self.assertTrue(t)
+
+
+class ChannelTest(unittest.TestCase):
+
+ def test_creation_noargs(self):
+ c = push.Channel(channel_type='my_channel_type', channel_args={})
+ self.assertEqual('my_channel_type', c.channel_type)
+ self.assertEqual({}, c.channel_args)
+
+ def test_creation_args(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b'})
+ self.assertEqual('my_channel_type', c.channel_type)
+ self.assertEqual({'a':'b'}, c.channel_args)
+
+ def test_as_header_value_noargs(self):
+ c = push.Channel(channel_type='my_channel_type', channel_args={})
+ self.assertEqual('my_channel_type?', c.as_header_value())
+
+ def test_as_header_value_args(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b'})
+ self.assertEqual('my_channel_type?a=b', c.as_header_value())
+
+ def test_as_header_value_args_space(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b c'})
+ self.assertEqual('my_channel_type?a=b+c', c.as_header_value())
+
+ def test_as_header_value_args_escape(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b%c'})
+ self.assertEqual('my_channel_type?a=b%25c', c.as_header_value())
+
+ def test_write_header_noargs(self):
+ c = push.Channel(channel_type='my_channel_type', channel_args={})
+ headers = {}
+ c.write_header(headers)
+ self.assertEqual('my_channel_type?', headers['X-GOOG-SUBSCRIBE'])
+
+ def test_write_header_args(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b'})
+ headers = {}
+ c.write_header(headers)
+ self.assertEqual('my_channel_type?a=b', headers['X-GOOG-SUBSCRIBE'])
+
+ def test_write_header_args_space(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b c'})
+ headers = {}
+ c.write_header(headers)
+ self.assertEqual('my_channel_type?a=b+c', headers['X-GOOG-SUBSCRIBE'])
+
+ def test_write_header_args_escape(self):
+ c = push.Channel(channel_type='my_channel_type',
+ channel_args={'a': 'b%c'})
+ headers = {}
+ c.write_header(headers)
+ self.assertEqual('my_channel_type?a=b%25c', headers['X-GOOG-SUBSCRIBE'])
+
+
+class WebhookChannelTest(unittest.TestCase):
+
+ def test_creation_no_appengine(self):
+ c = push.WebhookChannel('http://example.org')
+ self.assertEqual('web_hook?url=http%3A%2F%2Fexample.org&app_engine=false',
+ c.as_header_value())
+
+ def test_creation_appengine(self):
+ c = push.WebhookChannel('http://example.org', app_engine=True)
+ self.assertEqual('web_hook?url=http%3A%2F%2Fexample.org&app_engine=true',
+ c.as_header_value())
+
+
+class HeadersTest(unittest.TestCase):
+
+ def test_creation(self):
+ h = push.Headers()
+ self.assertEqual('', h[push.SUBSCRIBE])
+
+ def test_items(self):
+ h = push.Headers()
+ h[push.SUBSCRIBE] = 'my_channel_type'
+ self.assertEqual([(push.SUBSCRIBE, 'my_channel_type')], list(h.items()))
+
+ def test_items_non_whitelisted(self):
+ h = push.Headers()
+ def set_bad_header(h=h):
+ h['X-Banana'] = 'my_channel_type'
+ self.assertRaises(ValueError, set_bad_header)
+
+ def test_read(self):
+ h = push.Headers()
+ h.read({'x-goog-subscribe': 'my_channel_type'})
+ self.assertEqual([(push.SUBSCRIBE, 'my_channel_type')], list(h.items()))
+
+ def test_read_non_whitelisted(self):
+ h = push.Headers()
+ h.read({'X-Banana': 'my_channel_type'})
+ self.assertEqual([], list(h.items()))
+
+ def test_write(self):
+ h = push.Headers()
+ h[push.SUBSCRIBE] = 'my_channel_type'
+ headers = {}
+ h.write(headers)
+ self.assertEqual({'x-goog-subscribe': 'my_channel_type'}, headers)
+
+
+class SubscriptionTest(unittest.TestCase):
+
+ def test_create(self):
+ s = push.Subscription()
+ self.assertEqual('', s.client_token)
+
+ def test_create_for_channnel(self):
+ c = push.WebhookChannel('http://example.org')
+ s = push.Subscription.for_channel(c)
+ self.assertTrue(s.client_token)
+ self.assertEqual('web_hook?url=http%3A%2F%2Fexample.org&app_engine=false',
+ s.subscribe)
+
+ def test_create_for_channel_client_token(self):
+ c = push.WebhookChannel('http://example.org')
+ s = push.Subscription.for_channel(c, client_token='my_token')
+ self.assertEqual('my_token', s.client_token)
+ self.assertEqual('web_hook?url=http%3A%2F%2Fexample.org&app_engine=false',
+ s.subscribe)
+
+ def test_subscribe(self):
+ s = push.Subscription()
+ s.headers[push.SUBSCRIBE] = 'my_header'
+ self.assertEqual('my_header', s.subscribe)
+
+ def test_subscription_id(self):
+ s = push.Subscription()
+ s.headers[push.SUBSCRIPTION_ID] = 'my_header'
+ self.assertEqual('my_header', s.subscription_id)
+
+ def test_subscription_id_set(self):
+ c = push.WebhookChannel('http://example.org')
+ s = push.Subscription.for_channel(c)
+ self.assertTrue(s.subscription_id)
+
+ def test_topic_id(self):
+ s = push.Subscription()
+ s.headers[push.TOPIC_ID] = 'my_header'
+ self.assertEqual('my_header', s.topic_id)
+
+ def test_topic_uri(self):
+ s = push.Subscription()
+ s.headers[push.TOPIC_URI] = 'my_header'
+ self.assertEqual('my_header', s.topic_uri)
+
+ def test_client_token(self):
+ s = push.Subscription()
+ s.headers[push.CLIENT_TOKEN] = 'my_header'
+ self.assertEqual('my_header', s.client_token)
+
+ def test_event_type(self):
+ s = push.Subscription()
+ s.headers[push.EVENT_TYPE] = 'my_header'
+ self.assertEqual('my_header', s.event_type)
+
+ def test_unsubscribe(self):
+ s = push.Subscription()
+ s.headers[push.UNSUBSCRIBE] = 'my_header'
+ self.assertEqual('my_header', s.unsubscribe)
+
+ def test_do_subscribe(self):
+ m = model.JsonModel()
+ request = http.HttpRequest(
+ None,
+ m.response,
+ 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
+ method='GET',
+ body='{}',
+ headers={'content-type': 'application/json'})
+ h = http.HttpMockSequence([
+ ({'status': 200,
+ 'X-Goog-Subscription-ID': 'my_subscription'},
+ '{}')])
+ c = push.Channel('my_channel', {})
+ s = push.Subscription.for_request(request, c)
+ request.execute(http=h)
+ self.assertEqual('my_subscription', s.subscription_id)
+
+ def test_subscribe_with_token(self):
+ m = model.JsonModel()
+ request = http.HttpRequest(
+ None,
+ m.response,
+ 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
+ method='GET',
+ body='{}',
+ headers={'content-type': 'application/json'})
+ h = http.HttpMockSequence([
+ ({'status': 200,
+ 'X-Goog-Subscription-ID': 'my_subscription'},
+ '{}')])
+ c = push.Channel('my_channel', {})
+ s = push.Subscription.for_request(request, c, client_token='my_token')
+ request.execute(http=h)
+ self.assertEqual('my_subscription', s.subscription_id)
+ self.assertEqual('my_token', s.client_token)
+
+ def test_verify_good_token(self):
+ s = push.Subscription()
+ s.headers['X-Goog-Client-Token'] = '123'
+ notification_headers = {'x-goog-client-token': '123'}
+ self.assertTrue(s.verify(notification_headers))
+
+ def test_verify_bad_token(self):
+ s = push.Subscription()
+ s.headers['X-Goog-Client-Token'] = '321'
+ notification_headers = {'x-goog-client-token': '123'}
+ self.assertFalse(s.verify(notification_headers))
+
+ def test_request_is_post(self):
+ m = model.JsonModel()
+ request = http.HttpRequest(
+ None,
+ m.response,
+ 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
+ method='GET',
+ body='{}',
+ headers={'content-type': 'application/json'})
+ c = push.Channel('my_channel', {})
+ push.Subscription.for_request(request, c)
+ self.assertEqual('POST', request.method)
+
+ def test_non_get_error(self):
+ m = model.JsonModel()
+ request = http.HttpRequest(
+ None,
+ m.response,
+ 'https://www.googleapis.com/someapi/v1/collection/?foo=bar',
+ method='POST',
+ body='{}',
+ headers={'content-type': 'application/json'})
+ c = push.Channel('my_channel', {})
+ self.assertRaises(push.InvalidSubscriptionRequestError,
+ push.Subscription.for_request, request, c)