[ceph.git] / ceph / src / pybind / mgr / dashboard / tests / test_rgw_client.py

# -*- coding: utf-8 -*-
# pylint: disable=too-many-public-methods
import errno
from unittest import TestCase
from unittest.mock import Mock, patch

from .. import mgr
from ..exceptions import DashboardException
from ..services.rgw_client import NoCredentialsException, \
    NoRgwDaemonsException, RgwClient, _parse_frontend_config
from ..settings import Settings
from ..tests import CLICommandTestMixin, RgwStub


@patch('dashboard.services.rgw_client.RgwClient._get_user_id', Mock(
    return_value='dummy_admin'))
@patch('dashboard.services.ceph_service.CephService.send_command', Mock(
    return_value=''))
class RgwClientTest(TestCase, CLICommandTestMixin):
    _dashboard_user_realm1_access_key = 'VUOFXZFK24H81ISTVBTR'
    _dashboard_user_realm1_secret_key = '0PGsCvXPGWS3AGgibUZEcd9efLrbbshlUkY3jruR'
    _dashboard_user_realm2_access_key = 'OMDR282VYLBC1ZYMYDL0'
    _dashboard_user_realm2_secret_key = 'N3thf7jAiwQ90PsPrhC2DIcvCFOsBXtBvPJJMdC3'
    _radosgw_admin_result_error = (-errno.EINVAL, '', 'fake error')
    _radosgw_admin_result_no_realms = (0, {}, '')
    _radosgw_admin_result_realms = (0, {"realms": ["realm1", "realm2"]}, '')
    _radosgw_admin_result_user_realm1 = (
        0,
        {
            "keys": [
                {
                    "user": "dashboard",
                    "access_key": _dashboard_user_realm1_access_key,
                    "secret_key": _dashboard_user_realm1_secret_key
                }
            ],
            "system": "true"
        },
        '')
    _radosgw_admin_result_user_realm2 = (
        0,
        {
            "keys": [
                {
                    "user": "dashboard",
                    "access_key": _dashboard_user_realm2_access_key,
                    "secret_key": _dashboard_user_realm2_secret_key
                }
            ],
            "system": "true"
        },
        '')

    def setUp(self):
        RgwStub.get_daemons()
        self.mock_kv_store()
        self.CONFIG_KEY_DICT.update({
            'RGW_API_ACCESS_KEY': 'klausmustermann',
            'RGW_API_SECRET_KEY': 'supergeheim',
        })

    def test_configure_credentials_error(self):
        self.CONFIG_KEY_DICT.update({
            'RGW_API_ACCESS_KEY': '',
            'RGW_API_SECRET_KEY': '',
        })
        # Get no realms, get no user, user creation fails.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
        ]
        with self.assertRaises(NoCredentialsException) as cm:
            RgwClient.admin_instance()
        self.assertIn('No RGW credentials found', str(cm.exception))

    def test_configure_credentials_error_with_realms(self):
        self.CONFIG_KEY_DICT.update({
            'RGW_API_ACCESS_KEY': '',
            'RGW_API_SECRET_KEY': '',
        })
        # Get realms, get no user, user creation fails.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_realms,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
        ]
        with self.assertRaises(NoCredentialsException) as cm:
            RgwClient.admin_instance()
        self.assertIn('No RGW credentials found', str(cm.exception))

    def test_set_rgw_credentials_command(self):
        # Get no realms, get user.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_user_realm1
        ]
        result = self.exec_cmd('set-rgw-credentials')
        self.assertEqual(result, 'RGW credentials configured')
        self.assertEqual(Settings.RGW_API_ACCESS_KEY, self._dashboard_user_realm1_access_key)
        self.assertEqual(Settings.RGW_API_SECRET_KEY, self._dashboard_user_realm1_secret_key)

        # Get no realms, get no user, user creation.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_user_realm1
        ]
        result = self.exec_cmd('set-rgw-credentials')
        self.assertEqual(result, 'RGW credentials configured')
        self.assertEqual(Settings.RGW_API_ACCESS_KEY, self._dashboard_user_realm1_access_key)
        self.assertEqual(Settings.RGW_API_SECRET_KEY, self._dashboard_user_realm1_secret_key)

        # Get realms, get users.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_realms,
            self._radosgw_admin_result_user_realm1,
            self._radosgw_admin_result_user_realm2
        ]
        result = self.exec_cmd('set-rgw-credentials')
        self.assertEqual(result, 'RGW credentials configured')
        self.assertEqual(Settings.RGW_API_ACCESS_KEY, {
            'realm1': self._dashboard_user_realm1_access_key,
            'realm2': self._dashboard_user_realm2_access_key
        })
        self.assertEqual(Settings.RGW_API_SECRET_KEY, {
            'realm1': self._dashboard_user_realm1_secret_key,
            'realm2': self._dashboard_user_realm2_secret_key
        })

        # Get realms, get no users, users' creation.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_realms,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_user_realm1,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_user_realm2
        ]
        result = self.exec_cmd('set-rgw-credentials')
        self.assertEqual(result, 'RGW credentials configured')
        self.assertEqual(Settings.RGW_API_ACCESS_KEY, {
            'realm1': self._dashboard_user_realm1_access_key,
            'realm2': self._dashboard_user_realm2_access_key
        })
        self.assertEqual(Settings.RGW_API_SECRET_KEY, {
            'realm1': self._dashboard_user_realm1_secret_key,
            'realm2': self._dashboard_user_realm2_secret_key
        })

        # Get realms, get no users, realm 2 user creation fails.
        mgr.send_rgwadmin_command.side_effect = [
            self._radosgw_admin_result_realms,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_user_realm1,
            self._radosgw_admin_result_error,
            self._radosgw_admin_result_error,
        ]
        result = self.exec_cmd('set-rgw-credentials')
        self.assertEqual(result, 'RGW credentials configured')
        self.assertEqual(Settings.RGW_API_ACCESS_KEY, {
            'realm1': self._dashboard_user_realm1_access_key,
        })
        self.assertEqual(Settings.RGW_API_SECRET_KEY, {
            'realm1': self._dashboard_user_realm1_secret_key,
        })

    def test_ssl_verify(self):
        Settings.RGW_API_SSL_VERIFY = True
        instance = RgwClient.admin_instance()
        self.assertTrue(instance.session.verify)

    def test_no_ssl_verify(self):
        Settings.RGW_API_SSL_VERIFY = False
        instance = RgwClient.admin_instance()
        self.assertFalse(instance.session.verify)

    def test_no_daemons(self):
        RgwStub.get_mgr_no_services()
        with self.assertRaises(NoRgwDaemonsException) as cm:
            RgwClient.admin_instance()
        self.assertIn('No RGW service is running.', str(cm.exception))

    @patch.object(RgwClient, '_get_daemon_zone_info')
    def test_get_placement_targets_from_zone(self, zone_info):
        zone_info.return_value = {
            'id': 'a0df30ea-4b5b-4830-b143-2bedf684663d',
            'placement_pools': [
                {
                    'key': 'default-placement',
                    'val': {
                        'index_pool': 'default.rgw.buckets.index',
                        'storage_classes': {
                            'STANDARD': {
                                'data_pool': 'default.rgw.buckets.data'
                            }
                        }
                    }
                }
            ]
        }

        instance = RgwClient.admin_instance()
        expected_result = {
            'zonegroup': 'zonegroup1',
            'placement_targets': [
                {
                    'name': 'default-placement',
                    'data_pool': 'default.rgw.buckets.data'
                }
            ]
        }
        self.assertEqual(expected_result, instance.get_placement_targets())

    @patch.object(RgwClient, '_get_realms_info')
    def test_get_realms(self, realms_info):
        realms_info.side_effect = [
            {
                'default_info': '51de8373-bc24-4f74-a9b7-8e9ef4cb71f7',
                'realms': [
                    'realm1',
                    'realm2'
                ]
            },
            {}
        ]
        instance = RgwClient.admin_instance()

        self.assertEqual(['realm1', 'realm2'], instance.get_realms())
        self.assertEqual([], instance.get_realms())

    def test_set_bucket_locking_error(self):
        instance = RgwClient.admin_instance()
        test_params = [
            ('COMPLIANCE', 'null', None, 'must be a positive integer'),
            ('COMPLIANCE', None, 'null', 'must be a positive integer'),
            ('COMPLIANCE', -1, None, 'must be a positive integer'),
            ('COMPLIANCE', None, -1, 'must be a positive integer'),
            ('COMPLIANCE', 1, 1, 'You can\'t specify both at the same time'),
            ('COMPLIANCE', None, None, 'You must specify at least one'),
            ('COMPLIANCE', 0, 0, 'You must specify at least one'),
            (None, 1, 0, 'must be either COMPLIANCE or GOVERNANCE'),
            ('', 1, 0, 'must be either COMPLIANCE or GOVERNANCE'),
            ('FAKE_MODE', 1, 0, 'must be either COMPLIANCE or GOVERNANCE')
        ]
        for params in test_params:
            mode, days, years, error_msg = params
            with self.assertRaises(DashboardException) as cm:
                instance.set_bucket_locking(
                    bucket_name='test',
                    mode=mode,
                    retention_period_days=days,
                    retention_period_years=years
                )
            self.assertIn(error_msg, str(cm.exception))

    @patch('dashboard.rest_client._Request', Mock())
    def test_set_bucket_locking_success(self):
        instance = RgwClient.admin_instance()
        test_params = [
            ('Compliance', '1', None),
            ('Governance', 1, None),
            ('COMPLIANCE', None, '1'),
            ('GOVERNANCE', None, 1),
        ]
        for params in test_params:
            mode, days, years = params
            self.assertIsNone(instance.set_bucket_locking(
                bucket_name='test',
                mode=mode,
                retention_period_days=days,
                retention_period_years=years
            ))


