[ceph.git] / ceph / qa / tasks / mgr / dashboard / test_mgr_module.py

# -*- coding: utf-8 -*-
from __future__ import absolute_import

import logging
import requests

from .helper import DashboardTestCase, JAny, JObj, JList, JLeaf

logger = logging.getLogger(__name__)


class MgrModuleTestCase(DashboardTestCase):
    MGRS_REQUIRED = 1

    @classmethod
    def tearDownClass(cls):
        cls._ceph_cmd(['mgr', 'module', 'disable', 'telemetry'])
        super(MgrModuleTestCase, cls).tearDownClass()

    def wait_until_rest_api_accessible(self):
        """
        Wait until the REST API is accessible.
        """

        def _check_connection():
            try:
                # Try reaching an API endpoint successfully.
                self._get('/api/mgr/module')
                if self._resp.status_code == 200:
                    return True
            except requests.ConnectionError:
                pass
            return False

        self.wait_until_true(_check_connection, timeout=30)


class MgrModuleTest(MgrModuleTestCase):
    def test_list_disabled_module(self):
        self._ceph_cmd(['mgr', 'module', 'disable', 'telemetry'])
        self.wait_until_rest_api_accessible()
        data = self._get('/api/mgr/module')
        self.assertStatus(200)
        self.assertSchema(
            data,
            JList(
                JObj(sub_elems={
                    'name': JLeaf(str),
                    'enabled': JLeaf(bool),
                    'options': JObj(
                        {},
                        allow_unknown=True,
                        unknown_schema=JObj({
                            'name': str,
                            'type': str,
                            'level': str,
                            'flags': int,
                            'default_value': JAny(none=False),
                            'min': JAny(none=False),
                            'max': JAny(none=False),
                            'enum_allowed': JList(str),
                            'see_also': JList(str),
                            'desc': str,
                            'long_desc': str,
                            'tags': JList(str)
                        }))
                })))
        module_info = self.find_object_in_list('name', 'telemetry', data)
        self.assertIsNotNone(module_info)
        self.assertFalse(module_info['enabled'])

    def test_list_enabled_module(self):
        self._ceph_cmd(['mgr', 'module', 'enable', 'telemetry'])
        self.wait_until_rest_api_accessible()
        data = self._get('/api/mgr/module')
        self.assertStatus(200)
        self.assertSchema(
            data,
            JList(
                JObj(sub_elems={
                    'name': JLeaf(str),
                    'enabled': JLeaf(bool),
                    'options': JObj(
                        {},
                        allow_unknown=True,
                        unknown_schema=JObj({
                            'name': str,
                            'type': str,
                            'level': str,
                            'flags': int,
                            'default_value': JAny(none=False),
                            'min': JAny(none=False),
                            'max': JAny(none=False),
                            'enum_allowed': JList(str),
                            'see_also': JList(str),
                            'desc': str,
                            'long_desc': str,
                            'tags': JList(str)
                        }))
                })))
        module_info = self.find_object_in_list('name', 'telemetry', data)
        self.assertIsNotNone(module_info)
        self.assertTrue(module_info['enabled'])


