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