class RgwClientHelperTest(TestCase):
    def test_parse_frontend_config_1(self):
        self.assertEqual(_parse_frontend_config('beast port=8000'), (8000, False))

    def test_parse_frontend_config_2(self):
        self.assertEqual(_parse_frontend_config('beast port=80 port=8000'), (80, False))

    def test_parse_frontend_config_3(self):
        self.assertEqual(_parse_frontend_config('beast ssl_port=443 port=8000'), (443, True))

    def test_parse_frontend_config_4(self):
        self.assertEqual(_parse_frontend_config('beast endpoint=192.168.0.100:8000'), (8000, False))

    def test_parse_frontend_config_5(self):
        self.assertEqual(_parse_frontend_config('beast endpoint=[::1]'), (80, False))

    def test_parse_frontend_config_6(self):
        self.assertEqual(_parse_frontend_config(
            'beast ssl_endpoint=192.168.0.100:8443'), (8443, True))

    def test_parse_frontend_config_7(self):
        self.assertEqual(_parse_frontend_config('beast ssl_endpoint=192.168.0.100'), (443, True))

    def test_parse_frontend_config_8(self):
        self.assertEqual(_parse_frontend_config(
            'beast ssl_endpoint=[::1]:8443 endpoint=192.0.2.3:80'), (8443, True))

    def test_parse_frontend_config_9(self):
        self.assertEqual(_parse_frontend_config(
            'beast port=8080 endpoint=192.0.2.3:80'), (8080, False))

    def test_parse_frontend_config_10(self):
        self.assertEqual(_parse_frontend_config(
            'beast ssl_endpoint=192.0.2.3:8443 port=8080'), (8443, True))

    def test_parse_frontend_config_11(self):
        self.assertEqual(_parse_frontend_config('civetweb port=8000s'), (8000, True))

    def test_parse_frontend_config_12(self):
        self.assertEqual(_parse_frontend_config('civetweb port=443s port=8000'), (443, True))

    def test_parse_frontend_config_13(self):
        self.assertEqual(_parse_frontend_config('civetweb port=192.0.2.3:80'), (80, False))

    def test_parse_frontend_config_14(self):
        self.assertEqual(_parse_frontend_config('civetweb port=172.5.2.51:8080s'), (8080, True))

    def test_parse_frontend_config_15(self):
        self.assertEqual(_parse_frontend_config('civetweb port=[::]:8080'), (8080, False))

    def test_parse_frontend_config_16(self):
        self.assertEqual(_parse_frontend_config('civetweb port=ip6-localhost:80s'), (80, True))

    def test_parse_frontend_config_17(self):
        self.assertEqual(_parse_frontend_config('civetweb port=[2001:0db8::1234]:80'), (80, False))

    def test_parse_frontend_config_18(self):
        self.assertEqual(_parse_frontend_config('civetweb port=[::1]:8443s'), (8443, True))

    def test_parse_frontend_config_19(self):
        self.assertEqual(_parse_frontend_config('civetweb port=127.0.0.1:8443s+8000'), (8443, True))

    def test_parse_frontend_config_20(self):
        self.assertEqual(_parse_frontend_config('civetweb port=127.0.0.1:8080+443s'), (8080, False))

    def test_parse_frontend_config_21(self):
        with self.assertRaises(LookupError) as ctx:
            _parse_frontend_config('civetweb port=xyz')
        self.assertEqual(str(ctx.exception),
                         'Failed to determine RGW port from "civetweb port=xyz"')

    def test_parse_frontend_config_22(self):
        with self.assertRaises(LookupError) as ctx:
            _parse_frontend_config('civetweb')
        self.assertEqual(str(ctx.exception), 'Failed to determine RGW port from "civetweb"')

    def test_parse_frontend_config_23(self):
        with self.assertRaises(LookupError) as ctx:
            _parse_frontend_config('mongoose port=8080')
        self.assertEqual(str(ctx.exception),
                         'Failed to determine RGW port from "mongoose port=8080"')
