Coverage for manila/tests/share/drivers/quobyte/test_jsonrpc.py: 100%
93 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright (c) 2015 Quobyte, Inc.
2# All Rights Reserved.
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16import tempfile
17import time
18from unittest import mock
20import requests
21from requests import auth
22from requests import exceptions
24from manila import exception
25from manila.share.drivers.quobyte import jsonrpc
26from manila import test
29class FakeResponse(object):
30 def __init__(self, status, body):
31 self.status_code = status
32 self.reason = "HTTP reason"
33 self.body = body
34 self.text = str(body)
36 def json(self):
37 return self.body
40class QuobyteJsonRpcTestCase(test.TestCase):
42 def setUp(self):
43 super(QuobyteJsonRpcTestCase, self).setUp()
44 self.rpc = jsonrpc.JsonRpc(url="http://test",
45 user_credentials=("me", "team"))
46 self.mock_object(time, 'sleep')
48 @mock.patch.object(requests, 'post',
49 return_value=FakeResponse(200, {"result": "yes"}))
50 def test_request_generation_and_basic_auth(self, req_get_mock):
51 self.rpc.call('method', {'param': 'value'})
53 req_get_mock.assert_called_once_with(
54 url='http://test',
55 auth=auth.HTTPBasicAuth("me", "team"),
56 json=mock.ANY,
57 timeout=60)
59 def test_jsonrpc_init_with_ca(self):
60 foofile = tempfile.TemporaryFile()
61 fake_url = "https://foo.bar/"
62 fake_credentials = ('fakeuser', 'fakepwd')
63 fake_cert_file = tempfile.TemporaryFile()
64 fake_key_file = tempfile.TemporaryFile()
65 self.rpc = jsonrpc.JsonRpc(url=fake_url,
66 user_credentials=fake_credentials,
67 ca_file=foofile,
68 key_file=fake_key_file,
69 cert_file=fake_cert_file)
71 self.assertEqual("https", self.rpc._url_scheme)
72 self.assertEqual(fake_url, self.rpc._url)
73 self.assertEqual(foofile, self.rpc._ca_file)
74 self.assertEqual(fake_cert_file, self.rpc._cert_file)
75 self.assertEqual(fake_key_file, self.rpc._key_file)
77 @mock.patch.object(jsonrpc.LOG, "warning")
78 def test_jsonrpc_init_without_ca(self, mock_warning):
79 self.rpc = jsonrpc.JsonRpc("https://foo.bar/",
80 ('fakeuser', 'fakepwd'),
81 None)
83 mock_warning.assert_called_once_with(
84 "Will not verify the server certificate of the API service"
85 " because the CA certificate is not available.")
87 def test_jsonrpc_init_no_ssl(self):
88 self.rpc = jsonrpc.JsonRpc("http://foo.bar/",
89 ('fakeuser', 'fakepwd'))
91 self.assertEqual("http", self.rpc._url_scheme)
93 @mock.patch.object(requests, "post",
94 return_value=FakeResponse(
95 200, {"result": "Sweet gorilla of Manila"}))
96 def test_successful_call(self, mock_req_get):
97 result = self.rpc.call('method', {'param': 'value'})
99 mock_req_get.assert_called_once_with(
100 url=self.rpc._url,
101 json=mock.ANY, # not checking here as of undefined order in dict
102 auth=self.rpc._credentials,
103 timeout=60)
104 self.assertEqual("Sweet gorilla of Manila", result)
106 @mock.patch.object(requests, "post",
107 return_value=FakeResponse(
108 200, {"result": "Sweet gorilla of Manila"}))
109 def test_https_call_with_cert(self, mock_req_get):
110 fake_cert_file = tempfile.TemporaryFile()
111 fake_key_file = tempfile.TemporaryFile()
112 self.rpc = jsonrpc.JsonRpc(url="https://test",
113 user_credentials=("me", "team"),
114 cert_file=fake_cert_file,
115 key_file=fake_key_file)
117 result = self.rpc.call('method', {'param': 'value'})
119 mock_req_get.assert_called_once_with(
120 url=self.rpc._url,
121 json=mock.ANY, # not checking here as of undefined order in dict
122 auth=self.rpc._credentials,
123 verify=False,
124 cert=(fake_cert_file, fake_key_file),
125 timeout=60)
126 self.assertEqual("Sweet gorilla of Manila", result)
128 @mock.patch.object(requests, "post",
129 return_value=FakeResponse(
130 200, {"result": "Sweet gorilla of Manila"}))
131 def test_https_call_verify(self, mock_req_get):
132 fake_ca_file = tempfile.TemporaryFile()
133 self.rpc = jsonrpc.JsonRpc(url="https://test",
134 user_credentials=("me", "team"),
135 ca_file=fake_ca_file)
137 result = self.rpc.call('method', {'param': 'value'})
139 mock_req_get.assert_called_once_with(
140 url=self.rpc._url,
141 json=mock.ANY, # not checking here as of undefined order in dict
142 auth=self.rpc._credentials,
143 verify=fake_ca_file,
144 timeout=60)
145 self.assertEqual("Sweet gorilla of Manila", result)
147 @mock.patch.object(jsonrpc.JsonRpc, "_checked_for_application_error",
148 return_value="Sweet gorilla of Manila")
149 @mock.patch.object(requests, "post",
150 return_value=FakeResponse(
151 200, {"result": "Sweet gorilla of Manila"}))
152 def test_https_call_verify_expected_error(self, mock_req_get, mock_check):
153 fake_ca_file = tempfile.TemporaryFile()
154 self.rpc = jsonrpc.JsonRpc(url="https://test",
155 user_credentials=("me", "team"),
156 ca_file=fake_ca_file)
158 result = self.rpc.call('method', {'param': 'value'},
159 expected_errors=[42])
161 mock_req_get.assert_called_once_with(
162 url=self.rpc._url,
163 json=mock.ANY, # not checking here as of undefined order in dict
164 auth=self.rpc._credentials,
165 verify=fake_ca_file,
166 timeout=60)
167 mock_check.assert_called_once_with(
168 {'result': 'Sweet gorilla of Manila'}, [42])
169 self.assertEqual("Sweet gorilla of Manila", result)
171 @mock.patch.object(requests, "post", side_effect=exceptions.HTTPError)
172 def test_jsonrpc_call_http_exception(self, req_get_mock):
173 self.assertRaises(exceptions.HTTPError,
174 self.rpc.call,
175 'method', {'param': 'value'})
176 req_get_mock.assert_called_once_with(
177 url=self.rpc._url,
178 json=mock.ANY, # not checking here as of undefined order in dict
179 auth=self.rpc._credentials,
180 timeout=60)
182 @mock.patch.object(requests, "post",
183 return_value=FakeResponse(
184 200,
185 {"error": {"code": 28, "message": "text"}}))
186 def test_application_error(self, req_get_mock):
187 self.assertRaises(exception.QBRpcException,
188 self.rpc.call, 'method', {'param': 'value'})
189 req_get_mock.assert_called_once_with(
190 url=self.rpc._url,
191 json=mock.ANY, # not checking here as of undefined order in dict
192 auth=self.rpc._credentials,
193 timeout=60)
195 def test_checked_for_application_error(self):
196 resultdict = {"result": "Sweet gorilla of Manila"}
197 self.assertEqual("Sweet gorilla of Manila",
198 (self.rpc._checked_for_application_error(
199 result=resultdict)))
201 def test_checked_for_application_error_enf(self):
202 resultdict = {"result": "Sweet gorilla of Manila",
203 "error": {"message": "No Gorilla",
204 "code": jsonrpc.ERROR_ENTITY_NOT_FOUND}}
205 self.assertIsNone(
206 self.rpc._checked_for_application_error(
207 result=resultdict,
208 expected_errors=[jsonrpc.ERROR_ENTITY_NOT_FOUND]))
210 def test_checked_for_application_error_no_entry(self):
211 resultdict = {"result": "Sweet gorilla of Manila",
212 "error": {"message": "No Gorilla",
213 "code": jsonrpc.ERROR_ENOENT}}
214 self.assertIsNone(
215 self.rpc._checked_for_application_error(
216 result=resultdict, expected_errors=[jsonrpc.ERROR_ENOENT]))
218 def test_checked_for_application_error_exception(self):
219 self.assertRaises(exception.QBRpcException,
220 self.rpc._checked_for_application_error,
221 {"result": "Sweet gorilla of Manila",
222 "error": {"message": "No Gorilla",
223 "code": 666
224 }
225 }
226 )