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