Commit	Line	Data
11fdf7f2	1	# -- coding: utf-8 --
9f95a23c	2	# pylint: disable=too-many-public-methods
522d829b	3	import errno
f67539c2 TL	4	from unittest import TestCase
f67539c2 TL	5	from unittest.mock import Mock, patch
11fdf7f2	6
522d829b	7	from .. import mgr
f67539c2 TL	8	from ..exceptions import DashboardException
	9	from ..services.rgw_client import NoCredentialsException, \
	10	NoRgwDaemonsException, RgwClient, _parse_frontend_config
9f95a23c	11	from ..settings import Settings
a4b75251	12	from ..tests import CLICommandTestMixin, RgwStub
f91f0fd5 TL	13
f91f0fd5 TL	14
f67539c2 TL	15	@patch('dashboard.services.rgw_client.RgwClient._get_user_id', Mock(
f67539c2 TL	16	return_value='dummy_admin'))
aee94f69 TL	17	@patch('dashboard.services.ceph_service.CephService.send_command', Mock(
aee94f69 TL	18	return_value=''))
522d829b TL	19	class RgwClientTest(TestCase, CLICommandTestMixin):
	20	_dashboard_user_realm1_access_key = 'VUOFXZFK24H81ISTVBTR'
	21	_dashboard_user_realm1_secret_key = '0PGsCvXPGWS3AGgibUZEcd9efLrbbshlUkY3jruR'
	22	_dashboard_user_realm2_access_key = 'OMDR282VYLBC1ZYMYDL0'
	23	_dashboard_user_realm2_secret_key = 'N3thf7jAiwQ90PsPrhC2DIcvCFOsBXtBvPJJMdC3'
	24	_radosgw_admin_result_error = (-errno.EINVAL, '', 'fake error')
	25	_radosgw_admin_result_no_realms = (0, {}, '')
	26	_radosgw_admin_result_realms = (0, {"realms": ["realm1", "realm2"]}, '')
	27	_radosgw_admin_result_user_realm1 = (
	28	0,
	29	{
	30	"keys": [
	31	{
	32	"user": "dashboard",
	33	"access_key": _dashboard_user_realm1_access_key,
	34	"secret_key": _dashboard_user_realm1_secret_key
	35	}
	36	],
	37	"system": "true"
	38	},
	39	'')
	40	_radosgw_admin_result_user_realm2 = (
	41	0,
	42	{
	43	"keys": [
	44	{
	45	"user": "dashboard",
	46	"access_key": _dashboard_user_realm2_access_key,
	47	"secret_key": _dashboard_user_realm2_secret_key
	48	}
	49	],
	50	"system": "true"
	51	},
	52	'')
	53
