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