"""
import os
-import re
-from typing import NamedTuple
import cherrypy
-DEFAULT_API_VERSION = '1.0'
-
-
-class APIVersion(NamedTuple):
- major: int = 1
- minor: int = 0
-
- @classmethod
- def from_string(cls, version_string):
- result = re.match(r'application/vnd\.ceph\.api\.v(\d+)\.(\d+)\+json', version_string)
- if result:
- return cls(*(int(s) for s in result.groups()))
- return None
-
- def __str__(self):
- return f'{self.major}.{self.minor}'
-
- def supports(self, client_version):
- return self.major == client_version.major and client_version.minor <= self.minor
-
- @staticmethod
- def to_tuple(version: str):
- return APIVersion(int(version.split('.')[0]), int(version.split('.')[1]))
-
-
if 'COVERAGE_ENABLED' in os.environ:
import coverage # pylint: disable=import-error
__cov = coverage.Coverage(config_file="{}/.coveragerc".format(os.path.dirname(__file__)),
# pylint: disable=import-error
from ceph_argparse import ArgumentFormat
-from .. import DEFAULT_API_VERSION, APIVersion
from ..api.doc import SchemaInput, SchemaType
from ..exceptions import DashboardException, 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
+from ._version import APIVersion
try:
from typing import Any, Dict, List, Optional, Tuple, Union
class Endpoint:
def __init__(self, method=None, path=None, path_params=None, query_params=None, # noqa: N802
- json_response=True, proxy=False, xml=False, version=DEFAULT_API_VERSION):
+ json_response=True, proxy=False, xml=False,
+ version: Optional[APIVersion] = APIVersion.DEFAULT):
if method is None:
method = 'GET'
elif not isinstance(method, str) or \
@staticmethod
def _request_wrapper(func, method, json_response, xml, # pylint: disable=unused-argument
- version):
+ version: Optional[APIVersion]):
@wraps(func)
def inner(*args, **kwargs):
client_version = None
kwargs.update(params)
if version is not None:
- server_version = APIVersion.to_tuple(version)
- accept_header = cherrypy.request.headers.get('Accept')
- req_match = APIVersion.from_string(accept_header)
- if accept_header and req_match:
- client_version = req_match
- else:
- raise cherrypy.HTTPError(415, "Unable to find version in request header")
+ try:
+ client_version = APIVersion.from_mime_type(
+ cherrypy.request.headers['Accept'])
+ except Exception:
+ raise cherrypy.HTTPError(
+ 415, "Unable to find version in request header")
- if server_version.supports(client_version):
+ if version.supports(client_version):
ret = func(*args, **kwargs)
else:
- raise cherrypy.HTTPError(415,
- "Incorrect version: "
- "{} requested but {} is expected"
- "".format(
- f'{client_version.major}.{client_version.minor}',
- version))
+ raise cherrypy.HTTPError(
+ 415,
+ f"Incorrect version: endpoint is '{version!s}', "
+ f"client requested '{client_version!s}'"
+ )
else:
ret = func(*args, **kwargs)
if isinstance(ret, bytes):
if xml:
if version:
cherrypy.response.headers['Content-Type'] = \
- 'application/vnd.ceph.api.v{}+xml'.format(version)
+ version.to_mime_type(subtype='xml')
else:
cherrypy.response.headers['Content-Type'] = 'application/xml'
return ret.encode('utf8')
if json_response:
if version:
cherrypy.response.headers['Content-Type'] = \
- 'application/vnd.ceph.api.v{}+json'.format(version)
+ version.to_mime_type()
else:
cherrypy.response.headers['Content-Type'] = 'application/json'
ret = json.dumps(ret).encode('utf8')
}
_method_mapping = collections.OrderedDict([
- ('list', {'method': 'GET', 'resource': False, 'status': 200, 'version': DEFAULT_API_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('create', {'method': 'POST', 'resource': False, 'status': 201, 'version': DEFAULT_API_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('bulk_set', {'method': 'PUT', 'resource': False, 'status': 200, 'version': DEFAULT_API_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('bulk_delete', {'method': 'DELETE', 'resource': False, 'status': 204, 'version': DEFAULT_API_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('get', {'method': 'GET', 'resource': True, 'status': 200, 'version': DEFAULT_API_VERSION}),
- ('delete', {'method': 'DELETE', 'resource': True, 'status': 204, 'version': DEFAULT_API_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('set', {'method': 'PUT', 'resource': True, 'status': 200, 'version': DEFAULT_API_VERSION}),
- ('singleton_set', {'method': 'PUT', 'resource': False, 'status': 200, 'version': DEFAULT_API_VERSION}) # noqa E501 #pylint: disable=line-too-long
+ ('list', {'method': 'GET', 'resource': False, 'status': 200, 'version': APIVersion.DEFAULT}), # noqa E501 #pylint: disable=line-too-long
+ ('create', {'method': 'POST', 'resource': False, 'status': 201, 'version': APIVersion.DEFAULT}), # noqa E501 #pylint: disable=line-too-long
+ ('bulk_set', {'method': 'PUT', 'resource': False, 'status': 200, 'version': APIVersion.DEFAULT}), # noqa E501 #pylint: disable=line-too-long
+ ('bulk_delete', {'method': 'DELETE', 'resource': False, 'status': 204, 'version': APIVersion.DEFAULT}), # noqa E501 #pylint: disable=line-too-long
+ ('get', {'method': 'GET', 'resource': True, 'status': 200, 'version': APIVersion.DEFAULT}),
+ ('delete', {'method': 'DELETE', 'resource': True, 'status': 204, 'version': APIVersion.DEFAULT}), # noqa E501 #pylint: disable=line-too-long
+ ('set', {'method': 'PUT', 'resource': True, 'status': 200, 'version': APIVersion.DEFAULT}),
+ ('singleton_set', {'method': 'PUT', 'resource': False, 'status': 200, 'version': APIVersion.DEFAULT}) # noqa E501 #pylint: disable=line-too-long
])
@classmethod
'method': None,
'query_params': None,
'path': '',
- 'version': DEFAULT_API_VERSION,
+ 'version': APIVersion.DEFAULT,
'sec_permissions': hasattr(func, '_security_permissions'),
'permission': None,
}
func = cls._status_code_wrapper(func, endpoint_params['status'])
endp_func = Endpoint(endpoint_params['method'], path=endpoint_params['path'],
query_params=endpoint_params['query_params'],
- version=endpoint_params['version'])(func)
+ version=endpoint_params['version'])(func) # type: ignore
if endpoint_params['permission']:
_set_func_permissions(endp_func, [endpoint_params['permission']])
result.append(cls.Endpoint(cls, endp_func))
@staticmethod
def Resource(method=None, path=None, status=None, query_params=None, # noqa: N802
- version=DEFAULT_API_VERSION):
+ version: Optional[APIVersion] = APIVersion.DEFAULT):
if not method:
method = 'GET'
return _wrapper
@staticmethod
- def MethodMap(resource=False, status=None, version=DEFAULT_API_VERSION): # noqa: N802
+ def MethodMap(resource=False, status=None,
+ version: Optional[APIVersion] = APIVersion.DEFAULT): # noqa: N802
if status is None:
status = 200
@staticmethod
def Collection(method=None, path=None, status=None, query_params=None, # noqa: N802
- version=DEFAULT_API_VERSION):
+ version: Optional[APIVersion] = APIVersion.DEFAULT):
if not method:
method = 'GET'
--- /dev/null
+import re
+from typing import NamedTuple
+
+
+class APIVersion(NamedTuple):
+ """
+ >>> APIVersion(1,0)
+ APIVersion(major=1, minor=0)
+
+ >>> APIVersion._make([1,0])
+ APIVersion(major=1, minor=0)
+
+ >>> f'{APIVersion(1, 0)!r}'
+ 'APIVersion(major=1, minor=0)'
+ """
+ major: int
+ minor: int
+
+ DEFAULT = ... # type: ignore
+ EXPERIMENTAL = ... # type: ignore
+ NONE = ... # type: ignore
+
+ __MIME_TYPE_REGEX = re.compile( # type: ignore
+ r'^application/vnd\.ceph\.api\.v(\d+\.\d+)\+json$')
+
+ @classmethod
+ def from_string(cls, version_string: str) -> 'APIVersion':
+ """
+ >>> APIVersion.from_string("1.0")
+ APIVersion(major=1, minor=0)
+ """
+ return cls._make(int(s) for s in version_string.split('.'))
+
+ @classmethod
+ def from_mime_type(cls, mime_type: str) -> 'APIVersion':
+ """
+ >>> APIVersion.from_mime_type('application/vnd.ceph.api.v1.0+json')
+ APIVersion(major=1, minor=0)
+
+ """
+ return cls.from_string(cls.__MIME_TYPE_REGEX.match(mime_type).group(1))
+
+ def __str__(self):
+ """
+ >>> f'{APIVersion(1, 0)}'
+ '1.0'
+ """
+ return f'{self.major}.{self.minor}'
+
+ def to_mime_type(self, subtype='json'):
+ """
+ >>> APIVersion(1, 0).to_mime_type(subtype='xml')
+ 'application/vnd.ceph.api.v1.0+xml'
+ """
+ return f'application/vnd.ceph.api.v{self!s}+{subtype}'
+
+ def supports(self, client_version: "APIVersion") -> bool:
+ """
+ >>> APIVersion(1, 1).supports(APIVersion(1, 0))
+ True
+
+ >>> APIVersion(1, 0).supports(APIVersion(1, 1))
+ False
+
+ >>> APIVersion(2, 0).supports(APIVersion(1, 1))
+ False
+ """
+ return (self.major == client_version.major
+ and client_version.minor <= self.minor)
+
+
+# Sentinel Values
+APIVersion.DEFAULT = APIVersion(1, 0) # type: ignore
+APIVersion.EXPERIMENTAL = APIVersion(0, 1) # type: ignore
+APIVersion.NONE = APIVersion(0, 0) # type: ignore
from .. import mgr
from ..security import Scope
from ..services.ceph_service import CephService
-from . import ApiController, ControllerDoc, Endpoint, EndpointDoc, \
+from . import ApiController, APIVersion, ControllerDoc, Endpoint, EndpointDoc, \
ReadPermission, RESTController, UiApiController
LIST_SCHEMA = {
class CrushRule(RESTController):
@EndpointDoc("List Crush Rule Configuration",
responses={200: LIST_SCHEMA})
- @RESTController.MethodMap(version='2.0')
+ @RESTController.MethodMap(version=APIVersion(2, 0))
def list(self):
return mgr.get('osd_map_crush')['rules']
- @RESTController.MethodMap(version='2.0')
+ @RESTController.MethodMap(version=APIVersion(2, 0))
def get(self, name):
rules = mgr.get('osd_map_crush')['rules']
for r in rules:
# -*- coding: utf-8 -*-
import logging
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Optional, Union
import cherrypy
-from .. import DEFAULT_API_VERSION, mgr
+from .. import mgr
from ..api.doc import Schema, SchemaInput, SchemaType
-from . import ENDPOINT_MAP, BaseController, Controller, Endpoint
+from . import ENDPOINT_MAP, APIVersion, BaseController, Controller, Endpoint
NO_DESCRIPTION_AVAILABLE = "*No description available*"
return schema.as_dict()
@classmethod
- def _gen_responses(cls, method, resp_object=None, version=None):
+ def _gen_responses(cls, method, resp_object=None,
+ version: Optional[APIVersion] = None):
resp: Dict[str, Dict[str, Union[str, Any]]] = {
'400': {
"description": "Operation exception. Please check the "
}
if not version:
- version = DEFAULT_API_VERSION
+ version = APIVersion.DEFAULT
if method.lower() == 'get':
resp['200'] = {'description': "OK",
- 'content': {'application/vnd.ceph.api.v{}+json'.format(version):
+ 'content': {version.to_mime_type():
{'type': 'object'}}}
if method.lower() == 'post':
resp['201'] = {'description': "Resource created.",
- 'content': {'application/vnd.ceph.api.v{}+json'.format(version):
+ 'content': {version.to_mime_type():
{'type': 'object'}}}
if method.lower() == 'put':
resp['200'] = {'description': "Resource updated.",
- 'content': {'application/vnd.ceph.api.v{}+json'.format(version):
+ 'content': {version.to_mime_type():
{'type': 'object'}}}
if method.lower() == 'delete':
resp['204'] = {'description': "Resource deleted.",
- 'content': {'application/vnd.ceph.api.v{}+json'.format(version):
+ 'content': {version.to_mime_type():
{'type': 'object'}}}
if method.lower() in ['post', 'put', 'delete']:
resp['202'] = {'description': "Operation is still executing."
" Please check the task queue.",
- 'content': {'application/vnd.ceph.api.v{}+json'.format(version):
+ 'content': {version.to_mime_type():
{'type': 'object'}}}
if resp_object:
for status_code, response_body in resp_object.items():
if status_code in resp:
- resp[status_code].update({
- 'content': {
- 'application/vnd.ceph.api.v{}+json'.format(version): {
- 'schema': cls._gen_schema_for_content(response_body)}}})
+ resp[status_code].update(
+ {'content':
+ {version.to_mime_type():
+ {'schema': cls._gen_schema_for_content(response_body)}
+ }})
return resp
func = endpoint.func
summary = ''
- version = ''
+ version = None
resp = {}
p_info = []
from ..rest_client import RequestException
from ..security import Scope
from ..services import feedback
-from . import ApiController, ControllerDoc, RESTController
+from . import ControllerDoc, RESTController, UiApiController
-@ApiController('/feedback', Scope.CONFIG_OPT)
+@UiApiController('/feedback', Scope.CONFIG_OPT)
@ControllerDoc("Feedback API", "Report")
class FeedbackController(RESTController):
issueAPIkey = None
super(FeedbackController, self).__init__()
self.tracker_client = feedback.CephTrackerClient()
- @RESTController.MethodMap(version='0.1')
def create(self, project, tracker, subject, description):
"""
Create an issue.
http_status_code=401,
component='feedback')
- @RESTController.MethodMap(version='0.1')
def get(self, issue_number):
"""
Fetch issue details.
from ..services.exception import handle_orchestrator_error
from ..services.orchestrator import OrchClient, OrchFeature
from ..tools import TaskManager, str_to_bool
-from . import ApiController, BaseController, ControllerDoc, Endpoint, \
- EndpointDoc, ReadPermission, RESTController, Task, UiApiController, \
- UpdatePermission, allow_empty_body
+from . import ApiController, APIVersion, BaseController, ControllerDoc, \
+ Endpoint, EndpointDoc, ReadPermission, RESTController, Task, \
+ UiApiController, UpdatePermission, allow_empty_body
from .orchestrator import raise_if_no_orchestrator
LIST_HOST_SCHEMA = {
'status': (str, 'Host Status')
},
responses={200: None, 204: None})
- @RESTController.MethodMap(version='0.1')
+ @RESTController.MethodMap(version=APIVersion.EXPERIMENTAL)
def create(self, hostname: str,
addr: Optional[str] = None,
labels: Optional[List[str]] = None,
'force': (bool, 'Force Enter Maintenance')
},
responses={200: None, 204: None})
- @RESTController.MethodMap(version='0.1')
+ @RESTController.MethodMap(version=APIVersion.EXPERIMENTAL)
def set(self, hostname: str, update_labels: bool = False,
labels: List[str] = None, maintenance: bool = False,
force: bool = False):
summary: Get List Of Features
tags:
- FeatureTogglesEndpoint
- /api/feedback:
- post:
- description: "\n Create an issue.\n :param project: The affected\
- \ ceph component.\n :param tracker: The tracker type.\n :param\
- \ subject: The title of the issue.\n :param description: The description\
- \ of the issue.\n "
- parameters: []
- requestBody:
- content:
- application/json:
- schema:
- properties:
- description:
- type: string
- project:
- type: string
- subject:
- type: string
- tracker:
- type: string
- required:
- - project
- - tracker
- - subject
- - description
- type: object
- responses:
- '201':
- content:
- application/vnd.ceph.api.v0.1+json:
- type: object
- description: Resource created.
- '202':
- content:
- application/vnd.ceph.api.v0.1+json:
- type: object
- description: Operation is still executing. Please check the task queue.
- '400':
- description: Operation exception. Please check the response body for details.
- '401':
- description: Unauthenticated access. Please login first.
- '403':
- description: Unauthorized access. Please check your permissions.
- '500':
- description: Unexpected error. Please check the response body for the stack
- trace.
- security:
- - jwt: []
- tags:
- - Report
- /api/feedback/{issue_number}:
- get:
- description: "\n Fetch issue details.\n :param issueAPI: The issue\
- \ tracker API access key.\n "
- parameters:
- - in: path
- name: issue_number
- required: true
- schema:
- type: integer
- responses:
- '200':
- content:
- application/vnd.ceph.api.v0.1+json:
- type: object
- description: OK
- '400':
- description: Operation exception. Please check the response body for details.
- '401':
- description: Unauthenticated access. Please login first.
- '403':
- description: Unauthorized access. Please check your permissions.
- '500':
- description: Unexpected error. Please check the response body for the stack
- trace.
- security:
- - jwt: []
- tags:
- - Report
/api/grafana/dashboards:
post:
parameters: []
name: RbdSnapshot
- description: RBD Trash Management API
name: RbdTrash
-- description: Feedback API
- name: Report
- description: RGW Management API
name: Rgw
- description: RGW Bucket Management API
from mgr_module import HandleCommandResult
from pyfakefs import fake_filesystem
-from .. import DEFAULT_API_VERSION, mgr
-from ..controllers import generate_controller_routes, json_error_page
+from .. import mgr
+from ..controllers import APIVersion, generate_controller_routes, json_error_page
from ..module import Module
from ..plugins import PLUGIN_MANAGER, debug, feature_toggles # noqa
from ..services.auth import AuthManagerTool
if cls._request_logging:
cherrypy.config.update({'tools.request_logging.on': False})
- def _request(self, url, method, data=None, headers=None, version=DEFAULT_API_VERSION):
+ def _request(self, url, method, data=None, headers=None, version=APIVersion.DEFAULT):
if not data:
b = None
if version:
- h = [('Accept', 'application/vnd.ceph.api.v{}+json'.format(version)),
+ h = [('Accept', version.to_mime_type()),
('Content-Length', '0')]
else:
h = None
else:
b = json.dumps(data)
if version is not None:
- h = [('Accept', 'application/vnd.ceph.api.v{}+json'.format(version)),
+ h = [('Accept', version.to_mime_type()),
('Content-Type', 'application/json'),
('Content-Length', str(len(b)))]
h = headers
self.getPage(url, method=method, body=b, headers=h)
- def _get(self, url, headers=None, version=DEFAULT_API_VERSION):
+ def _get(self, url, headers=None, version=APIVersion.DEFAULT):
self._request(url, 'GET', headers=headers, version=version)
- def _post(self, url, data=None, version=DEFAULT_API_VERSION):
+ def _post(self, url, data=None, version=APIVersion.DEFAULT):
self._request(url, 'POST', data, version=version)
- def _delete(self, url, data=None, version=DEFAULT_API_VERSION):
+ def _delete(self, url, data=None, version=APIVersion.DEFAULT):
self._request(url, 'DELETE', data, version=version)
- def _put(self, url, data=None, version=DEFAULT_API_VERSION):
+ def _put(self, url, data=None, version=APIVersion.DEFAULT):
self._request(url, 'PUT', data, version=version)
- def _task_request(self, method, url, data, timeout, version=DEFAULT_API_VERSION):
+ def _task_request(self, method, url, data, timeout, version=APIVersion.DEFAULT):
self._request(url, method, data, version=version)
if self.status != '202 Accepted':
logger.info("task finished immediately")
elif method == 'DELETE':
self.status = '204 No Content'
- def _task_post(self, url, data=None, timeout=60, version=DEFAULT_API_VERSION):
+ def _task_post(self, url, data=None, timeout=60, version=APIVersion.DEFAULT):
self._task_request('POST', url, data, timeout, version=version)
- def _task_delete(self, url, timeout=60, version=DEFAULT_API_VERSION):
+ def _task_delete(self, url, timeout=60, version=APIVersion.DEFAULT):
self._task_request('DELETE', url, None, timeout, version=version)
- def _task_put(self, url, data=None, timeout=60, version=DEFAULT_API_VERSION):
+ def _task_put(self, url, data=None, timeout=60, version=APIVersion.DEFAULT):
self._task_request('PUT', url, data, timeout, version=version)
def json_body(self):
import unittest
from ..api.doc import SchemaType
-from ..controllers import ApiController, ControllerDoc, Endpoint, EndpointDoc, RESTController
+from ..controllers import ApiController, APIVersion, ControllerDoc, Endpoint, \
+ EndpointDoc, RESTController
from ..controllers.docs import Docs
from . import ControllerTestCase # pylint: disable=no-name-in-module
},
)
@Endpoint(json_response=False)
- @RESTController.Resource('PUT', version='0.1')
+ @RESTController.Resource('PUT', version=APIVersion(0, 1))
def decorated_func(self, parameter):
pass
- @RESTController.MethodMap(version='0.1')
+ @RESTController.MethodMap(version=APIVersion(0, 1))
def list(self):
pass
expected_response_content = {
'200': {
- 'application/vnd.ceph.api.v0.1+json': {
+ APIVersion(0, 1).to_mime_type(): {
'schema': {'type': 'array',
'items': {'type': 'object', 'properties': {
'my_prop': {
'description': '200 property desc.'}}},
'required': ['my_prop']}}},
'202': {
- 'application/vnd.ceph.api.v0.1+json': {
+ APIVersion(0, 1).to_mime_type(): {
'schema': {'type': 'object',
'properties': {'my_prop': {
'type': 'string',
def test_gen_method_paths(self):
outcome = Docs().gen_paths(False)['/api/doctest/']['get']
- self.assertEqual({'application/vnd.ceph.api.v0.1+json': {'type': 'object'}},
+ self.assertEqual({APIVersion(0, 1).to_mime_type(): {'type': 'object'}},
outcome['responses']['200']['content'])
def test_gen_paths_all(self):
from orchestrator import HostSpec, InventoryHost
from .. import mgr
+from ..controllers import APIVersion
from ..controllers.host import Host, HostUi, get_device_osd_map, get_hosts, get_inventories
from ..tools import NotificationQueue, TaskManager
from . import ControllerTestCase # pylint: disable=no-name-in-module
'labels': 'mon',
'status': 'maintenance'
}
- self._post(self.URL_HOST, payload, version='0.1')
+ self._post(self.URL_HOST, payload, version=APIVersion(0, 1))
self.assertStatus(201)
mock_add_host.assert_called()
fake_client.hosts.add_label = mock.Mock()
payload = {'update_labels': True, 'labels': ['bbb', 'ccc']}
- self._put('{}/node0'.format(self.URL_HOST), payload, version='0.1')
+ self._put('{}/node0'.format(self.URL_HOST), payload, version=APIVersion(0, 1))
self.assertStatus(200)
self.assertHeader('Content-Type',
'application/vnd.ceph.api.v0.1+json')
fake_client.hosts.add_label.assert_called_once_with('node0', 'ccc')
# return 400 if type other than List[str]
- self._put('{}/node0'.format(self.URL_HOST), {'update_labels': True,
- 'labels': 'ddd'}, version='0.1')
+ self._put('{}/node0'.format(self.URL_HOST),
+ {'update_labels': True, 'labels': 'ddd'},
+ version=APIVersion(0, 1))
self.assertStatus(400)
def test_host_maintenance(self):
]
with patch_orch(True, hosts=orch_hosts):
# enter maintenance mode
- self._put('{}/node0'.format(self.URL_HOST), {'maintenance': True}, version='0.1')
+ self._put('{}/node0'.format(self.URL_HOST), {'maintenance': True},
+ version=APIVersion(0, 1))
self.assertStatus(200)
self.assertHeader('Content-Type',
'application/vnd.ceph.api.v0.1+json')
# force enter maintenance mode
self._put('{}/node1'.format(self.URL_HOST), {'maintenance': True, 'force': True},
- version='0.1')
+ version=APIVersion(0, 1))
self.assertStatus(200)
# exit maintenance mode
- self._put('{}/node0'.format(self.URL_HOST), {'maintenance': True}, version='0.1')
+ self._put('{}/node0'.format(self.URL_HOST), {'maintenance': True},
+ version=APIVersion(0, 1))
self.assertStatus(200)
- self._put('{}/node1'.format(self.URL_HOST), {'maintenance': True}, version='0.1')
+ self._put('{}/node1'.format(self.URL_HOST), {'maintenance': True},
+ version=APIVersion(0, 1))
self.assertStatus(200)
# maintenance without orchestrator service
with patch_orch(False):
- self._put('{}/node0'.format(self.URL_HOST), {'maintenance': True}, version='0.1')
+ self._put('{}/node0'.format(self.URL_HOST), {'maintenance': True},
+ version=APIVersion(0, 1))
self.assertStatus(503)
@mock.patch('dashboard.controllers.host.time')
except ImportError:
from unittest.mock import patch
-from .. import DEFAULT_API_VERSION
-from ..controllers import ApiController, BaseController, Controller, Proxy, RESTController
+from ..controllers import ApiController, APIVersion, BaseController, \
+ Controller, Proxy, RESTController
from ..services.exception import handle_rados_error
from ..tools import dict_contains_path, dict_get, json_str_to_object, partial_dict
from . import ControllerTestCase # pylint: disable=no-name-in-module
self.assertStatus(204)
self._get("/foo")
self.assertStatus('200 OK')
- self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
+ self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type())
self.assertBody('[]')
def test_fill(self):
self._post("/foo", data)
self.assertJsonBody(data)
self.assertStatus(201)
- self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
+ self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type())
self._get("/foo")
self.assertStatus('200 OK')
- self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
+ self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type())
self.assertJsonBody([data] * 5)
self._put('/foo/0', {'newdata': 'newdata'})
self.assertStatus('200 OK')
- self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
+ self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type())
self.assertJsonBody({'newdata': 'newdata', 'key': '0'})
def test_not_implemented(self):
import unittest
-from ..controllers import ApiController, RESTController
+from ..controllers import ApiController, APIVersion, RESTController
from . import ControllerTestCase # pylint: disable=no-name-in-module
class VTest(RESTController):
RESOURCE_ID = "vid"
- @RESTController.MethodMap(version="0.1")
+ @RESTController.MethodMap(version=APIVersion(0, 1))
def list(self):
return {'version': ""}
def get(self):
return {'version': ""}
- @RESTController.Collection('GET', version="1.0")
+ @RESTController.Collection('GET', version=APIVersion(1, 0))
def vmethod(self):
return {'version': '1.0'}
- @RESTController.Collection('GET', version="1.1")
+ @RESTController.Collection('GET', version=APIVersion(1, 1))
def vmethodv1_1(self):
return {'version': '1.1'}
- @RESTController.Collection('GET', version="2.0")
+ @RESTController.Collection('GET', version=APIVersion(2, 0))
def vmethodv2(self):
return {'version': '2.0'}
def test_list(self):
for (version, expected_status) in [
- ("0.1", 200),
- ("2.0", 415)
+ ((0, 1), 200),
+ ((2, 0), 415)
]:
with self.subTest(version=version):
- self._get('/test/api/vtest', version=version)
+ self._get('/test/api/vtest', version=APIVersion._make(version))
self.assertStatus(expected_status)
def test_v1(self):
for (version, expected_status) in [
- ("1.0", 200),
- ("2.0", 415)
+ ((1, 0), 200),
+ ((2, 0), 415)
]:
with self.subTest(version=version):
- self._get('/test/api/vtest/vmethod', version=version)
+ self._get('/test/api/vtest/vmethod',
+ version=APIVersion._make(version))
self.assertStatus(expected_status)
def test_v2(self):
for (version, expected_status) in [
- ("2.0", 200),
- ("1.0", 415)
+ ((2, 0), 200),
+ ((1, 0), 415)
]:
with self.subTest(version=version):
- self._get('/test/api/vtest/vmethodv2', version=version)
+ self._get('/test/api/vtest/vmethodv2',
+ version=APIVersion._make(version))
self.assertStatus(expected_status)
def test_backward_compatibility(self):
for (version, expected_status) in [
- ("1.1", 200),
- ("1.0", 200),
- ("2.0", 415)
+ ((1, 1), 200),
+ ((1, 0), 200),
+ ((2, 0), 415)
]:
with self.subTest(version=version):
- self._get('/test/api/vtest/vmethodv1_1', version=version)
+ self._get('/test/api/vtest/vmethodv1_1',
+ version=APIVersion._make(version))
self.assertStatus(expected_status)
projects / ceph.git / commitdiff