summaryrefslogtreecommitdiffstats
path: root/src/pybind/mgr/dashboard/tests/test_rgw.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/pybind/mgr/dashboard/tests/test_rgw.py')
-rw-r--r--src/pybind/mgr/dashboard/tests/test_rgw.py241
1 files changed, 241 insertions, 0 deletions
diff --git a/src/pybind/mgr/dashboard/tests/test_rgw.py b/src/pybind/mgr/dashboard/tests/test_rgw.py
new file mode 100644
index 000000000..bfb1dbc70
--- /dev/null
+++ b/src/pybind/mgr/dashboard/tests/test_rgw.py
@@ -0,0 +1,241 @@
+from unittest.mock import Mock, call, patch
+
+from .. import mgr
+from ..controllers.rgw import Rgw, RgwDaemon, RgwUser
+from ..rest_client import RequestException
+from ..services.rgw_client import RgwClient
+from ..tests import ControllerTestCase, RgwStub
+
+
+class RgwControllerTestCase(ControllerTestCase):
+ @classmethod
+ def setup_server(cls):
+ cls.setup_controllers([Rgw], '/test')
+
+ def setUp(self) -> None:
+ RgwStub.get_daemons()
+ RgwStub.get_settings()
+
+ @patch.object(RgwClient, '_get_user_id', Mock(return_value='fake-user'))
+ @patch.object(RgwClient, 'is_service_online', Mock(return_value=True))
+ @patch.object(RgwClient, '_is_system_user', Mock(return_value=True))
+ @patch('dashboard.services.ceph_service.CephService.send_command')
+ def test_status_available(self, send_command):
+ send_command.return_value = ''
+ self._get('/test/ui-api/rgw/status')
+ self.assertStatus(200)
+ self.assertJsonBody({'available': True, 'message': None})
+
+ @patch.object(RgwClient, '_get_user_id', Mock(return_value='fake-user'))
+ @patch.object(RgwClient, 'is_service_online', Mock(
+ side_effect=RequestException('My test error')))
+ @patch('dashboard.services.ceph_service.CephService.send_command')
+ def test_status_online_check_error(self, send_command):
+ send_command.return_value = ''
+ self._get('/test/ui-api/rgw/status')
+ self.assertStatus(200)
+ self.assertJsonBody({'available': False,
+ 'message': 'My test error'})
+
+ @patch.object(RgwClient, '_get_user_id', Mock(return_value='fake-user'))
+ @patch.object(RgwClient, 'is_service_online', Mock(return_value=False))
+ @patch('dashboard.services.ceph_service.CephService.send_command')
+ def test_status_not_online(self, send_command):
+ send_command.return_value = ''
+ self._get('/test/ui-api/rgw/status')
+ self.assertStatus(200)
+ self.assertJsonBody({'available': False,
+ 'message': "Failed to connect to the Object Gateway's Admin Ops API."})
+
+ @patch.object(RgwClient, '_get_user_id', Mock(return_value='fake-user'))
+ @patch.object(RgwClient, 'is_service_online', Mock(return_value=True))
+ @patch.object(RgwClient, '_is_system_user', Mock(return_value=False))
+ @patch('dashboard.services.ceph_service.CephService.send_command')
+ def test_status_not_system_user(self, send_command):
+ send_command.return_value = ''
+ self._get('/test/ui-api/rgw/status')
+ self.assertStatus(200)
+ self.assertJsonBody({'available': False,
+ 'message': 'The system flag is not set for user "fake-user".'})
+
+ def test_status_no_service(self):
+ RgwStub.get_mgr_no_services()
+ self._get('/test/ui-api/rgw/status')
+ self.assertStatus(200)
+ self.assertJsonBody({'available': False, 'message': 'No RGW service is running.'})
+
+
+class RgwDaemonControllerTestCase(ControllerTestCase):
+ @classmethod
+ def setup_server(cls):
+ cls.setup_controllers([RgwDaemon], '/test')
+
+ @patch('dashboard.services.rgw_client.RgwClient._get_user_id', Mock(
+ return_value='dummy_admin'))
+ @patch('dashboard.services.ceph_service.CephService.send_command')
+ def test_list(self, send_command):
+ send_command.return_value = ''
+ RgwStub.get_daemons()
+ RgwStub.get_settings()
+ mgr.list_servers.return_value = [{
+ 'hostname': 'host1',
+ 'services': [{'id': '4832', 'type': 'rgw'}, {'id': '5356', 'type': 'rgw'}]
+ }]
+ mgr.get_metadata.side_effect = [
+ {
+ 'ceph_version': 'ceph version master (dev)',
+ 'id': 'daemon1',
+ 'realm_name': 'realm1',
+ 'zonegroup_name': 'zg1',
+ 'zone_name': 'zone1',
+ 'frontend_config#0': 'beast port=80'
+ },
+ {
+ 'ceph_version': 'ceph version master (dev)',
+ 'id': 'daemon2',
+ 'realm_name': 'realm2',
+ 'zonegroup_name': 'zg2',
+ 'zone_name': 'zone2',
+ 'frontend_config#0': 'beast port=80 ssl_port=443 ssl_certificate=config:/config'
+ }]
+ self._get('/test/api/rgw/daemon')
+ self.assertStatus(200)
+ self.assertJsonBody([{
+ 'id': 'daemon1',
+ 'service_map_id': '4832',
+ 'version': 'ceph version master (dev)',
+ 'server_hostname': 'host1',
+ 'realm_name': 'realm1',
+ 'zonegroup_name': 'zg1',
+ 'zone_name': 'zone1', 'default': True,
+ 'port': 80
+ },
+ {
+ 'id': 'daemon2',
+ 'service_map_id': '5356',
+ 'version': 'ceph version master (dev)',
+ 'server_hostname': 'host1',
+ 'realm_name': 'realm2',
+ 'zonegroup_name': 'zg2',
+ 'zone_name': 'zone2',
+ 'default': False,
+ 'port': 80
+ }])
+
+ def test_list_empty(self):
+ RgwStub.get_mgr_no_services()
+ self._get('/test/api/rgw/daemon')
+ self.assertStatus(200)
+ self.assertJsonBody([])
+
+
+class RgwUserControllerTestCase(ControllerTestCase):
+ @classmethod
+ def setup_server(cls):
+ cls.setup_controllers([RgwUser], '/test')
+
+ @patch('dashboard.controllers.rgw.RgwRESTController.proxy')
+ def test_user_list(self, mock_proxy):
+ mock_proxy.side_effect = [{
+ 'count': 3,
+ 'keys': ['test1', 'test2', 'test3'],
+ 'truncated': False
+ }]
+ self._get('/test/api/rgw/user?daemon_name=dummy-daemon')
+ self.assertStatus(200)
+ mock_proxy.assert_has_calls([
+ call('dummy-daemon', 'GET', 'user?list', {})
+ ])
+ self.assertJsonBody(['test1', 'test2', 'test3'])
+
+ @patch('dashboard.controllers.rgw.RgwRESTController.proxy')
+ def test_user_list_marker(self, mock_proxy):
+ mock_proxy.side_effect = [{
+ 'count': 3,
+ 'keys': ['test1', 'test2', 'test3'],
+ 'marker': 'foo:bar',
+ 'truncated': True
+ }, {
+ 'count': 1,
+ 'keys': ['admin'],
+ 'truncated': False
+ }]
+ self._get('/test/api/rgw/user')
+ self.assertStatus(200)
+ mock_proxy.assert_has_calls([
+ call(None, 'GET', 'user?list', {}),
+ call(None, 'GET', 'user?list', {'marker': 'foo:bar'})
+ ])
+ self.assertJsonBody(['test1', 'test2', 'test3', 'admin'])
+
+ @patch('dashboard.controllers.rgw.RgwRESTController.proxy')
+ @patch('dashboard.services.ceph_service.CephService.send_command')
+ def test_user_list_duplicate_marker(self, mock_proxy, send_command):
+ send_command.return_value = ''
+ mock_proxy.side_effect = [{
+ 'count': 3,
+ 'keys': ['test1', 'test2', 'test3'],
+ 'marker': 'foo:bar',
+ 'truncated': True
+ }, {
+ 'count': 3,
+ 'keys': ['test4', 'test5', 'test6'],
+ 'marker': 'foo:bar',
+ 'truncated': True
+ }, {
+ 'count': 1,
+ 'keys': ['admin'],
+ 'truncated': False
+ }]
+ self._get('/test/api/rgw/user')
+ self.assertStatus(500)
+
+ @patch('dashboard.controllers.rgw.RgwRESTController.proxy')
+ def test_user_list_invalid_marker(self, mock_proxy):
+ mock_proxy.side_effect = [{
+ 'count': 3,
+ 'keys': ['test1', 'test2', 'test3'],
+ 'marker': 'foo:bar',
+ 'truncated': True
+ }, {
+ 'count': 3,
+ 'keys': ['test4', 'test5', 'test6'],
+ 'marker': '',
+ 'truncated': True
+ }, {
+ 'count': 1,
+ 'keys': ['admin'],
+ 'truncated': False
+ }]
+ self._get('/test/api/rgw/user')
+ self.assertStatus(500)
+
+ @patch('dashboard.controllers.rgw.RgwRESTController.proxy')
+ @patch.object(RgwUser, '_keys_allowed')
+ def test_user_get_with_keys(self, keys_allowed, mock_proxy):
+ keys_allowed.return_value = True
+ mock_proxy.return_value = {
+ 'tenant': '',
+ 'user_id': 'my_user_id',
+ 'keys': [],
+ 'swift_keys': []
+ }
+ self._get('/test/api/rgw/user/testuser')
+ self.assertStatus(200)
+ self.assertInJsonBody('keys')
+ self.assertInJsonBody('swift_keys')
+
+ @patch('dashboard.controllers.rgw.RgwRESTController.proxy')
+ @patch.object(RgwUser, '_keys_allowed')
+ def test_user_get_without_keys(self, keys_allowed, mock_proxy):
+ keys_allowed.return_value = False
+ mock_proxy.return_value = {
+ 'tenant': '',
+ 'user_id': 'my_user_id',
+ 'keys': [],
+ 'swift_keys': []
+ }
+ self._get('/test/api/rgw/user/testuser')
+ self.assertStatus(200)
+ self.assertNotIn('keys', self.json_body())
+ self.assertNotIn('swift_keys', self.json_body())
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-11 19:24:22 +0000