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