openstack/deb-python-oauth2client
Code Issues Proposed changes

A new push implementation. This updates us to the latest revision of the push

Browse Source 
API which has a much simpler interface. Look in the module doc string for how
this API would be used.

Reviewed in https://codereview.appspot.com/9885043/.
This commit is contained in:
Joe Gregorio
2013-06-25 15:35:47 -04:00
parent d60d1e7de7
commit 1a5e30e432
5 changed files with 412 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files

285
apiclient/channel.py Normal file
View File

@@ -0,0 +1,285 @@
"""Channel notifications support.
Classes and functions to support channel subscriptions and notifications
on those channels.
Notes:
- This code is based on experimental APIs and is subject to change.
- Notification does not do deduplication of notification ids, that's up to
the receiver.
- Storing the Channel between calls is up to the caller.
Example setting up a channel:
# Create a new channel that gets notifications via webhook.
channel = new_webhook_channel("https://example.com/my_web_hook")
# Store the channel, keyed by 'channel.id'. Store it before calling the
# watch method because notifications may start arriving before the watch
# method returns.
...
resp = service.objects().watchAll(
bucket="some_bucket_id", body=channel.body()).execute()
channel.update(resp)
# Store the channel, keyed by 'channel.id'. Store it after being updated
# since the resource_id value will now be correct, and that's needed to
# stop a subscription.
...
An example Webhook implementation using webapp2. Note that webapp2 puts
headers in a case insensitive dictionary, as headers aren't guaranteed to
always be upper case.
id = self.request.headers[X_GOOG_CHANNEL_ID]
# Retrieve the channel by id.
channel = ...
# Parse notification from the headers, including validating the id.
n = notification_from_headers(channel, self.request.headers)
# Do app specific stuff with the notification here.
if n.resource_state == 'sync':
# Code to handle sync state.
elif n.resource_state == 'exists':
# Code to handle the exists state.
elif n.resource_state == 'not_exists':
# Code to handle the not exists state.
Example of unsubscribing.
service.channels().stop(channel.body())
"""
import datetime
import uuid
from apiclient import errors
from oauth2client import util
# The unix time epoch starts at midnight 1970.
EPOCH = datetime.datetime.utcfromtimestamp(0)
# Map the names of the parameters in the JSON channel description to
# the parameter names we use in the Channel class.
CHANNEL_PARAMS = {
'address': 'address',
'id': 'id',
'expiration': 'expiration',
'params': 'params',
'resourceId': 'resource_id',
'resourceUri': 'resource_uri',
'type': 'type',
'token': 'token',
}
X_GOOG_CHANNEL_ID = 'X-GOOG-CHANNEL-ID'
X_GOOG_MESSAGE_NUMBER = 'X-GOOG-MESSAGE-NUMBER'
X_GOOG_RESOURCE_STATE = 'X-GOOG-RESOURCE-STATE'
X_GOOG_RESOURCE_URI = 'X-GOOG-RESOURCE-URI'
X_GOOG_RESOURCE_ID = 'X-GOOG-RESOURCE-ID'
def _upper_header_keys(headers):
new_headers = {}
for k, v in headers.iteritems():
new_headers[k.upper()] = v
return new_headers
class Notification(object):
"""A Notification from a Channel.
Notifications are not usually constructed directly, but are returned
from functions like notification_from_headers().
Attributes:
message_number: int, The unique id number of this notification.
state: str, The state of the resource being monitored.
uri: str, The address of the resource being monitored.
resource_id: str, The unique identifier of the version of the resource at
this event.
"""
@util.positional(5)
def __init__(self, message_number, state, resource_uri, resource_id):
"""Notification constructor.
Args:
message_number: int, The unique id number of this notification.
state: str, The state of the resource being monitored. Can be one
of "exists", "not_exists", or "sync".
resource_uri: str, The address of the resource being monitored.
resource_id: str, The identifier of the watched resource.
"""
self.message_number = message_number
self.state = state
self.resource_uri = resource_uri
self.resource_id = resource_id
class Channel(object):
"""A Channel for notifications.
Usually not constructed directly, instead it is returned from helper
functions like new_webhook_channel().
Attributes:
type: str, The type of delivery mechanism used by this channel. For
example, 'web_hook'.
id: str, A UUID for the channel.
token: str, An arbitrary string associated with the channel that
is delivered to the target address with each event delivered
over this channel.
address: str, The address of the receiving entity where events are
delivered. Specific to the channel type.
expiration: int, The time, in milliseconds from the epoch, when this
channel will expire.
params: dict, A dictionary of string to string, with additional parameters
controlling delivery channel behavior.
resource_id: str, An opaque id that identifies the resource that is
being watched. Stable across different API versions.
resource_uri: str, The canonicalized ID of the watched resource.
"""
@util.positional(5)
def __init__(self, type, id, token, address, expiration=None,
params=None, resource_id="", resource_uri=""):
"""Create a new Channel.
In user code, this Channel constructor will not typically be called
manually since there are functions for creating channels for each specific
type with a more customized set of arguments to pass.
Args:
type: str, The type of delivery mechanism used by this channel. For
example, 'web_hook'.
id: str, A UUID for the channel.
token: str, An arbitrary string associated with the channel that
is delivered to the target address with each event delivered
over this channel.
address: str, The address of the receiving entity where events are
delivered. Specific to the channel type.
expiration: int, The time, in milliseconds from the epoch, when this
channel will expire.
params: dict, A dictionary of string to string, with additional parameters
controlling delivery channel behavior.
resource_id: str, An opaque id that identifies the resource that is
being watched. Stable across different API versions.
resource_uri: str, The canonicalized ID of the watched resource.
"""
self.type = type
self.id = id
self.token = token
self.address = address
self.expiration = expiration
self.params = params
self.resource_id = resource_id
self.resource_uri = resource_uri
def body(self):
"""Build a body from the Channel.
Constructs a dictionary that's appropriate for passing into watch()
methods as the value of body argument.
Returns:
A dictionary representation of the channel.
"""
result = {
'id': self.id,
'token': self.token,
'type': self.type,
'address': self.address
}
if self.params:
result['params'] = self.params
if self.resource_id:
result['resourceId'] = self.resource_id
if self.resource_uri:
result['resourceUri'] = self.resource_uri
if self.expiration:
result['expiration'] = self.expiration
return result
def update(self, resp):
"""Update a channel with information from the response of watch().
When a request is sent to watch() a resource, the response returned
from the watch() request is a dictionary with updated channel information,
such as the resource_id, which is needed when stopping a subscription.
Args:
resp: dict, The response from a watch() method.
"""
for json_name, param_name in CHANNEL_PARAMS.iteritems():
value = resp.get(json_name)
if value is not None:
setattr(self, param_name, value)
def notification_from_headers(channel, headers):
"""Parse a notification from the webhook request headers, validate
the notification, and return a Notification object.
Args:
channel: Channel, The channel that the notification is associated with.
headers: dict, A dictionary like object that contains the request headers
from the webhook HTTP request.
Returns:
A Notification object.
Raises:
errors.InvalidNotificationError if the notification is invalid.
ValueError if the X-GOOG-MESSAGE-NUMBER can't be converted to an int.
"""
headers = _upper_header_keys(headers)
channel_id = headers[X_GOOG_CHANNEL_ID]
if channel.id != channel_id:
raise errors.InvalidNotificationError(
'Channel id mismatch: %s != %s' % (channel.id, channel_id))
else:
message_number = int(headers[X_GOOG_MESSAGE_NUMBER])
state = headers[X_GOOG_RESOURCE_STATE]
resource_uri = headers[X_GOOG_RESOURCE_URI]
resource_id = headers[X_GOOG_RESOURCE_ID]
return Notification(message_number, state, resource_uri, resource_id)
@util.positional(2)
def new_webhook_channel(url, token=None, expiration=None, params=None):
"""Create a new webhook Channel.
Args:
url: str, URL to post notifications to.
token: str, An arbitrary string associated with the channel that
is delivered to the target address with each notification delivered
over this channel.
expiration: datetime.datetime, A time in the future when the channel
should expire. Can also be None if the subscription should use the
default expiration. Note that different services may have different
limits on how long a subscription lasts. Check the response from the
watch() method to see the value the service has set for an expiration
time.
params: dict, Extra parameters to pass on channel creation. Currently
not used for webhook channels.
"""
expiration_ms = 0
if expiration:
delta = expiration - EPOCH
expiration_ms = delta.microseconds/1000 + (
delta.seconds + delta.days*24*3600)*1000
if expiration_ms < 0:
expiration_ms = 0
return Channel('web_hook', str(uuid.uuid4()),
token, url, expiration=expiration_ms,
params=params)

