* Add 'isort' tox environment.
* Apply 'isort' fixes.
Signed-off-by: Volker Theile <vtheile@suse.com>
from __future__ import absolute_import
import os
+
import cherrypy
if 'COVERAGE_ENABLED' in os.environ:
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import hmac
-
+from email.utils import formatdate
from hashlib import sha1 as sha
+from requests.auth import AuthBase
+
py3k = False
try:
- from urlparse import urlparse, unquote
from base64 import encodestring
+
+ from urlparse import unquote, urlparse
except ImportError:
py3k = True
- from urllib.parse import urlparse, unquote
from base64 import encodebytes as encodestring
-
-from email.utils import formatdate
-
-from requests.auth import AuthBase
+ from urllib.parse import unquote, urlparse
class S3Auth(AuthBase):
# It was fixed in 3.7.0. Exact lower bound version is probably earlier,
# but 3.5.0 is what this monkey patch is tested on.
if StrictVersion("3.5.0") <= v < StrictVersion("3.7.0"):
- from cherrypy.wsgiserver.wsgiserver2 import \
- HTTPConnection, CP_fileobject
+ from cherrypy.wsgiserver.wsgiserver2 import CP_fileobject, HTTPConnection
def fixed_init(hc_self, server, sock, makefile=CP_fileobject):
hc_self.server = server
if v < StrictVersion('11.1.0'):
import uuid
from functools import update_wrapper
+
from cherrypy._cprequest import Request
class LazyUUID4(object):
import copy
import json
import os
-
from html.parser import HTMLParser
import re
import sys
import urllib
-
from functools import wraps
# pylint: disable=wrong-import-position
import cherrypy
-from ..security import Scope, Permission
-from ..tools import getargspec, TaskManager, get_request_body_params
-from ..exceptions import ScopeNotValid, PermissionNotValid
-from ..services.auth import AuthManager, JwtManager
+from ..exceptions import PermissionNotValid, ScopeNotValid
from ..plugins import PLUGIN_MANAGER
+from ..security import Permission, Scope
+from ..services.auth import AuthManager, JwtManager
+from ..tools import TaskManager, get_request_body_params, getargspec
try:
from typing import Any, List, Optional
from __future__ import absolute_import
import logging
+
import cherrypy
-from . import ApiController, RESTController, \
- allow_empty_body, ControllerDoc, EndpointDoc
from .. import mgr
from ..exceptions import DashboardException
from ..services.auth import AuthManager, JwtManager
-
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController, allow_empty_body
logger = logging.getLogger('controllers.auth')
"""
Provide authenticates and returns JWT token.
"""
+
def create(self, username, password):
user_data = AuthManager.authenticate(username, password)
user_perms, pwd_expiration_date, pwd_update_required = None, None, None
# -*- coding: utf-8 -*-
-from collections import defaultdict
-
import os
+from collections import defaultdict
-import cherrypy
import cephfs
+import cherrypy
-from . import ApiController, ControllerDoc, RESTController, UiApiController, \
- allow_empty_body, EndpointDoc
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
-from ..services.cephfs import CephFS as CephFS_
from ..services.ceph_service import CephService
+from ..services.cephfs import CephFS as CephFS_
from ..tools import ViewCache
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController, \
+ UiApiController, allow_empty_body
GET_QUOTAS_SCHEMA = {
'max_bytes': (int, ''),
import cherrypy
-from . import ApiController, RESTController, ControllerDoc, EndpointDoc
from .. import mgr
+from ..exceptions import DashboardException
from ..security import Scope
from ..services.ceph_service import CephService
-from ..exceptions import DashboardException
-
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController
FILTER_SCHEMA = [{
"name": (str, 'Name of the config option'),
from cherrypy import NotFound
-from . import ApiController, ControllerDoc, RESTController, Endpoint, ReadPermission, \
- UiApiController, EndpointDoc
+from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
-from .. import mgr
-
+from . import ApiController, ControllerDoc, Endpoint, EndpointDoc, \
+ ReadPermission, RESTController, UiApiController
LIST_SCHEMA = {
"rule_id": (int, 'Rule ID'),
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from typing import Any, Dict, Union
import logging
+from typing import Any, Dict, Union
+
import cherrypy
-from . import Controller, BaseController, Endpoint, ENDPOINT_MAP, \
- allow_empty_body
from .. import mgr
+from . import ENDPOINT_MAP, BaseController, Controller, Endpoint, allow_empty_body
NO_DESCRIPTION_AVAILABLE = "*No description available*"
if __name__ == "__main__":
import sys
+
import yaml
+
from . import generate_routes
def fix_null_descr(obj):
from cherrypy import NotFound
-from . import ApiController, ControllerDoc, RESTController, Endpoint, ReadPermission, \
- UiApiController, EndpointDoc
+from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
-from .. import mgr
+from . import ApiController, ControllerDoc, Endpoint, EndpointDoc, \
+ ReadPermission, RESTController, UiApiController
LIST_CODE__SCHEMA = {
"crush-failure-domain": (str, ''),
import logging
-from . import UiApiController, BaseController, Endpoint
-
+from . import BaseController, Endpoint, UiApiController
logger = logging.getLogger('frontend.error')
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import (ApiController, BaseController, Endpoint, ReadPermission,
- UpdatePermission)
-from . import ControllerDoc, EndpointDoc
from ..exceptions import DashboardException
from ..grafana import GrafanaRestClient, push_local_dashboards
from ..security import Scope
from ..settings import Settings
-
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, ReadPermission, UpdatePermission
URL_SCHEMA = {
"instance": (str, "grafana instance")
import json
-from . import ApiController, Endpoint, BaseController, ControllerDoc, EndpointDoc
-
from .. import mgr
from ..rest_client import RequestException
from ..security import Permission, Scope
from ..services.iscsi_cli import IscsiGatewaysConfig
from ..services.iscsi_client import IscsiClient
from ..tools import partial_dict
+from . import ApiController, BaseController, ControllerDoc, Endpoint, EndpointDoc
from .host import get_hosts
HEALTH_MINIMAL_SCHEMA = ({
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import os
-import re
import json
import logging
+import os
+import re
+
try:
from functools import lru_cache
except ImportError:
from ..plugins.lru_cache import lru_cache
+
import cherrypy
from cherrypy.lib.static import serve_file
-from . import Controller, UiApiController, BaseController, Proxy, Endpoint
from .. import mgr
-
+from . import BaseController, Controller, Endpoint, Proxy, UiApiController
logger = logging.getLogger("controllers.home")
from __future__ import absolute_import
import copy
-
-from typing import List, Dict
+from typing import Dict, List
import cherrypy
-
from mgr_util import merge_dicts
from orchestrator import HostSpec
-from . import ApiController, RESTController, Task, Endpoint, ReadPermission, \
- UiApiController, BaseController, allow_empty_body, ControllerDoc, EndpointDoc
-from .orchestrator import raise_if_no_orchestrator
+
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
-from ..services.orchestrator import OrchClient, OrchFeature
from ..services.ceph_service import CephService
from ..services.exception import handle_orchestrator_error
+from ..services.orchestrator import OrchClient, OrchFeature
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, ReadPermission, RESTController, Task, UiApiController, \
+ allow_empty_body
+from .orchestrator import raise_if_no_orchestrator
LIST_HOST_SCHEMA = {
"hostname": (str, "Hostname"),
# pylint: disable=too-many-lines
from __future__ import absolute_import
-from copy import deepcopy
-import re
import json
-import cherrypy
+import re
+from copy import deepcopy
+import cherrypy
import rados
import rbd
-from . import ApiController, UiApiController, RESTController, BaseController, Endpoint,\
- ReadPermission, UpdatePermission, Task, ControllerDoc, EndpointDoc
from .. import mgr
+from ..exceptions import DashboardException
from ..rest_client import RequestException
from ..security import Scope
-from ..services.iscsi_client import IscsiClient
from ..services.iscsi_cli import IscsiGatewaysConfig
+from ..services.iscsi_client import IscsiClient
from ..services.iscsi_config import IscsiGatewayDoesNotExist
from ..services.rbd import format_bitmask
from ..services.tcmu_service import TcmuService
-from ..exceptions import DashboardException
-from ..tools import str_to_bool, TaskManager
+from ..tools import TaskManager, str_to_bool
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, ReadPermission, RESTController, Task, UiApiController, \
+ UpdatePermission
try:
from typing import Any, Dict, List, no_type_check
import collections
-from . import ApiController, Endpoint, BaseController, ReadPermission, ControllerDoc, EndpointDoc
from ..security import Scope
from ..services.ceph_service import CephService
from ..tools import NotificationQueue
-
+from . import ApiController, BaseController, ControllerDoc, Endpoint, EndpointDoc, ReadPermission
LOG_BUFFER_SIZE = 30
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import ApiController, RESTController, \
- allow_empty_body, ControllerDoc, EndpointDoc
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
from ..services.exception import handle_send_command_error
from ..tools import find_object_in_list, str_to_bool
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController, allow_empty_body
MGR_MODULE_SCHEMA = ([{
"name": (str, "Module Name"),
import json
-from . import ApiController, Endpoint, BaseController, ReadPermission, ControllerDoc, EndpointDoc
from .. import mgr
from ..security import Scope
+from . import ApiController, BaseController, ControllerDoc, Endpoint, EndpointDoc, ReadPermission
MONITOR_SCHEMA = {
"mon_status": ({
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from functools import partial
import logging
+from functools import partial
-import cherrypy
import cephfs
+import cherrypy
-from . import ApiController, RESTController, UiApiController, BaseController, \
- Endpoint, Task, ReadPermission, ControllerDoc, EndpointDoc
from ..security import Scope
from ..services.cephfs import CephFS
from ..services.cephx import CephX
from ..services.exception import serialize_dashboard_exception
from ..services.ganesha import Ganesha, GaneshaConf, NFSException
from ..services.rgw_client import RgwClient
-
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, ReadPermission, RESTController, Task, UiApiController
logger = logging.getLogger('controllers.ganesha')
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import os.path
+import os.path
import time
from functools import wraps
-from . import ApiController, Endpoint, ReadPermission, UpdatePermission, ControllerDoc, EndpointDoc
-from . import RESTController, Task
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
from ..services.exception import handle_orchestrator_error
from ..services.orchestrator import OrchClient, OrchFeature
from ..tools import TaskManager, str_to_bool
+from . import ApiController, ControllerDoc, Endpoint, EndpointDoc, \
+ ReadPermission, RESTController, Task, UpdatePermission
STATUS_SCHEMA = {
"available": (bool, "Orchestrator status"),
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+
import json
import logging
import time
+from typing import Any, Dict, List, Union
from ceph.deployment.drive_group import DriveGroupSpec, DriveGroupValidationError # type: ignore
from mgr_util import get_most_recent_rate
-from . import ApiController, RESTController, Endpoint, Task
-from . import CreatePermission, ReadPermission, UpdatePermission, DeletePermission, \
- allow_empty_body, ControllerDoc, EndpointDoc
-from .orchestrator import raise_if_no_orchestrator
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
from ..services.ceph_service import CephService, SendCommandError
-from ..services.exception import handle_send_command_error, handle_orchestrator_error
+from ..services.exception import handle_orchestrator_error, handle_send_command_error
from ..services.orchestrator import OrchClient, OrchFeature
from ..tools import str_to_bool
-
-from typing import Any, Dict, List, Union # pylint: disable=C0411
-
+from . import ApiController, ControllerDoc, CreatePermission, \
+ DeletePermission, Endpoint, EndpointDoc, ReadPermission, RESTController, \
+ Task, UpdatePermission, allow_empty_body
+from .orchestrator import raise_if_no_orchestrator
logger = logging.getLogger('controllers.osd')
import cherrypy
-from . import ApiController, RESTController, EndpointDoc, ControllerDoc
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController
PERF_SCHEMA = {
"mon.a": ({
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from typing import Any, cast, Dict, Iterable, List, Optional, Union
import time
+from typing import Any, Dict, Iterable, List, Optional, Union, cast
+
import cherrypy
-from . import ApiController, ControllerDoc, RESTController, Endpoint, ReadPermission, Task, \
- UiApiController, EndpointDoc
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
-from ..services.rbd import RbdConfiguration
from ..services.exception import handle_send_command_error
-from ..tools import str_to_bool, TaskManager
+from ..services.rbd import RbdConfiguration
+from ..tools import TaskManager, str_to_bool
+from . import ApiController, ControllerDoc, Endpoint, EndpointDoc, \
+ ReadPermission, RESTController, Task, UiApiController
POOL_SCHEMA = ([{
"pool": (int, "pool id"),
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from datetime import datetime
import json
+from datetime import datetime
+
import requests
-from . import Controller, ApiController, BaseController, RESTController, Endpoint, ControllerDoc
+from ..exceptions import DashboardException
from ..security import Scope
from ..settings import Settings
-from ..exceptions import DashboardException
+from . import ApiController, BaseController, Controller, ControllerDoc, Endpoint, RESTController
@Controller('/api/prometheus_receiver', secure=False)
import logging
import math
-from functools import partial
from datetime import datetime
+from functools import partial
import rbd
-from . import ApiController, RESTController, Task, UpdatePermission, \
- DeletePermission, CreatePermission, allow_empty_body, ControllerDoc, EndpointDoc
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
from ..services.ceph_service import CephService
+from ..services.exception import handle_rados_error, handle_rbd_error, serialize_dashboard_exception
from ..services.rbd import RbdConfiguration, RbdService, RbdSnapshotService, \
- format_bitmask, format_features, parse_image_spec, rbd_call, rbd_image_call
+ format_bitmask, format_features, parse_image_spec, rbd_call, \
+ rbd_image_call
from ..tools import ViewCache, str_to_bool
-from ..services.exception import handle_rados_error, handle_rbd_error, \
- serialize_dashboard_exception
+from . import ApiController, ControllerDoc, CreatePermission, \
+ DeletePermission, EndpointDoc, RESTController, Task, UpdatePermission, \
+ allow_empty_body
logger = logging.getLogger(__name__)
from __future__ import absolute_import
import json
-import re
import logging
-
+import re
from functools import partial
import cherrypy
-
import rbd
-from . import ApiController, Endpoint, Task, BaseController, ReadPermission, \
- UpdatePermission, RESTController, allow_empty_body, ControllerDoc, EndpointDoc
-
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
+from ..services.exception import handle_rados_error, handle_rbd_error, serialize_dashboard_exception
from ..services.rbd import rbd_call
from ..tools import ViewCache
-from ..services.exception import handle_rados_error, handle_rbd_error, \
- serialize_dashboard_exception
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, ReadPermission, RESTController, Task, UpdatePermission, \
+ allow_empty_body
try:
from typing import no_type_check
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import logging
import json
+import logging
import cherrypy
-from . import ApiController, BaseController, RESTController, Endpoint, \
- ReadPermission, allow_empty_body, ControllerDoc, EndpointDoc
+
from ..exceptions import DashboardException
from ..rest_client import RequestException
-from ..security import Scope, Permission
+from ..security import Permission, Scope
from ..services.auth import AuthManager, JwtManager
from ..services.ceph_service import CephService
from ..services.rgw_client import RgwClient
from ..tools import json_str_to_object, str_to_bool
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, ReadPermission, RESTController, allow_empty_body
try:
from typing import Any, List
import cherrypy
-from . import ApiController, RESTController, UiApiController,\
- CreatePermission, ControllerDoc, EndpointDoc
from .. import mgr
-from ..exceptions import RoleDoesNotExist, DashboardException,\
- RoleIsAssociatedWithUser, RoleAlreadyExists
-from ..security import Scope as SecurityScope, Permission
+from ..exceptions import DashboardException, RoleAlreadyExists, \
+ RoleDoesNotExist, RoleIsAssociatedWithUser
+from ..security import Permission
+from ..security import Scope as SecurityScope
from ..services.access_control import SYSTEM_ROLES
+from . import ApiController, ControllerDoc, CreatePermission, EndpointDoc, \
+ RESTController, UiApiController
ROLE_SCHEMA = [{
"name": (str, "Role Name"),
from ..exceptions import UserDoesNotExist
from ..services.auth import JwtManager
from ..tools import prepare_url_prefix
-from . import Controller, Endpoint, BaseController
+from . import BaseController, Controller, Endpoint
@Controller('/auth/saml2', secure=False)
-from typing import List, Optional, Dict
-import cherrypy
+from typing import Dict, List, Optional
+import cherrypy
from ceph.deployment.service_spec import ServiceSpec
-from . import ApiController, ControllerDoc, RESTController, Task, Endpoint, ReadPermission, \
- CreatePermission, DeletePermission
-from .orchestrator import raise_if_no_orchestrator
+
from ..exceptions import DashboardException
from ..security import Scope
-from ..services.orchestrator import OrchClient, OrchFeature
from ..services.exception import handle_orchestrator_error
+from ..services.orchestrator import OrchClient, OrchFeature
+from . import ApiController, ControllerDoc, CreatePermission, \
+ DeletePermission, Endpoint, ReadPermission, RESTController, Task
+from .orchestrator import raise_if_no_orchestrator
def service_task(name, metadata, wait_for=2.0):
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+
from contextlib import contextmanager
import cherrypy
-from . import ApiController, RESTController, UiApiController, ControllerDoc, EndpointDoc
-from ..settings import Settings as SettingsModule, Options
from ..security import Scope
+from ..settings import Options
+from ..settings import Settings as SettingsModule
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController, UiApiController
SETTINGS_SCHEMA = [{
"name": (str, 'Settings Name'),
import json
-from . import ApiController, Endpoint, BaseController, ControllerDoc, EndpointDoc
from .. import mgr
-from ..security import Permission, Scope
from ..controllers.rbd_mirroring import get_daemons_and_pools
from ..exceptions import ViewCacheNoDataException
-from ..tools import TaskManager
+from ..security import Permission, Scope
from ..services import progress
+from ..tools import TaskManager
+from . import ApiController, BaseController, ControllerDoc, Endpoint, EndpointDoc
SUMMARY_SCHEMA = {
"health_status": (str, ""),
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import ApiController, RESTController, ControllerDoc, EndpointDoc
-from ..tools import TaskManager
from ..services import progress
+from ..tools import TaskManager
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController
TASK_SCHEMA = {
"executing_tasks": (str, "ongoing executing tasks"),
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import ApiController, RESTController, ControllerDoc, EndpointDoc
from .. import mgr
from ..exceptions import DashboardException
from ..security import Scope
+from . import ApiController, ControllerDoc, EndpointDoc, RESTController
REPORT_SCHEMA = {
"report": ({
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from datetime import datetime
-
import time
+from datetime import datetime
import cherrypy
-from . import BaseController, ApiController, RESTController, Endpoint, \
- allow_empty_body, ControllerDoc, EndpointDoc
from .. import mgr
-from ..exceptions import DashboardException, UserAlreadyExists, \
- UserDoesNotExist, PasswordPolicyException, PwdExpirationDateNotValid
+from ..exceptions import DashboardException, PasswordPolicyException, \
+ PwdExpirationDateNotValid, UserAlreadyExists, UserDoesNotExist
from ..security import Scope
from ..services.access_control import SYSTEM_ROLES, PasswordPolicy
from ..services.auth import JwtManager
+from . import ApiController, BaseController, ControllerDoc, Endpoint, \
+ EndpointDoc, RESTController, allow_empty_body
USER_SCHEMA = ([{
"username": (str, 'Username of the user'),
import logging
import os
import time
+
import requests
from .exceptions import GrafanaError
from .settings import Settings
-
logger = logging.getLogger('grafana')
import threading
import time
-from mgr_module import MgrModule, MgrStandbyModule, Option, CLIWriteCommand
-from mgr_util import get_default_addr, ServerConfigException, verify_tls_files, \
- create_self_signed_cert
+from mgr_module import CLIWriteCommand, MgrModule, MgrStandbyModule, Option
+from mgr_util import ServerConfigException, create_self_signed_cert, \
+ get_default_addr, verify_tls_files
+
+from . import mgr
+from .controllers import generate_routes, json_error_page
+from .grafana import push_local_dashboards
+from .services.auth import AuthManager, AuthManagerTool, JwtManager
+from .services.exception import dashboard_exception_handler
+from .services.sso import SSO_COMMANDS, handle_sso_command
+from .settings import handle_option_command, options_command_list, options_schema_list
+from .tools import NotificationQueue, RequestLoggingTool, TaskManager, \
+ prepare_url_prefix, str_to_bool
try:
import cherrypy
patch_cherrypy(cherrypy.__version__)
# pylint: disable=wrong-import-position
-from . import mgr
-from .controllers import generate_routes, json_error_page
-from .grafana import push_local_dashboards
-from .tools import NotificationQueue, RequestLoggingTool, TaskManager, \
- prepare_url_prefix, str_to_bool
-from .services.auth import AuthManager, AuthManagerTool, JwtManager
-from .services.sso import SSO_COMMANDS, \
- handle_sso_command
-from .services.exception import dashboard_exception_handler
-from .settings import options_command_list, options_schema_list, \
- handle_option_command
-
-from .plugins import PLUGIN_MANAGER
-from .plugins import feature_toggles, debug # noqa # pylint: disable=unused-import
-
+from .plugins import PLUGIN_MANAGER, debug, feature_toggles # noqa # pylint: disable=unused-import
PLUGIN_MANAGER.hook.init()
def set_ssl_certificate(self, mgr_id=None, inbuf=None):
if inbuf is None:
return -errno.EINVAL, '',\
- 'Please specify the certificate file with "-i" option'
+ 'Please specify the certificate file with "-i" option'
if mgr_id is not None:
self.set_store('{}/crt'.format(mgr_id), inbuf)
else:
def set_ssl_certificate_key(self, mgr_id=None, inbuf=None):
if inbuf is None:
return -errno.EINVAL, '',\
- 'Please specify the certificate key file with "-i" option'
+ 'Please specify the certificate key file with "-i" option'
if mgr_id is not None:
self.set_store('{}/key'.format(mgr_id), inbuf)
else:
import abc
-from .pluggy import HookspecMarker, HookimplMarker, PluginManager
+from .pluggy import HookimplMarker, HookspecMarker, PluginManager
class Interface(object, metaclass=abc.ABCMeta):
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from enum import Enum
import json
+from enum import Enum
from . import PLUGIN_MANAGER as PM
from . import interfaces as I # noqa: E741,N812
from __future__ import absolute_import
from enum import Enum
+
import cherrypy
from mgr_module import CLICommand, Option
+
+from ..controllers.cephfs import CephFS
+from ..controllers.iscsi import Iscsi, IscsiTarget
+from ..controllers.nfsganesha import NFSGanesha, NFSGaneshaExports, NFSGaneshaService
+from ..controllers.rbd import Rbd, RbdSnapshot, RbdTrash
+from ..controllers.rbd_mirroring import RbdMirroringPoolMode, \
+ RbdMirroringPoolPeer, RbdMirroringSummary
+from ..controllers.rgw import Rgw, RgwBucket, RgwDaemon, RgwUser
from . import PLUGIN_MANAGER as PM
from . import interfaces as I # noqa: E741,N812
from .ttl_cache import ttl_cache
-from ..controllers.rbd import Rbd, RbdSnapshot, RbdTrash
-from ..controllers.rbd_mirroring import (
- RbdMirroringSummary, RbdMirroringPoolMode, RbdMirroringPoolPeer)
-from ..controllers.iscsi import Iscsi, IscsiTarget
-from ..controllers.cephfs import CephFS
-from ..controllers.rgw import Rgw, RgwDaemon, RgwBucket, RgwUser
-from ..controllers.nfsganesha import NFSGanesha, NFSGaneshaService, NFSGaneshaExports
-
try:
- from typing import no_type_check, Set
+ from typing import Set, no_type_check
except ImportError:
no_type_check = object() # Just for type checking
404, "Feature='{}' disabled by option '{}'".format(
feature.value,
self.OPTION_FMT.format(feature.value),
- )
)
+ )
@PM.add_hook
def get_controllers(self):
- from ..controllers import ApiController, RESTController, ControllerDoc, EndpointDoc
+ from ..controllers import ApiController, ControllerDoc, EndpointDoc, RESTController
FEATURES_SCHEMA = {
"rbd": (bool, ''),
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import PLUGIN_MANAGER as PM, Interface, Mixin # pylint: disable=cyclic-import
+from . import PLUGIN_MANAGER as PM # pylint: disable=cyclic-import
+from . import Interface, Mixin
class CanMgr(Mixin):
"""
from __future__ import absolute_import
-from functools import wraps
from collections import OrderedDict
+from functools import wraps
from threading import RLock
class HookspecMarker(object):
""" Dummy implementation. No spec validation. """
+
def __init__(self, project_name):
self.project_name = project_name
Provides the PluginManager.hook.<method_name>() syntax and
functionality.
"""
+
def __init__(self):
from collections import defaultdict
self._registry = defaultdict(list) # type: DefaultDict[str, list]
-from mgr_module import Option, Command
+from mgr_module import Command, Option
from . import PLUGIN_MANAGER as PM
from . import interfaces as I # noqa: E741,N812
"""
from __future__ import absolute_import
-from functools import wraps
from collections import OrderedDict
+from functools import wraps
from threading import RLock
from time import time
flake8-colors==0.1.6; python_version >= '3'
#TODO: Fix docstring issues: https://tracker.ceph.com/issues/41224
#flake8-docstrings
-#flake8-import-order
-#flake8-typing-imports; python_version >= '3'
#pep8-naming
rstcheck==3.3.1; python_version >= '3'
autopep8; python_version >= '3'
pyfakefs; python_version >= '3'
+isort==5.5.3
import inspect
import logging
import re
+
import requests
from requests.exceptions import ConnectionError, InvalidURL, Timeout
+
from .settings import Settings
from .tools import build_url
"""
Set timeout argument for all requests if this is not already done.
"""
+
def request(self, *args, **kwargs):
if ((args[8] if len(args) > 8 else None) is None) \
and kwargs.get('timeout') is None:
# pylint: disable=too-many-branches, too-many-locals, too-many-statements
from __future__ import absolute_import
-from string import punctuation, ascii_lowercase, digits, ascii_uppercase
-
import errno
import json
import logging
+import re
import threading
import time
-import re
-
from datetime import datetime, timedelta
+from string import ascii_lowercase, ascii_uppercase, digits, punctuation
import bcrypt
-
from mgr_module import CLIReadCommand, CLIWriteCommand
from .. import mgr
-from ..security import Scope, Permission
+from ..exceptions import PasswordPolicyException, PermissionNotValid, \
+ PwdExpirationDateNotValid, RoleAlreadyExists, RoleDoesNotExist, \
+ RoleIsAssociatedWithUser, RoleNotInUser, ScopeNotInRole, ScopeNotValid, \
+ UserAlreadyExists, UserDoesNotExist
+from ..security import Permission, Scope
from ..settings import Settings
-from ..exceptions import RoleAlreadyExists, RoleDoesNotExist, ScopeNotValid, \
- PermissionNotValid, RoleIsAssociatedWithUser, \
- UserAlreadyExists, UserDoesNotExist, ScopeNotInRole, \
- RoleNotInUser, PasswordPolicyException, PwdExpirationDateNotValid
-
logger = logging.getLogger('access_control')
except RoleDoesNotExist as ex:
if rolename in SYSTEM_ROLES:
return -errno.EPERM, '', "Cannot delete system role '{}'" \
- .format(rolename)
+ .format(rolename)
return -errno.ENOENT, '', str(ex)
except RoleIsAssociatedWithUser as ex:
return -errno.EPERM, '', str(ex)
except RoleDoesNotExist as ex:
if rolename in SYSTEM_ROLES:
return -errno.EPERM, '', "Cannot update system role '{}'" \
- .format(rolename)
+ .format(rolename)
return -errno.ENOENT, '', str(ex)
except ScopeNotValid as ex:
return -errno.EINVAL, '', str(ex) + "\n Possible values: {}" \
.format(Scope.all_scopes())
except PermissionNotValid as ex:
return -errno.EINVAL, '', str(ex) + \
- "\n Possible values: {}" \
- .format(Permission.all_permissions())
+ "\n Possible values: {}" \
+ .format(Permission.all_permissions())
@CLIWriteCommand('dashboard ac-role-del-scope-perms',
except RoleDoesNotExist as ex:
if rolename in SYSTEM_ROLES:
return -errno.EPERM, '', "Cannot update system role '{}'" \
- .format(rolename)
+ .format(rolename)
return -errno.ENOENT, '', str(ex)
except ScopeNotInRole as ex:
return -errno.ENOENT, '', str(ex)
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from base64 import b64encode
import json
import logging
import os
import threading
import time
import uuid
+from base64 import b64encode
import cherrypy
import jwt
-from .access_control import LocalAuthenticator, UserDoesNotExist
from .. import mgr
+from .access_control import LocalAuthenticator, UserDoesNotExist
cherrypy.config.update({
'response.headers.server': 'Ceph-Dashboard'
- })
+})
class JwtManager(object):
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+
import json
import logging
import rados
-
from mgr_module import CommandResult
-from mgr_util import get_time_series_rates, get_most_recent_rate
+from mgr_util import get_most_recent_rate, get_time_series_rates
from .. import mgr
from ..exceptions import DashboardException
try:
- from typing import Dict, Any, Union
+ from typing import Any, Dict, Union
except ImportError:
pass # For typing only
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from contextlib import contextmanager
-import logging
-
import datetime
+import logging
import os
+from contextlib import contextmanager
+
import cephfs
from .. import mgr
-
logger = logging.getLogger('cephfs')
from __future__ import absolute_import
import json
-from contextlib import contextmanager
import logging
+from contextlib import contextmanager
import cherrypy
-
-from orchestrator import OrchestratorError
-import rbd
import rados
+import rbd
+from orchestrator import OrchestratorError
+from ..exceptions import DashboardException, ViewCacheNoDataException
from ..services.ceph_service import SendCommandError
-from ..exceptions import ViewCacheNoDataException, DashboardException
-
logger = logging.getLogger('exception')
import re
from orchestrator import OrchestratorError
+
+from .. import mgr
+from ..exceptions import DashboardException
+from ..settings import Settings
from .cephfs import CephFS
from .cephx import CephX
from .orchestrator import OrchClient
-from .rgw_client import RgwClient, RequestException, NoCredentialsException
-from .. import mgr
-from ..settings import Settings
-from ..exceptions import DashboardException
-
+from .rgw_client import NoCredentialsException, RequestException, RgwClient
logger = logging.getLogger('ganesha')
from mgr_module import CLIReadCommand, CLIWriteCommand
-from .iscsi_client import IscsiClient
-from .iscsi_config import IscsiGatewaysConfig, IscsiGatewayAlreadyExists, InvalidServiceUrl, \
- ManagedByOrchestratorException, IscsiGatewayDoesNotExist
from ..rest_client import RequestException
+from .iscsi_client import IscsiClient
+from .iscsi_config import InvalidServiceUrl, IscsiGatewayAlreadyExists, \
+ IscsiGatewayDoesNotExist, IscsiGatewaysConfig, \
+ ManagedByOrchestratorException
@CLIReadCommand('dashboard iscsi-gateway-list', desc='List iSCSI gateways')
except ImportError:
from urllib.parse import urlparse
-from .iscsi_config import IscsiGatewaysConfig
-from ..settings import Settings
from ..rest_client import RestClient
-
+from ..settings import Settings
+from .iscsi_config import IscsiGatewaysConfig
logger = logging.getLogger('iscsi_client')
"""
for gateway_name, gateway_config in config['gateways'].items():
if '.' not in gateway_name:
- from .iscsi_client import IscsiClient # pylint: disable=cyclic-import
from ..rest_client import RequestException
+ from .iscsi_client import IscsiClient # pylint: disable=cyclic-import
try:
service_url = gateway_config['service_url']
new_gateway_name = IscsiClient.instance(
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import logging
+import logging
from functools import wraps
-from typing import List, Optional, Dict, Any
+from typing import Any, Dict, List, Optional
from ceph.deployment.service_spec import ServiceSpec
-from orchestrator import InventoryFilter, DeviceLightLoc, Completion
-from orchestrator import ServiceDescription, DaemonDescription
-from orchestrator import OrchestratorClientMixin, raise_if_exception, OrchestratorError
-from orchestrator import HostSpec
+from orchestrator import Completion, DaemonDescription, DeviceLightLoc, \
+ HostSpec, InventoryFilter, OrchestratorClientMixin, OrchestratorError, \
+ ServiceDescription, raise_if_exception
+
from .. import mgr
logger = logging.getLogger('orchestrator')
from __future__ import absolute_import
-from datetime import datetime
import logging
+from datetime import datetime
-from . import rbd # pylint: disable=no-name-in-module
from .. import mgr
-
+from . import rbd # pylint: disable=no-name-in-module
logger = logging.getLogger('progress')
from __future__ import absolute_import
import cherrypy
-
import rbd
from .. import mgr
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import re
-import logging
import ipaddress
-from distutils.util import strtobool
+import logging
+import re
import xml.etree.ElementTree as ET # noqa: N814
+from distutils.util import strtobool
+
+from .. import mgr
from ..awsauth import S3Auth
from ..exceptions import DashboardException
-from ..settings import Settings, Options
-from ..rest_client import RestClient, RequestException
-from ..tools import build_url, dict_contains_path, json_str_to_object,\
- partial_dict, dict_get
-from .. import mgr
+from ..rest_client import RequestException, RestClient
+from ..settings import Options, Settings
+from ..tools import build_url, dict_contains_path, dict_get, json_str_to_object, partial_dict
try:
from typing import Dict, List, Optional
# pylint: disable=too-many-return-statements,too-many-branches
from __future__ import absolute_import
-import os
import errno
import json
import logging
+import os
import threading
import warnings
-
from urllib import parse
from .. import mgr
from ..tools import prepare_url_prefix
-
logger = logging.getLogger('sso')
try:
- from onelogin.saml2.settings import OneLogin_Saml2_Settings as Saml2Settings
from onelogin.saml2.errors import OneLogin_Saml2_Error as Saml2Error
from onelogin.saml2.idp_metadata_parser import OneLogin_Saml2_IdPMetadataParser as Saml2Parser
+ from onelogin.saml2.settings import OneLogin_Saml2_Settings as Saml2Settings
python_saml_imported = True
except ImportError:
Saml2Settings(mgr.SSO_DB.saml2.onelogin_settings)
except Saml2Error:
return -errno.EPERM, '', 'Single Sign-On is not configured: ' \
- 'use `ceph dashboard sso setup saml2`'
+ 'use `ceph dashboard sso setup saml2`'
mgr.SSO_DB.protocol = 'saml2'
mgr.SSO_DB.save()
return 0, 'SSO is "enabled" with "SAML2" protocol.', ''
from mgr_util import get_most_recent_rate
from dashboard.services.ceph_service import CephService
+
from .. import mgr
try:
perf_key = "{}lock_acquired_time".format(perf_key_prefix)
lock_acquired_time = (mgr.get_counter(
'tcmu-runner', service_id, perf_key)[perf_key]
- or [[0, 0]])[-1][1] / 1000000000
+ or [[0, 0]])[-1][1] / 1000000000
if lock_acquired_time > image.get('optimized_since', 0):
image['optimized_daemon'] = hostname
image['optimized_since'] = lock_acquired_time
@staticmethod
def has_default_value(name):
return getattr(Settings, name, None) is None or \
- getattr(Settings, name) == getattr(Options, name)[0]
+ getattr(Settings, name) == getattr(Options, name)[0]
class SettingsMeta(type):
import cherrypy
from cherrypy._cptools import HandlerWrapperTool
from cherrypy.test import helper
-from pyfakefs import fake_filesystem
-
from mgr_module import CLICommand
+from pyfakefs import fake_filesystem
from .. import mgr
-from ..controllers import json_error_page, generate_controller_routes
+from ..controllers import generate_controller_routes, json_error_page
+from ..plugins import PLUGIN_MANAGER, debug, feature_toggles # noqa
from ..services.auth import AuthManagerTool
from ..services.exception import dashboard_exception_handler
-from ..plugins import PLUGIN_MANAGER
-from ..plugins import feature_toggles, debug # noqa
-
-
PLUGIN_MANAGER.hook.init()
PLUGIN_MANAGER.hook.register_commands()
from __future__ import absolute_import
try:
- from typing import Dict, Any
+ from typing import Any, Dict
except ImportError:
pass
import json
import time
import unittest
-
from datetime import datetime, timedelta
-from . import CmdException, CLICommandTestMixin # pylint: disable=no-name-in-module
from .. import mgr
-from ..security import Scope, Permission
-from ..services.access_control import load_access_control_db, \
- password_hash, AccessControlDB, \
- SYSTEM_ROLES, PasswordPolicy
+from ..security import Permission, Scope
+from ..services.access_control import SYSTEM_ROLES, AccessControlDB, \
+ PasswordPolicy, load_access_control_db, password_hash
from ..settings import Settings
+from . import CLICommandTestMixin, CmdException # pylint: disable=no-name-in-module
class AccessControlTest(unittest.TestCase, CLICommandTestMixin):
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import re
import json
+import re
+
import cherrypy
+
try:
import mock
except ImportError:
import unittest.mock as mock
-from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
-from ..controllers import RESTController, Controller
-from ..tools import RequestLoggingTool
from .. import mgr
+from ..controllers import Controller, RESTController
+from ..tools import RequestLoggingTool
+from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
# pylint: disable=W0613
from __future__ import absolute_import
import unittest
+
try:
import mock
except ImportError:
# -*- coding: utf-8 -*-
from collections import defaultdict
+
try:
from mock import Mock
except ImportError:
from unittest.mock import Mock
from .. import mgr
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from ..controllers.cephfs import CephFS
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class MetaDataMock(object):
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+from ..controllers import ApiController, BaseController, Controller, Endpoint, RESTController
from . import ControllerTestCase # pylint: disable=no-name-in-module
-from ..controllers import BaseController, RESTController, Controller, \
- ApiController, Endpoint
@Controller("/btest/{key}", base_url="/ui", secure=False)
# # -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import ControllerTestCase # pylint: disable=no-name-in-module
-from ..controllers import RESTController, ApiController, Endpoint, EndpointDoc, ControllerDoc
+from ..controllers import ApiController, ControllerDoc, Endpoint, EndpointDoc, RESTController
from ..controllers.docs import Docs
+from . import ControllerTestCase # pylint: disable=no-name-in-module
# Dummy controller and endpoint that can be assigned with @EndpointDoc and @GroupDoc
# -*- coding: utf-8 -*-
from .. import mgr
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from ..controllers.erasure_code_profile import ErasureCodeProfile
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class ErasureCodeProfileTest(ControllerTestCase):
import rados
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..controllers import Controller, Endpoint, RESTController, Task
from ..services.ceph_service import SendCommandError
-from ..controllers import RESTController, Controller, Task, Endpoint
-from ..services.exception import handle_rados_error, handle_send_command_error, \
- serialize_dashboard_exception
-from ..tools import ViewCache, TaskManager, NotificationQueue
+from ..services.exception import handle_rados_error, \
+ handle_send_command_error, serialize_dashboard_exception
+from ..tools import NotificationQueue, TaskManager, ViewCache
+from . import ControllerTestCase # pylint: disable=no-name-in-module
# pylint: disable=W0613
from __future__ import absolute_import
import unittest
+
try:
from mock import Mock, patch
except ImportError:
from unittest.mock import Mock, patch
+from ..plugins.feature_toggles import Features, FeatureToggles
from . import KVStoreMockMixin # pylint: disable=no-name-in-module
-from ..plugins.feature_toggles import FeatureToggles, Features
class SettingsTest(unittest.TestCase, KVStoreMockMixin):
from unittest.mock import MagicMock, Mock
import orchestrator
-from . import KVStoreMockMixin # pylint: disable=no-name-in-module
+
from .. import mgr
-from ..settings import Settings
from ..services import ganesha
-from ..services.ganesha import GaneshaConf, Export, GaneshaConfParser
+from ..services.ganesha import Export, GaneshaConf, GaneshaConfParser
+from ..settings import Settings
+from . import KVStoreMockMixin # pylint: disable=no-name-in-module
class GaneshaConfTest(unittest.TestCase, KVStoreMockMixin):
from requests import RequestException
-from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
from ..controllers.grafana import Grafana
from ..grafana import GrafanaRestClient
from ..settings import Settings
+from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
class GrafanaTest(ControllerTestCase, KVStoreMockMixin):
except ImportError:
import unittest.mock as mock
-from . import ControllerTestCase, FakeFsMixin # pylint: disable=no-name-in-module
from .. import mgr
-
from ..controllers.home import HomeController, LanguageMixin
+from . import ControllerTestCase, FakeFsMixin # pylint: disable=no-name-in-module
logger = logging.getLogger()
from orchestrator import HostSpec
-from . import ControllerTestCase # pylint: disable=no-name-in-module
-from ..controllers.host import get_hosts, Host, HostUi
from .. import mgr
+from ..controllers.host import Host, HostUi, get_hosts
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class HostControllerTest(ControllerTestCase):
-# pylint: disable=too-many-public-methods
+# pylint: disable=too-many-public-methods,too-many-lines
import copy
import errno
except ImportError:
import unittest.mock as mock
-from . import CmdException, ControllerTestCase, CLICommandTestMixin, \
- KVStoreMockMixin # pylint: disable=no-name-in-module
from .. import mgr
from ..controllers.iscsi import Iscsi, IscsiTarget
+from ..rest_client import RequestException
from ..services.iscsi_client import IscsiClient
from ..services.orchestrator import OrchClient
-from ..rest_client import RequestException
from ..tools import NotificationQueue, TaskManager
+from . import CLICommandTestMixin # pylint: disable=no-name-in-module
+from . import CmdException # pylint: disable=no-name-in-module
+from . import ControllerTestCase # pylint: disable=no-name-in-module
+from . import KVStoreMockMixin # pylint: disable=no-name-in-module
class IscsiTestCli(unittest.TestCase, CLICommandTestMixin):
import time
import unittest
-
from ..tools import NotificationQueue
import inspect
import unittest
+
try:
import mock
except ImportError:
from orchestrator import InventoryHost
from orchestrator import Orchestrator as OrchestratorBase
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from .. import mgr
-from ..controllers.orchestrator import get_device_osd_map
-from ..controllers.orchestrator import Orchestrator
-from ..controllers.orchestrator import OrchestratorInventory
+from ..controllers.orchestrator import Orchestrator, OrchestratorInventory, get_device_osd_map
from ..services.orchestrator import OrchFeature
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class OrchestratorControllerTest(ControllerTestCase):
# -*- coding: utf-8 -*-
import uuid
from contextlib import contextmanager
-
+from typing import Any, Dict, List, Optional
from unittest import mock
+
from ceph.deployment.drive_group import DeviceSelection, DriveGroupSpec # type: ignore
from ceph.deployment.service_spec import PlacementSpec # type: ignore
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from .. import mgr
from ..controllers.osd import Osd
from ..tools import NotificationQueue, TaskManager
-from .. import mgr
+from . import ControllerTestCase # pylint: disable=no-name-in-module
from .helper import update_dict # pylint: disable=import-error
-from typing import Any, Dict, List, Optional # pylint: disable=C0411
-
class OsdHelper(object):
DEFAULT_OSD_IDS = [0, 1, 2]
# -*- coding: utf-8 -*-
# pylint: disable=protected-access
import time
+
try:
import mock
except ImportError:
import unittest.mock as mock
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from ..controllers.pool import Pool
from ..controllers.task import Task
from ..tools import NotificationQueue, TaskManager
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class MockTask(object):
except ImportError:
from unittest.mock import patch
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from .. import mgr
-from ..controllers.prometheus import Prometheus, PrometheusReceiver, PrometheusNotifications
+from ..controllers.prometheus import Prometheus, PrometheusNotifications, PrometheusReceiver
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class PrometheusControllerTest(ControllerTestCase):
from __future__ import absolute_import
import json
+
try:
import mock
except ImportError:
import unittest.mock as mock
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from .. import mgr
+from ..controllers.rbd_mirroring import RbdMirroring, RbdMirroringPoolBootstrap, RbdMirroringSummary
from ..controllers.summary import Summary
-from ..controllers.rbd_mirroring import RbdMirroring, RbdMirroringSummary, \
- RbdMirroringPoolBootstrap
from ..services import progress
-
+from . import ControllerTestCase # pylint: disable=no-name-in-module
mock_list_servers = [{
'hostname': 'ceph-host',
'mgr_map': {
'services': {
'dashboard': 'https://ceph.dev:11000/'
- },
+ },
}
}[key]
mgr.url_prefix = ''
from __future__ import absolute_import
import unittest
+
try:
import mock
except ImportError:
import unittest.mock as mock
-from ..services.rbd import get_image_spec, parse_image_spec, RbdConfiguration
+from ..services.rbd import RbdConfiguration, get_image_spec, parse_image_spec
class RbdServiceTest(unittest.TestCase):
# -*- coding: utf-8 -*-
import unittest
+
import requests.exceptions
try:
from unittest.mock import patch
from urllib3.exceptions import MaxRetryError, ProtocolError
+
from .. import mgr
from ..rest_client import RequestException, RestClient
except ImportError:
import unittest.mock as mock
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from ..controllers import Controller, RESTController, Task
from ..controllers.task import Task as TaskController
from ..services import progress
from ..tools import NotificationQueue, TaskManager
+from . import ControllerTestCase # pylint: disable=no-name-in-module
@Controller('/test/task', secure=False)
except ImportError:
import unittest.mock as mock
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from ..controllers.rgw import RgwUser
+from . import ControllerTestCase # pylint: disable=no-name-in-module
class RgwUserControllerTestCase(ControllerTestCase):
import errno
import unittest
-from . import KVStoreMockMixin, ControllerTestCase # pylint: disable=no-name-in-module
+
from .. import settings
from ..controllers.settings import Settings as SettingsController
from ..settings import Settings, handle_option_command
+from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
class SettingsTest(unittest.TestCase, KVStoreMockMixin):
import errno
import unittest
-from . import CmdException, exec_dashboard_cmd, \
- KVStoreMockMixin # pylint: disable=no-name-in-module
from ..services.sso import handle_sso_command, load_sso_db
+from . import CmdException # pylint: disable=no-name-in-module
+from . import KVStoreMockMixin # pylint: disable=no-name-in-module
+from . import exec_dashboard_cmd # pylint: disable=no-name-in-module
class AccessControlTest(unittest.TestCase, KVStoreMockMixin):
from __future__ import absolute_import
import json
-import unittest
import threading
import time
+import unittest
from collections import defaultdict
from functools import partial
from ..services.exception import serialize_dashboard_exception
-from ..tools import NotificationQueue, TaskManager, TaskExecutor
+from ..tools import NotificationQueue, TaskExecutor, TaskManager
class MyTask(object):
import cherrypy
from cherrypy.lib.sessions import RamSession
+
try:
from mock import patch
except ImportError:
from unittest.mock import patch
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..controllers import ApiController, BaseController, Controller, Proxy, RESTController
from ..services.exception import handle_rados_error
-from ..controllers import RESTController, ApiController, Controller, \
- BaseController, Proxy
-from ..tools import dict_contains_path, json_str_to_object, partial_dict,\
- dict_get, RequestLoggingTool
+from ..tools import RequestLoggingTool, dict_contains_path, dict_get, \
+ json_str_to_object, partial_dict
+from . import ControllerTestCase # pylint: disable=no-name-in-module
# pylint: disable=W0613
# -*- coding: utf-8 -*-
from __future__ import absolute_import
+import collections
+import fnmatch
import inspect
import json
import logging
-
-import collections
-from datetime import datetime, timedelta
-from distutils.util import strtobool
-import fnmatch
-import time
import threading
+import time
import urllib
+from datetime import datetime, timedelta
+from distutils.util import strtobool
import cherrypy
-
from ceph.deployment.utils import wrap_ipv6
from . import mgr
from .exceptions import ViewCacheNoDataException
-from .settings import Settings
from .services.auth import JwtManager
+from .settings import Settings
try:
- from typing import Any, AnyStr, Callable, DefaultDict, Deque,\
- Dict, List, Set, Tuple, Union # noqa
+ from typing import Any, AnyStr, Callable, DefaultDict, Deque, Dict, List, Set, Tuple, Union
except ImportError:
pass # For typing only
#max-complexity = 10
format = ${cyan}%(path)s${reset}:${yellow_bold}%(row)d${reset}:${green_bold}%(col)d${reset}: ${red_bold}%(code)s${reset} %(text)s
+[isort]
+atomic = true
+multi_line_output = 2
+line_length = 100
+wrap_length = 80
+
[pylint]
# Allow similarity/code duplication detection
jobs = 1
{[base-lint]deps}
commands =
flake8
+ isort . --check
{[base-pylint]commands}
{[base-rst]commands}
commands =
python --version
autopep8 {[autopep8]addopts} {posargs:.}
-
+ isort .
[testenv:check]
commands =
projects / ceph.git / commitdiff