from __future__ import absolute_import
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__)),
--- /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 ..security import Scope
from ..services.ceph_service import CephService
from . import APIDoc, APIRouter, Endpoint, EndpointDoc, ReadPermission, RESTController, UIRouter
-from ._version import APIVersion
LIST_SCHEMA = {
"rule_id": (int, 'Rule ID'),
from __future__ import absolute_import
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, Endpoint, Router
from ._version import APIVersion
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 = []
+++ /dev/null
-# # -*- coding: utf-8 -*-
-
-from ..exceptions import DashboardException
-from ..model.feedback import Feedback
-from ..rest_client import RequestException
-from ..security import Scope
-from ..services import feedback
-from . import APIDoc, APIRouter, RESTController
-
-
-@APIRouter('/feedback', Scope.CONFIG_OPT)
-@APIDoc("Feedback API", "Report")
-class FeedbackController(RESTController):
- issueAPIkey = None
-
- def __init__(self): # pragma: no cover
- super().__init__()
- self.tracker_client = feedback.CephTrackerClient()
-
- def create(self, project, tracker, subject, description):
- """
- Create an issue.
- :param project: The affected ceph component.
- :param tracker: The tracker type.
- :param subject: The title of the issue.
- :param description: The description of the issue.
- """
- try:
- new_issue = Feedback(Feedback.Project[project].value,
- Feedback.TrackerType[tracker].value, subject, description)
- except KeyError:
- raise DashboardException(msg=f'{"Invalid arguments"}', component='feedback')
- try:
- return self.tracker_client.create_issue(new_issue)
- except RequestException as error:
- if error.status_code == 401:
- raise DashboardException(msg=f'{"Invalid API key"}',
- http_status_code=error.status_code,
- component='feedback')
- raise error
- except Exception:
- raise DashboardException(msg=f'{"API key not set"}',
- http_status_code=401,
- component='feedback')
-
- def get(self, issue_number):
- """
- Fetch issue details.
- :param issueAPI: The issue tracker API access key.
- """
- try:
- return self.tracker_client.get_issues(issue_number)
- except RequestException as error:
- if error.status_code == 404:
- raise DashboardException(msg=f'Issue {issue_number} not found',
- http_status_code=error.status_code,
- component='feedback')
- raise error
'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.v1.0+json:
- type: object
- description: Resource created.
- '202':
- content:
- application/vnd.ceph.api.v1.0+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.v1.0+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 .. import mgr
from ..controllers import generate_controller_routes, json_error_page
from ..controllers._version import APIVersion
from ..module import Module
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")
self.status = 500
self.body = json.dumps(thread.res_task['exception'])
- 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):
},
)
@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):
'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 APIRouter, BaseController, Proxy, RESTController, Router
from ..controllers._version import APIVersion
from ..services.exception import handle_rados_error
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):
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