1# Licensed under the Apache License, Version 2.0 (the "License"); you may
2# not use this file except in compliance with the License. You may obtain
3# a copy of the License at
4#
5#      http://www.apache.org/licenses/LICENSE-2.0
6#
7# Unless required by applicable law or agreed to in writing, software
8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10# License for the specific language governing permissions and limitations
11# under the License.
12
13import copy
14from unittest import mock
15import uuid
16
17from openstack.message.v2 import subscription
18from openstack.tests.unit import base
19
20
21FAKE1 = {
22    "age": 1632,
23    "id": "576b54963990b48c644bb7e7",
24    "subscriber": "http://10.229.49.117:5679",
25    "subscription_id": "576b54963990b48c644bb7e7",
26    "source": "test",
27    "ttl": 3600,
28    "options": {
29        "name": "test"
30    },
31    "queue_name": "queue1"
32}
33
34
35FAKE2 = {
36    "age": 1632,
37    "id": "576b54963990b48c644bb7e7",
38    "subscriber": "http://10.229.49.117:5679",
39    "subscription_id": "576b54963990b48c644bb7e7",
40    "source": "test",
41    "ttl": 3600,
42    "options": {
43        "name": "test"
44    },
45    "queue_name": "queue1",
46    "client_id": "OLD_CLIENT_ID",
47    "project_id": "OLD_PROJECT_ID"
48}
49
50
51class TestSubscription(base.TestCase):
52    def test_basic(self):
53        sot = subscription.Subscription()
54        self.assertEqual("subscriptions", sot.resources_key)
55        self.assertEqual("/queues/%(queue_name)s/subscriptions", sot.base_path)
56        self.assertTrue(sot.allow_create)
57        self.assertTrue(sot.allow_fetch)
58        self.assertTrue(sot.allow_delete)
59        self.assertTrue(sot.allow_list)
60
61    def test_make_it(self):
62        sot = subscription.Subscription.new(**FAKE2)
63        self.assertEqual(FAKE2["age"], sot.age)
64        self.assertEqual(FAKE2["id"], sot.id)
65        self.assertEqual(FAKE2["options"], sot.options)
66        self.assertEqual(FAKE2["source"], sot.source)
67        self.assertEqual(FAKE2["subscriber"], sot.subscriber)
68        self.assertEqual(FAKE2["subscription_id"], sot.subscription_id)
69        self.assertEqual(FAKE2["ttl"], sot.ttl)
70        self.assertEqual(FAKE2["queue_name"], sot.queue_name)
71        self.assertEqual(FAKE2["client_id"], sot.client_id)
72        self.assertEqual(FAKE2["project_id"], sot.project_id)
73
74    @mock.patch.object(uuid, "uuid4")
75    def test_create(self, mock_uuid):
76        sess = mock.Mock()
77        resp = mock.Mock()
78        sess.post.return_value = resp
79        sess.get_project_id.return_value = "NEW_PROJECT_ID"
80        mock_uuid.return_value = "NEW_CLIENT_ID"
81        FAKE = copy.deepcopy(FAKE1)
82
83        sot = subscription.Subscription(**FAKE1)
84        sot._translate_response = mock.Mock()
85        res = sot.create(sess)
86
87        url = "/queues/%(queue)s/subscriptions" % {
88            "queue": FAKE.pop("queue_name")}
89        headers = {"Client-ID": "NEW_CLIENT_ID",
90                   "X-PROJECT-ID": "NEW_PROJECT_ID"}
91        sess.post.assert_called_once_with(url,
92                                          headers=headers, json=FAKE)
93        sess.get_project_id.assert_called_once_with()
94        self.assertEqual(sot, res)
95
96    def test_create_client_id_project_id_exist(self):
97        sess = mock.Mock()
98        resp = mock.Mock()
99        sess.post.return_value = resp
100        FAKE = copy.deepcopy(FAKE2)
101
102        sot = subscription.Subscription(**FAKE2)
103        sot._translate_response = mock.Mock()
104        res = sot.create(sess)
105
106        url = "/queues/%(queue)s/subscriptions" % {
107            "queue": FAKE.pop("queue_name")}
108        headers = {"Client-ID": FAKE.pop("client_id"),
109                   "X-PROJECT-ID": FAKE.pop("project_id")}
110        sess.post.assert_called_once_with(url,
111                                          headers=headers, json=FAKE)
112        self.assertEqual(sot, res)
113
114    @mock.patch.object(uuid, "uuid4")
115    def test_get(self, mock_uuid):
116        sess = mock.Mock()
117        resp = mock.Mock()
118        sess.get.return_value = resp
119        sess.get_project_id.return_value = "NEW_PROJECT_ID"
120        mock_uuid.return_value = "NEW_CLIENT_ID"
121
122        sot = subscription.Subscription(**FAKE1)
123        sot._translate_response = mock.Mock()
124        res = sot.fetch(sess)
125
126        url = "queues/%(queue)s/subscriptions/%(subscription)s" % {
127            "queue": FAKE1["queue_name"], "subscription": FAKE1["id"]}
128        headers = {"Client-ID": "NEW_CLIENT_ID",
129                   "X-PROJECT-ID": "NEW_PROJECT_ID"}
130        sess.get.assert_called_with(url,
131                                    headers=headers)
132        sess.get_project_id.assert_called_once_with()
133        sot._translate_response.assert_called_once_with(resp)
134        self.assertEqual(sot, res)
135
136    def test_get_client_id_project_id_exist(self):
137        sess = mock.Mock()
138        resp = mock.Mock()
139        sess.get.return_value = resp
140
141        sot = subscription.Subscription(**FAKE2)
142        sot._translate_response = mock.Mock()
143        res = sot.fetch(sess)
144
145        url = "queues/%(queue)s/subscriptions/%(subscription)s" % {
146            "queue": FAKE2["queue_name"], "subscription": FAKE2["id"]}
147        headers = {"Client-ID": "OLD_CLIENT_ID",
148                   "X-PROJECT-ID": "OLD_PROJECT_ID"}
149        sess.get.assert_called_with(url,
150                                    headers=headers)
151        sot._translate_response.assert_called_once_with(resp)
152        self.assertEqual(sot, res)
153
154    @mock.patch.object(uuid, "uuid4")
155    def test_delete(self, mock_uuid):
156        sess = mock.Mock()
157        resp = mock.Mock()
158        sess.delete.return_value = resp
159        sess.get_project_id.return_value = "NEW_PROJECT_ID"
160        mock_uuid.return_value = "NEW_CLIENT_ID"
161
162        sot = subscription.Subscription(**FAKE1)
163        sot._translate_response = mock.Mock()
164        sot.delete(sess)
165
166        url = "queues/%(queue)s/subscriptions/%(subscription)s" % {
167            "queue": FAKE1["queue_name"], "subscription": FAKE1["id"]}
168        headers = {"Client-ID": "NEW_CLIENT_ID",
169                   "X-PROJECT-ID": "NEW_PROJECT_ID"}
170        sess.delete.assert_called_with(url,
171                                       headers=headers)
172        sess.get_project_id.assert_called_once_with()
173        sot._translate_response.assert_called_once_with(resp, has_body=False)
174
175    def test_delete_client_id_project_id_exist(self):
176        sess = mock.Mock()
177        resp = mock.Mock()
178        sess.delete.return_value = resp
179
180        sot = subscription.Subscription(**FAKE2)
181        sot._translate_response = mock.Mock()
182        sot.delete(sess)
183
184        url = "queues/%(queue)s/subscriptions/%(subscription)s" % {
185            "queue": FAKE2["queue_name"], "subscription": FAKE2["id"]}
186        headers = {"Client-ID": "OLD_CLIENT_ID",
187                   "X-PROJECT-ID": "OLD_PROJECT_ID"}
188        sess.delete.assert_called_with(url,
189                                       headers=headers)
190        sot._translate_response.assert_called_once_with(resp, has_body=False)
191