11fdf7f2	54	def setUp(self):
f67539c2	55	RgwStub.get_daemons()
92f5a8d4 TL	56	self.mock_kv_store()
	57	self.CONFIG_KEY_DICT.update({
	58	'RGW_API_ACCESS_KEY': 'klausmustermann',
	59	'RGW_API_SECRET_KEY': 'supergeheim',
92f5a8d4	60	})
11fdf7f2	61
522d829b TL	62	def test_configure_credentials_error(self):
	63	self.CONFIG_KEY_DICT.update({
	64	'RGW_API_ACCESS_KEY': '',
	65	'RGW_API_SECRET_KEY': '',
	66	})
	67	# Get no realms, get no user, user creation fails.
	68	mgr.send_rgwadmin_command.side_effect = [
	69	self._radosgw_admin_result_error,
	70	self._radosgw_admin_result_error,
	71	self._radosgw_admin_result_error,
	72	]
	73	with self.assertRaises(NoCredentialsException) as cm:
	74	RgwClient.admin_instance()
	75	self.assertIn('No RGW credentials found', str(cm.exception))
	76
	77	def test_configure_credentials_error_with_realms(self):
	78	self.CONFIG_KEY_DICT.update({
	79	'RGW_API_ACCESS_KEY': '',
	80	'RGW_API_SECRET_KEY': '',
	81	})
	82	# Get realms, get no user, user creation fails.
	83	mgr.send_rgwadmin_command.side_effect = [
	84	self._radosgw_admin_result_realms,
	85	self._radosgw_admin_result_error,
	86	self._radosgw_admin_result_error,
	87	self._radosgw_admin_result_error,
	88	self._radosgw_admin_result_error,
	89	]
	90	with self.assertRaises(NoCredentialsException) as cm:
	91	RgwClient.admin_instance()
	92	self.assertIn('No RGW credentials found', str(cm.exception))
	93
	94	def test_set_rgw_credentials_command(self):
	95	# Get no realms, get user.
	96	mgr.send_rgwadmin_command.side_effect = [
	97	self._radosgw_admin_result_error,
	98	self._radosgw_admin_result_user_realm1
	99	]
	100	result = self.exec_cmd('set-rgw-credentials')
	101	self.assertEqual(result, 'RGW credentials configured')
	102	self.assertEqual(Settings.RGW_API_ACCESS_KEY, self._dashboard_user_realm1_access_key)
	103	self.assertEqual(Settings.RGW_API_SECRET_KEY, self._dashboard_user_realm1_secret_key)
	104
	105	# Get no realms, get no user, user creation.
	106	mgr.send_rgwadmin_command.side_effect = [
	107	self._radosgw_admin_result_error,
	108	self._radosgw_admin_result_error,
	109	self._radosgw_admin_result_user_realm1
	110	]
	111	result = self.exec_cmd('set-rgw-credentials')
	112	self.assertEqual(result, 'RGW credentials configured')
	113	self.assertEqual(Settings.RGW_API_ACCESS_KEY, self._dashboard_user_realm1_access_key)
	114	self.assertEqual(Settings.RGW_API_SECRET_KEY, self._dashboard_user_realm1_secret_key)
	115
	116	# Get realms, get users.
	117	mgr.send_rgwadmin_command.side_effect = [
	118	self._radosgw_admin_result_realms,
	119	self._radosgw_admin_result_user_realm1,
	120	self._radosgw_admin_result_user_realm2
	121	]
	122	result = self.exec_cmd('set-rgw-credentials')
	123	self.assertEqual(result, 'RGW credentials configured')
	124	self.assertEqual(Settings.RGW_API_ACCESS_KEY, {
	125	'realm1': self._dashboard_user_realm1_access_key,
126	'realm2': self._dashboard_user_realm2_access_key
127	})
128	self.assertEqual(Settings.RGW_API_SECRET_KEY, {
129	'realm1': self._dashboard_user_realm1_secret_key,
130	'realm2': self._dashboard_user_realm2_secret_key
131	})
132
133	# Get realms, get no users, users' creation.
134	mgr.send_rgwadmin_command.side_effect = [
135	self._radosgw_admin_result_realms,
136	self._radosgw_admin_result_error,
137	self._radosgw_admin_result_user_realm1,
138	self._radosgw_admin_result_error,
139	self._radosgw_admin_result_user_realm2
140	]
141	result = self.exec_cmd('set-rgw-credentials')
142	self.assertEqual(result, 'RGW credentials configured')
143	self.assertEqual(Settings.RGW_API_ACCESS_KEY, {
144	'realm1': self._dashboard_user_realm1_access_key,
145	'realm2': self._dashboard_user_realm2_access_key
146	})
147	self.assertEqual(Settings.RGW_API_SECRET_KEY, {
148	'realm1': self._dashboard_user_realm1_secret_key,
149	'realm2': self._dashboard_user_realm2_secret_key
150	})
151
152	# Get realms, get no users, realm 2 user creation fails.
153	mgr.send_rgwadmin_command.side_effect = [
154	self._radosgw_admin_result_realms,
155	self._radosgw_admin_result_error,
156	self._radosgw_admin_result_user_realm1,
157	self._radosgw_admin_result_error,
158	self._radosgw_admin_result_error,
159	]
160	result = self.exec_cmd('set-rgw-credentials')
161	self.assertEqual(result, 'RGW credentials configured')
162	self.assertEqual(Settings.RGW_API_ACCESS_KEY, {
163	'realm1': self._dashboard_user_realm1_access_key,
164	})
165	self.assertEqual(Settings.RGW_API_SECRET_KEY, {
166	'realm1': self._dashboard_user_realm1_secret_key,
167	})
168
11fdf7f2	169	def test_ssl_verify(self):
9f95a23c	170	Settings.RGW_API_SSL_VERIFY = True
11fdf7f2 TL	171	instance = RgwClient.admin_instance()
	172	self.assertTrue(instance.session.verify)
	173
	174	def test_no_ssl_verify(self):