class MgrModuleTelemetryTest(MgrModuleTestCase):
    def test_get(self):
        data = self._get('/api/mgr/module/telemetry')
        self.assertStatus(200)
        self.assertSchema(
            data,
            JObj(
                sub_elems={
                    'contact': JLeaf(str),
                    'description': JLeaf(str),
                    'enabled': JLeaf(bool),
                    'interval': JLeaf(int),
                    'leaderboard': JLeaf(bool),
                    'organization': JLeaf(str),
                    'proxy': JLeaf(str),
                    'url': JLeaf(str)
                }))

    def test_put(self):
        self.set_config_key('config/mgr/mgr/telemetry/contact', '')
        self.set_config_key('config/mgr/mgr/telemetry/description', '')
        self.set_config_key('config/mgr/mgr/telemetry/enabled', 'True')
        self.set_config_key('config/mgr/mgr/telemetry/interval', '72')
        self.set_config_key('config/mgr/mgr/telemetry/leaderboard', 'False')
        self.set_config_key('config/mgr/mgr/telemetry/organization', '')
        self.set_config_key('config/mgr/mgr/telemetry/proxy', '')
        self.set_config_key('config/mgr/mgr/telemetry/url', '')
        self._put(
            '/api/mgr/module/telemetry',
            data={
                'config': {
                    'contact': 'tux@suse.com',
                    'description': 'test',
                    'enabled': False,
                    'interval': 4711,
                    'leaderboard': True,
                    'organization': 'SUSE Linux',
                    'proxy': 'foo',
                    'url': 'https://foo.bar/report'
                }
            })
        self.assertStatus(200)
        data = self._get('/api/mgr/module/telemetry')
        self.assertStatus(200)
        self.assertEqual(data['contact'], 'tux@suse.com')
        self.assertEqual(data['description'], 'test')
        self.assertFalse(data['enabled'])
        self.assertEqual(data['interval'], 4711)
        self.assertTrue(data['leaderboard'])
        self.assertEqual(data['organization'], 'SUSE Linux')
        self.assertEqual(data['proxy'], 'foo')
        self.assertEqual(data['url'], 'https://foo.bar/report')

    def test_enable(self):
        self._ceph_cmd(['mgr', 'module', 'disable', 'telemetry'])
        self.wait_until_rest_api_accessible()
        try:
            # Note, an exception is thrown because the Ceph Mgr
            # modules are reloaded.
            self._post('/api/mgr/module/telemetry/enable')
        except requests.ConnectionError:
            pass
        self.wait_until_rest_api_accessible()
        data = self._get('/api/mgr/module')
        self.assertStatus(200)
        module_info = self.find_object_in_list('name', 'telemetry', data)
        self.assertIsNotNone(module_info)
        self.assertTrue(module_info['enabled'])

    def test_disable(self):
        # Enable the 'telemetry' module (all CephMgr modules are restarted)
        # and wait until the Dashboard REST API is accessible.
        self._ceph_cmd(['mgr', 'module', 'enable', 'telemetry'])
        self.wait_until_rest_api_accessible()
        try:
            # Note, an exception is thrown because the Ceph Mgr
            # modules are reloaded.
            self._post('/api/mgr/module/telemetry/disable')
        except requests.ConnectionError:
            pass
        self.wait_until_rest_api_accessible()
        data = self._get('/api/mgr/module')
        self.assertStatus(200)
        module_info = self.find_object_in_list('name', 'telemetry', data)
        self.assertIsNotNone(module_info)
        self.assertFalse(module_info['enabled'])
