import cherrypy
import cephfs
-from . import ApiController, RESTController, UiApiController
+from . import ApiController, ControllerDoc, RESTController, UiApiController
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
@UiApiController('/cephfs', Scope.CEPHFS)
+@ControllerDoc("Dashboard UI helper function; not part of the public API", "CephFSUi")
class CephFsUi(CephFS):
RESOURCE_ID = 'fs_id'
from cherrypy import NotFound
-from . import ApiController, RESTController, Endpoint, ReadPermission, UiApiController
+from . import ApiController, ControllerDoc, RESTController, Endpoint, ReadPermission, \
+ UiApiController
from ..security import Scope
from ..services.ceph_service import CephService
from .. import mgr
@UiApiController('/crush_rule', Scope.POOL)
+@ControllerDoc("Dashboard UI helper function; not part of the public API", "CrushRuleUi")
class CrushRuleUi(CrushRule):
@Endpoint()
@ReadPermission
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+from typing import Any, Dict, Union
import logging
import cherrypy
if endpoint.is_api or all_endpoints:
list_of_ctrl.add(endpoint.ctrl)
- tag_map = {}
+ tag_map: Dict[str, str] = {}
for ctrl in list_of_ctrl:
tag_name = ctrl.__name__
tag_descr = ""
@classmethod
def _gen_responses(cls, method, resp_object=None):
- resp = {
+ resp: Dict[str, Dict[str, Union[str, Any]]] = {
'400': {
"description": "Operation exception. Please check the "
"response body for details."
return parameters
@classmethod
- def _gen_paths(cls, all_endpoints, base_url):
+ def _gen_paths(cls, all_endpoints):
method_order = ['get', 'post', 'put', 'delete']
paths = {}
for path, endpoints in sorted(list(ENDPOINT_MAP.items()),
methods[method.lower()]['security'] = [{'jwt': []}]
if not skip:
- paths[path[len(base_url):]] = methods
+ paths[path] = methods
return paths
host = host[host.index(':')+3:]
logger.debug("Host: %s", host)
- paths = self._gen_paths(all_endpoints, base_url)
+ paths = self._gen_paths(all_endpoints)
if not base_url:
base_url = "/"
@Endpoint(path="api.json")
def api_json(self):
- return self._gen_spec(False, "/api")
+ return self._gen_spec(False, "/")
@Endpoint(path="api-all.json")
def api_all_json(self):
- return self._gen_spec(True, "/api")
+ return self._gen_spec(True, "/")
def _swagger_ui_page(self, all_endpoints=False, token=None):
base = cherrypy.request.base
from cherrypy import NotFound
-from . import ApiController, RESTController, Endpoint, ReadPermission, UiApiController
+from . import ApiController, ControllerDoc, RESTController, Endpoint, ReadPermission, \
+ UiApiController
from ..security import Scope
from ..services.ceph_service import CephService
from .. import mgr
@UiApiController('/erasure_code_profile', Scope.POOL)
+@ControllerDoc("Dashboard UI helper function; not part of the public API", "ErasureCodeProfileUi")
class ErasureCodeProfileUi(ErasureCodeProfile):
@Endpoint()
@ReadPermission
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+from typing import Any, cast, Dict, Iterable, List, Optional, Union
import time
import cherrypy
-from . import ApiController, RESTController, Endpoint, ReadPermission, Task, UiApiController
+from . import ApiController, ControllerDoc, RESTController, Endpoint, ReadPermission, Task, \
+ UiApiController
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
crush_rules = {r['rule_id']: r["rule_name"] for r in mgr.get('osd_map_crush')['rules']}
- res = {}
+ res: Dict[Union[int, str], Union[str, List[Any]]] = {}
for attr in attrs:
if attr not in pool:
continue
return self._pool_list(attrs, stats)
@classmethod
- def _get(cls, pool_name, attrs=None, stats=False):
- # type: (str, str, bool) -> dict
+ def _get(cls, pool_name: str, attrs: Optional[str] = None, stats: bool = False) -> dict:
pools = cls._pool_list(attrs, stats)
pool = [p for p in pools if p['pool_name'] == pool_name]
if not pool:
raise cherrypy.NotFound('No such pool')
return pool[0]
- def get(self, pool_name, attrs=None, stats=False):
- # type: (str, str, bool) -> dict
+ def get(self, pool_name: str, attrs: Optional[str] = None, stats: bool = False) -> dict:
pool = self._get(pool_name, attrs, stats)
pool['configuration'] = RbdConfiguration(pool_name).list()
return pool
yes_i_really_mean_it=True)
if update_existing:
original_app_metadata = set(
- current_pool.get('application_metadata'))
+ cast(Iterable[Any], current_pool.get('application_metadata')))
else:
original_app_metadata = set()
@UiApiController('/pool', Scope.POOL)
+@ControllerDoc("Dashboard UI helper function; not part of the public API", "PoolUi")
class PoolUi(Pool):
@Endpoint()
@ReadPermission
if o['name'] == conf_name][0]
profiles = CephService.get_erasure_code_profiles()
- used_rules = {}
- used_profiles = {}
+ used_rules: Dict[str, List[str]] = {}
+ used_profiles: Dict[str, List[str]] = {}
pool_names = []
for p in self._pool_list():
name = p['pool_name']
self.assertEqual(Docs()._type_to_str(str), "string")
def test_gen_paths(self):
- outcome = Docs()._gen_paths(False, "")['/api/doctest//decorated_func/{parameter}']['get']
+ outcome = Docs()._gen_paths(False)['/api/doctest//decorated_func/{parameter}']['get']
self.assertIn('tags', outcome)
self.assertIn('summary', outcome)
self.assertIn('parameters', outcome)
self.assertIn('responses', outcome)
+ def test_gen_paths_all(self):
+ paths = Docs()._gen_paths(False)
+ for key in paths:
+ self.assertTrue(any(base in key.split('/')[1] for base in ['api', 'ui-api']))
+
def test_gen_tags(self):
outcome = Docs()._gen_tags(False)[0]
self.assertEqual({'description': 'Group description', 'name': 'FooGroup'}, outcome)
projects / ceph.git / commitdiff