9f95a23c	175	Settings.RGW_API_SSL_VERIFY = False
11fdf7f2 TL	176	instance = RgwClient.admin_instance()
11fdf7f2 TL	177	self.assertFalse(instance.session.verify)
9f95a23c	178
f67539c2 TL	179	def test_no_daemons(self):
	180	RgwStub.get_mgr_no_services()
	181	with self.assertRaises(NoRgwDaemonsException) as cm:
	182	RgwClient.admin_instance()
	183	self.assertIn('No RGW service is running.', str(cm.exception))
	184
9f95a23c	185	@patch.object(RgwClient, '_get_daemon_zone_info')
f91f0fd5	186	def test_get_placement_targets_from_zone(self, zone_info):
9f95a23c TL	187	zone_info.return_value = {
	188	'id': 'a0df30ea-4b5b-4830-b143-2bedf684663d',
	189	'placement_pools': [
	190	{
	191	'key': 'default-placement',
	192	'val': {
	193	'index_pool': 'default.rgw.buckets.index',
	194	'storage_classes': {
	195	'STANDARD': {
	196	'data_pool': 'default.rgw.buckets.data'
	197	}
	198	}
	199	}
	200	}
f91f0fd5	201	]
9f95a23c TL	202	}
9f95a23c TL	203
9f95a23c TL	204	instance = RgwClient.admin_instance()
9f95a23c TL	205	expected_result = {
f67539c2	206	'zonegroup': 'zonegroup1',
9f95a23c TL	207	'placement_targets': [
	208	{
	209	'name': 'default-placement',
	210	'data_pool': 'default.rgw.buckets.data'
	211	}
	212	]
	213	}
	214	self.assertEqual(expected_result, instance.get_placement_targets())
	215