Commit	Line	Data
11fdf7f2 TL	1	# -- coding: utf-8 --
	2	from __future__ import absolute_import
	3
	4	import logging
	5	import requests
	6
	7	from .helper import DashboardTestCase, JAny, JObj, JList, JLeaf
	8
	9	logger = logging.getLogger(__name__)
	10
	11
	12	class MgrModuleTestCase(DashboardTestCase):
	13	MGRS_REQUIRED = 1
	14
	15	@classmethod
	16	def tearDownClass(cls):
	17	cls._ceph_cmd(['mgr', 'module', 'disable', 'telemetry'])
	18	super(MgrModuleTestCase, cls).tearDownClass()
	19
	20	def wait_until_rest_api_accessible(self):
	21	"""
	22	Wait until the REST API is accessible.
	23	"""
	24
	25	def _check_connection():
	26	try:
	27	# Try reaching an API endpoint successfully.
	28	self._get('/api/mgr/module')
	29	if self._resp.status_code == 200:
	30	return True
	31	except requests.ConnectionError:
	32	pass
	33	return False
	34
	35	self.wait_until_true(_check_connection, timeout=30)
	36
	37
	38	class MgrModuleTest(MgrModuleTestCase):
	39	def test_list_disabled_module(self):
	40	self._ceph_cmd(['mgr', 'module', 'disable', 'telemetry'])
	41	self.wait_until_rest_api_accessible()
	42	data = self._get('/api/mgr/module')
	43	self.assertStatus(200)
	44	self.assertSchema(
	45	data,
	46	JList(
	47	JObj(sub_elems={
	48	'name': JLeaf(str),
	49	'enabled': JLeaf(bool),
	50	'options': JObj(
	51	{},
	52	allow_unknown=True,
	53	unknown_schema=JObj({
	54	'name': str,
	55	'type': str,
	56	'level': str,
	57	'flags': int,
	58	'default_value': JAny(none=False),
	59	'min': JAny(none=False),
	60	'max': JAny(none=False),
	61	'enum_allowed': JList(str),
	62	'see_also': JList(str),
	63	'desc': str,
	64	'long_desc': str,
65	'tags': JList(str)
66	}))
67	})))
68	module_info = self.find_object_in_list('name', 'telemetry', data)
69	self.assertIsNotNone(module_info)
70	self.assertFalse(module_info['enabled'])
71
72	def test_list_enabled_module(self):
73	self._ceph_cmd(['mgr', 'module', 'enable', 'telemetry'])
74	self.wait_until_rest_api_accessible()
75	data = self._get('/api/mgr/module')
76	self.assertStatus(200)
77	self.assertSchema(
78	data,
79	JList(
80	JObj(sub_elems={
81	'name': JLeaf(str),
82	'enabled': JLeaf(bool),
83	'options': JObj(
84	{},
85	allow_unknown=True,
86	unknown_schema=JObj({
87	'name': str,
88	'type': str,
89	'level': str,
90	'flags': int,
91	'default_value': JAny(none=False),
92	'min': JAny(none=False),
93	'max': JAny(none=False),
94	'enum_allowed': JList(str),
95	'see_also': JList(str),
96	'desc': str,
97	'long_desc': str,
98	'tags': JList(str)
99	}))
100	})))
101	module_info = self.find_object_in_list('name', 'telemetry', data)
102	self.assertIsNotNone(module_info)
103	self.assertTrue(module_info['enabled'])
104
105
106	class MgrModuleTelemetryTest(MgrModuleTestCase):
107	def test_get(self):
108	data = self._get('/api/mgr/module/telemetry')
109	self.assertStatus(200)
110	self.assertSchema(
111	data,
112	JObj(
113	sub_elems={
114	'contact': JLeaf(str),
115	'description': JLeaf(str),
116	'enabled': JLeaf(bool),
117	'interval': JLeaf(int),
118	'leaderboard': JLeaf(bool),
119	'organization': JLeaf(str),
120	'proxy': JLeaf(str),
121	'url': JLeaf(str)
122	}))
123
124	def test_put(self):
125	self.set_config_key('config/mgr/mgr/telemetry/contact', '')
126	self.set_config_key('config/mgr/mgr/telemetry/description', '')
127	self.set_config_key('config/mgr/mgr/telemetry/enabled', 'True')
128	self.set_config_key('config/mgr/mgr/telemetry/interval', '72')
129	self.set_config_key('config/mgr/mgr/telemetry/leaderboard', 'False')
130	self.set_config_key('config/mgr/mgr/telemetry/organization', '')
131	self.set_config_key('config/mgr/mgr/telemetry/proxy', '')
132	self.set_config_key('config/mgr/mgr/telemetry/url', '')
133	self._put(
134	'/api/mgr/module/telemetry',
135	data={
136	'config': {
137	'contact': 'tux@suse.com',
138	'description': 'test',
139	'enabled': False,
140	'interval': 4711,
141	'leaderboard': True,
142	'organization': 'SUSE Linux',
143	'proxy': 'foo',
144	'url': 'https://foo.bar/report'
145	}
146	})
147	self.assertStatus(200)
148	data = self._get('/api/mgr/module/telemetry')
149	self.assertStatus(200)
150	self.assertEqual(data['contact'], 'tux@suse.com')
151	self.assertEqual(data['description'], 'test')
152	self.assertFalse(data['enabled'])
153	self.assertEqual(data['interval'], 4711)
154	self.assertTrue(data['leaderboard'])
155	self.assertEqual(data['organization'], 'SUSE Linux')
156	self.assertEqual(data['proxy'], 'foo')
157	self.assertEqual(data['url'], 'https://foo.bar/report')
158
159	def test_enable(self):
160	self._ceph_cmd(['mgr', 'module', 'disable', 'telemetry'])
161	self.wait_until_rest_api_accessible()
162	try:
163	# Note, an exception is thrown because the Ceph Mgr
164	# modules are reloaded.
165	self._post('/api/mgr/module/telemetry/enable')
166	except requests.ConnectionError:
167	pass
168	self.wait_until_rest_api_accessible()
169	data = self._get('/api/mgr/module')
170	self.assertStatus(200)
171	module_info = self.find_object_in_list('name', 'telemetry', data)
172	self.assertIsNotNone(module_info)
173	self.assertTrue(module_info['enabled'])
174
175	def test_disable(self):
176	# Enable the 'telemetry' module (all CephMgr modules are restarted)
177	# and wait until the Dashboard REST API is accessible.
178	self._ceph_cmd(['mgr', 'module', 'enable', 'telemetry'])
179	self.wait_until_rest_api_accessible()
180	try:
181	# Note, an exception is thrown because the Ceph Mgr
182	# modules are reloaded.
183	self._post('/api/mgr/module/telemetry/disable')
184	except requests.ConnectionError:
185	pass
186	self.wait_until_rest_api_accessible()
187	data = self._get('/api/mgr/module')
188	self.assertStatus(200)
189	module_info = self.find_object_in_list('name', 'telemetry', data)
190	self.assertIsNotNone(module_info)
191	self.assertFalse(module_info['enabled'])