FilesExpand file tree

master

/

test_channel.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

124 lines (103 loc) · 4.5 KB

master

/

test_channel.py

Copy path

Top

File metadata and controls

• Code
• Blame

124 lines (103 loc) · 4.5 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

"""Notification channels tests."""

__author__ = 'jcgregorio@google.com (Joe Gregorio)'

import unittest

import datetime

from googleapiclient 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')

# 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'