e306af50 TL	216	@patch.object(RgwClient, '_get_realms_info')
	217	def test_get_realms(self, realms_info):
	218	realms_info.side_effect = [
	219	{
	220	'default_info': '51de8373-bc24-4f74-a9b7-8e9ef4cb71f7',
	221	'realms': [
	222	'realm1',
	223	'realm2'
	224	]
	225	},
	226	{}
	227	]
	228	instance = RgwClient.admin_instance()
	229
	230	self.assertEqual(['realm1', 'realm2'], instance.get_realms())
	231	self.assertEqual([], instance.get_realms())
	232
b3b6e05e TL	233	def test_set_bucket_locking_error(self):
	234	instance = RgwClient.admin_instance()
	235	test_params = [
	236	('COMPLIANCE', 'null', None, 'must be a positive integer'),
	237	('COMPLIANCE', None, 'null', 'must be a positive integer'),
	238	('COMPLIANCE', -1, None, 'must be a positive integer'),
	239	('COMPLIANCE', None, -1, 'must be a positive integer'),
	240	('COMPLIANCE', 1, 1, 'You can\'t specify both at the same time'),
	241	('COMPLIANCE', None, None, 'You must specify at least one'),
	242	('COMPLIANCE', 0, 0, 'You must specify at least one'),
	243	(None, 1, 0, 'must be either COMPLIANCE or GOVERNANCE'),
	244	('', 1, 0, 'must be either COMPLIANCE or GOVERNANCE'),
	245	('FAKE_MODE', 1, 0, 'must be either COMPLIANCE or GOVERNANCE')
	246	]
	247	for params in test_params:
	248	mode, days, years, error_msg = params
	249	with self.assertRaises(DashboardException) as cm:
	250	instance.set_bucket_locking(
	251	bucket_name='test',
	252	mode=mode,
	253	retention_period_days=days,
	254	retention_period_years=years
	255	)
	256	self.assertIn(error_msg, str(cm.exception))
	257
	258	@patch('dashboard.rest_client._Request', Mock())
	259	def test_set_bucket_locking_success(self):
	260	instance = RgwClient.admin_instance()
	261	test_params = [
	262	('Compliance', '1', None),
	263	('Governance', 1, None),
	264	('COMPLIANCE', None, '1'),
	265	('GOVERNANCE', None, 1),
	266	]
	267	for params in test_params:
	268	mode, days, years = params
	269	self.assertIsNone(instance.set_bucket_locking(
	270	bucket_name='test',
	271	mode=mode,
	272	retention_period_days=days,
	273	retention_period_years=years
	274	))
	275