3
apiclient/errors.py
View File

@@ -102,6 +102,9 @@ class InvalidChunkSizeError(Error):
"""The given chunksize is not valid."""
pass
class InvalidNotificationError(Error):
"""The channel Notification is invalid."""
pass
class BatchError(HttpError):
"""Error occured during batch operations."""

274
apiclient/push.py
View File

@@ -1,274 +0,0 @@
# 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 support.
This code is based on experimental APIs and is subject to change.
"""
__author__ = 'afshar@google.com (Ali Afshar)'
import binascii
import collections
import os
import urllib
SUBSCRIBE = 'X-GOOG-SUBSCRIBE'
SUBSCRIPTION_ID = 'X-GOOG-SUBSCRIPTION-ID'
TOPIC_ID = 'X-GOOG-TOPIC-ID'
TOPIC_URI = 'X-GOOG-TOPIC-URI'
CLIENT_TOKEN = 'X-GOOG-CLIENT-TOKEN'
EVENT_TYPE = 'X-GOOG-EVENT-TYPE'
UNSUBSCRIBE = 'X-GOOG-UNSUBSCRIBE'
class InvalidSubscriptionRequestError(ValueError):
"""The request cannot be subscribed."""
def new_token():
"""Gets a random token for use as a client_token in push notifications.
Returns:
str, a new random token.
"""
return binascii.hexlify(os.urandom(32))
class Channel(object):
"""Base class for channel types."""
def __init__(self, channel_type, channel_args):
"""Create a new Channel.
You probably won't need to create this channel manually, since there are
subclassed Channel for each specific type with a more customized set of
arguments to pass. However, you may wish to just create it manually here.
Args:
channel_type: str, the type of channel.
channel_args: dict, arguments to pass to the channel.
"""
self.channel_type = channel_type
self.channel_args = channel_args
def as_header_value(self):
"""Create the appropriate header for this channel.
Returns:
str encoded channel description suitable for use as a header.
"""
return '%s?%s' % (self.channel_type, urllib.urlencode(self.channel_args))
def write_header(self, headers):
"""Write the appropriate subscribe header to a headers dict.
Args:
headers: dict, headers to add subscribe header to.
"""
headers[SUBSCRIBE] = self.as_header_value()
class WebhookChannel(Channel):
"""Channel for registering web hook notifications."""
def __init__(self, url, app_engine=False):
"""Create a new WebhookChannel
Args:
url: str, URL to post notifications to.
app_engine: bool, default=False, whether the destination for the
notifications is an App Engine application.
"""
super(WebhookChannel, self).__init__(
channel_type='web_hook',
channel_args={
'url': url,
'app_engine': app_engine and 'true' or 'false',
}
)
class Headers(collections.defaultdict):
"""Headers for managing subscriptions."""
ALL_HEADERS = set([SUBSCRIBE, SUBSCRIPTION_ID, TOPIC_ID, TOPIC_URI,
CLIENT_TOKEN, EVENT_TYPE, UNSUBSCRIBE])
def __init__(self):
"""Create a new subscription configuration instance."""
collections.defaultdict.__init__(self, str)
def __setitem__(self, key, value):
"""Set a header value, ensuring the key is an allowed value.
Args:
key: str, the header key.
value: str, the header value.
Raises:
ValueError if key is not one of the accepted headers.
"""
normal_key = self._normalize_key(key)
if normal_key not in self.ALL_HEADERS:
raise ValueError('Header name must be one of %s.' % self.ALL_HEADERS)
else:
return collections.defaultdict.__setitem__(self, normal_key, value)
def __getitem__(self, key):
"""Get a header value, normalizing the key case.
Args:
key: str, the header key.
Returns:
String header value.
Raises:
KeyError if the key is not one of the accepted headers.
"""
normal_key = self._normalize_key(key)
if normal_key not in self.ALL_HEADERS:
raise ValueError('Header name must be one of %s.' % self.ALL_HEADERS)
else:
return collections.defaultdict.__getitem__(self, normal_key)
def _normalize_key(self, key):
"""Normalize a header name for use as a key."""
return key.upper()
def items(self):
"""Generator for each header."""
for header in self.ALL_HEADERS:
value = self[header]
if value:
yield header, value
def write(self, headers):
"""Applies the subscription headers.
Args:
headers: dict of headers to insert values into.
"""
for header, value in self.items():
headers[header.lower()] = value
def read(self, headers):
"""Read from headers.
Args:
headers: dict of headers to read from.
"""
for header in self.ALL_HEADERS:
if header.lower() in headers:
self[header] = headers[header.lower()]
class Subscription(object):
"""Information about a subscription."""
def __init__(self):
"""Create a new Subscription."""
self.headers = Headers()
@classmethod
def for_request(cls, request, channel, client_token=None):
"""Creates a subscription and attaches it to a request.
Args:
request: An http.HttpRequest to modify for making a subscription.
channel: A apiclient.push.Channel describing the subscription to
create.
client_token: (optional) client token to verify the notification.
Returns:
New subscription object.
"""
subscription = cls.for_channel(channel=channel, client_token=client_token)
subscription.headers.write(request.headers)
if request.method != 'GET':
raise InvalidSubscriptionRequestError(
'Can only subscribe to requests which are GET.')
request.method = 'POST'
def _on_response(response, subscription=subscription):
"""Called with the response headers. Reads the subscription headers."""
subscription.headers.read(response)
request.add_response_callback(_on_response)
return subscription
@classmethod
def for_channel(cls, channel, client_token=None):
"""Alternate constructor to create a subscription from a channel.
Args:
channel: A apiclient.push.Channel describing the subscription to
create.
client_token: (optional) client token to verify the notification.
Returns:
New subscription object.
"""
subscription = cls()
channel.write_header(subscription.headers)
if client_token is None:
client_token = new_token()
subscription.headers[SUBSCRIPTION_ID] = new_token()
subscription.headers[CLIENT_TOKEN] = client_token
return subscription
def verify(self, headers):
"""Verifies that a webhook notification has the correct client_token.
Args:
headers: dict of request headers for a push notification.
Returns:
Boolean value indicating whether the notification is verified.
"""
new_subscription = Subscription()
new_subscription.headers.read(headers)
return new_subscription.client_token == self.client_token
@property
def subscribe(self):
"""Subscribe header value."""
return self.headers[SUBSCRIBE]
@property
def subscription_id(self):
"""Subscription ID header value."""
return self.headers[SUBSCRIPTION_ID]
@property
def topic_id(self):
"""Topic ID header value."""
return self.headers[TOPIC_ID]
@property
def topic_uri(self):
"""Topic URI header value."""
return self.headers[TOPIC_URI]
@property
def client_token(self):
"""Client Token header value."""
return self.headers[CLIENT_TOKEN]
@property
def event_type(self):
"""Event Type header value."""
return self.headers[EVENT_TYPE]
@property
def unsubscribe(self):
"""Unsuscribe header value."""
return self.headers[UNSUBSCRIBE]

124
tests/test_channel.py Normal file
View File

@@ -0,0 +1,124 @@
"""Notification channels tests."""
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
import unittest
import datetime
from apiclient import channel
from apiclient 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'

277
tests/test_push.py
View File

@@ -1,277 +0,0 @@
# 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
from test_discovery import assertUrisEqual
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')
assertUrisEqual(self,
'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)
assertUrisEqual(self,
'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)
assertUrisEqual(self,
'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)
assertUrisEqual(self,
'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)