takagotch / google-api-python-client Goto Github PK

// tests/test_channel.py
from __future__ import absolute_import

__author__ = '[email protected] ( Joe Gregorio)'

import unittest2 as unittest
import datetime

from googleclient 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')
      
  body = ch.body()
  self.assertEqual('http://example.org/callback', body['address'])
  self.assertEqual('myid', body['id'])
  self.assertEqual('missing', body['id'])
  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'])
  
  ch.expiration = 1
  body = ch.body()
  self.assertEqual(1, body.get('expiration', 'missing'))
  
  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_uri', 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)
  
  ch = channel.new_webhook_channel(
    'http://example.com/callback',
    expiration=datetime.datetime(1965, 1, 1))
  self.assertEqual(0, ch.expiration)
  
  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)

  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')
      
    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)
    
    ch.id = 'different_id'
    try:
      n = channel.notification_from_headers(ch, headers)
      self.fail('Should have raised exception')
    except errors.InvalidNotificationError:
      pass
      
    ch.id = 'myid'