9f95a23c	276
f67539c2	277	class RgwClientHelperTest(TestCase):
9f95a23c TL	278	def test_parse_frontend_config_1(self):
	279	self.assertEqual(_parse_frontend_config('beast port=8000'), (8000, False))
	280
	281	def test_parse_frontend_config_2(self):
	282	self.assertEqual(_parse_frontend_config('beast port=80 port=8000'), (80, False))
	283
	284	def test_parse_frontend_config_3(self):
	285	self.assertEqual(_parse_frontend_config('beast ssl_port=443 port=8000'), (443, True))
	286
	287	def test_parse_frontend_config_4(self):
	288	self.assertEqual(_parse_frontend_config('beast endpoint=192.168.0.100:8000'), (8000, False))
	289
	290	def test_parse_frontend_config_5(self):
	291	self.assertEqual(_parse_frontend_config('beast endpoint=[::1]'), (80, False))
	292
	293	def test_parse_frontend_config_6(self):
	294	self.assertEqual(_parse_frontend_config(
	295	'beast ssl_endpoint=192.168.0.100:8443'), (8443, True))
	296
	297	def test_parse_frontend_config_7(self):
	298	self.assertEqual(_parse_frontend_config('beast ssl_endpoint=192.168.0.100'), (443, True))
	299
	300	def test_parse_frontend_config_8(self):
	301	self.assertEqual(_parse_frontend_config(
	302	'beast ssl_endpoint=[::1]:8443 endpoint=192.0.2.3:80'), (8443, True))
	303
	304	def test_parse_frontend_config_9(self):
	305	self.assertEqual(_parse_frontend_config(
	306	'beast port=8080 endpoint=192.0.2.3:80'), (8080, False))
	307
	308	def test_parse_frontend_config_10(self):
	309	self.assertEqual(_parse_frontend_config(
	310	'beast ssl_endpoint=192.0.2.3:8443 port=8080'), (8443, True))
	311
	312	def test_parse_frontend_config_11(self):
	313	self.assertEqual(_parse_frontend_config('civetweb port=8000s'), (8000, True))
	314
	315	def test_parse_frontend_config_12(self):
	316	self.assertEqual(_parse_frontend_config('civetweb port=443s port=8000'), (443, True))
	317
	318	def test_parse_frontend_config_13(self):
	319	self.assertEqual(_parse_frontend_config('civetweb port=192.0.2.3:80'), (80, False))
	320
	321	def test_parse_frontend_config_14(self):
	322	self.assertEqual(_parse_frontend_config('civetweb port=172.5.2.51:8080s'), (8080, True))
	323
	324	def test_parse_frontend_config_15(self):
	325	self.assertEqual(_parse_frontend_config('civetweb port=[::]:8080'), (8080, False))
	326
	327	def test_parse_frontend_config_16(self):
	328	self.assertEqual(_parse_frontend_config('civetweb port=ip6-localhost:80s'), (80, True))
	329
	330	def test_parse_frontend_config_17(self):
	331	self.assertEqual(_parse_frontend_config('civetweb port=[2001:0db8::1234]:80'), (80, False))
	332
	333	def test_parse_frontend_config_18(self):
	334	self.assertEqual(_parse_frontend_config('civetweb port=[::1]:8443s'), (8443, True))
	335
	336	def test_parse_frontend_config_19(self):
	337	self.assertEqual(_parse_frontend_config('civetweb port=127.0.0.1:8443s+8000'), (8443, True))
	338
	339	def test_parse_frontend_config_20(self):
	340	self.assertEqual(_parse_frontend_config('civetweb port=127.0.0.1:8080+443s'), (8080, False))
	341
342	def test_parse_frontend_config_21(self):
343	with self.assertRaises(LookupError) as ctx:
344	_parse_frontend_config('civetweb port=xyz')
345	self.assertEqual(str(ctx.exception),
346	'Failed to determine RGW port from "civetweb port=xyz"')
347
348	def test_parse_frontend_config_22(self):
349	with self.assertRaises(LookupError) as ctx:
350	_parse_frontend_config('civetweb')
351	self.assertEqual(str(ctx.exception), 'Failed to determine RGW port from "civetweb"')
352
353	def test_parse_frontend_config_23(self):
354	with self.assertRaises(LookupError) as ctx:
355	_parse_frontend_config('mongoose port=8080')
356	self.assertEqual(str(ctx.exception),
357	'Failed to determine RGW port from "mongoose port=8080"')