-DEFAULT_VERSION = '1.0'
+DEFAULT_API_VERSION = '1.0'
from teuthology.exceptions import \
CommandFailedError # pylint: disable=import-error
-from . import DEFAULT_VERSION
+from . import DEFAULT_API_VERSION
log = logging.getLogger(__name__)
# pylint: disable=inconsistent-return-statements, too-many-arguments, too-many-branches
@classmethod
- def _request(cls, url, method, data=None, params=None, version=DEFAULT_VERSION,
+ def _request(cls, url, method, data=None, params=None, version=DEFAULT_API_VERSION,
set_cookies=False):
url = "{}{}".format(cls._base_uri, url)
log.debug("Request %s to %s", method, url)
raise ex
@classmethod
- def _get(cls, url, params=None, version=DEFAULT_VERSION, set_cookies=False):
+ def _get(cls, url, params=None, version=DEFAULT_API_VERSION, set_cookies=False):
return cls._request(url, 'GET', params=params, version=version, set_cookies=set_cookies)
@classmethod
retry = True
while retry and retries > 0:
retry = False
- res = cls._get(url, version=DEFAULT_VERSION)
+ res = cls._get(url, version=DEFAULT_API_VERSION)
if isinstance(res, dict):
res = [res]
for view in res:
return res
@classmethod
- def _post(cls, url, data=None, params=None, version=DEFAULT_VERSION, set_cookies=False):
+ def _post(cls, url, data=None, params=None, version=DEFAULT_API_VERSION, set_cookies=False):
cls._request(url, 'POST', data, params, version=version, set_cookies=set_cookies)
@classmethod
- def _delete(cls, url, data=None, params=None, version=DEFAULT_VERSION, set_cookies=False):
+ def _delete(cls, url, data=None, params=None, version=DEFAULT_API_VERSION, set_cookies=False):
cls._request(url, 'DELETE', data, params, version=version, set_cookies=set_cookies)
@classmethod
- def _put(cls, url, data=None, params=None, version=DEFAULT_VERSION, set_cookies=False):
+ def _put(cls, url, data=None, params=None, version=DEFAULT_API_VERSION, set_cookies=False):
cls._request(url, 'PUT', data, params, version=version, set_cookies=set_cookies)
@classmethod
# pylint: disable=too-many-arguments
@classmethod
- def _task_request(cls, method, url, data, timeout, version=DEFAULT_VERSION, set_cookies=False):
+ def _task_request(cls, method, url, data, timeout, version=DEFAULT_API_VERSION,
+ set_cookies=False):
res = cls._request(url, method, data, version=version, set_cookies=set_cookies)
cls._assertIn(cls._resp.status_code, [200, 201, 202, 204, 400, 403, 404])
return res_task['exception']
@classmethod
- def _task_post(cls, url, data=None, timeout=60, version=DEFAULT_VERSION, set_cookies=False):
+ def _task_post(cls, url, data=None, timeout=60, version=DEFAULT_API_VERSION, set_cookies=False):
return cls._task_request('POST', url, data, timeout, version=version,
set_cookies=set_cookies)
@classmethod
- def _task_delete(cls, url, timeout=60, version=DEFAULT_VERSION, set_cookies=False):
+ def _task_delete(cls, url, timeout=60, version=DEFAULT_API_VERSION, set_cookies=False):
return cls._task_request('DELETE', url, None, timeout, version=version,
set_cookies=set_cookies)
@classmethod
- def _task_put(cls, url, data=None, timeout=60, version=DEFAULT_VERSION, set_cookies=False):
+ def _task_put(cls, url, data=None, timeout=60, version=DEFAULT_API_VERSION, set_cookies=False):
return cls._task_request('PUT', url, data, timeout, version=version,
set_cookies=set_cookies)
import unittest
-from . import DEFAULT_VERSION
+from . import DEFAULT_API_VERSION
from .helper import DashboardTestCase
class VersionReqTest(DashboardTestCase, unittest.TestCase):
def test_version(self):
for (version, expected_status) in [
- (DEFAULT_VERSION, 200),
+ (DEFAULT_API_VERSION, 200),
(None, 415),
("99.99", 415)
]:
from __future__ import absolute_import
-from . import DEFAULT_VERSION
+from . import DEFAULT_API_VERSION
from .helper import DashboardTestCase
self._get('/api/summary')
self.assertHeaders({
'Content-Encoding': 'gzip',
- 'Content-Type': 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_VERSION)
+ 'Content-Type': 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION)
})
def test_force_no_gzip(self):
self._get('/api/summary')
self.assertHeaders({
'server': 'Ceph-Dashboard',
- 'Content-Type': 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_VERSION),
+ 'Content-Type': 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION),
'Content-Security-Policy': "frame-ancestors 'self';",
'X-Content-Type-Options': 'nosniff',
'Strict-Transport-Security': 'max-age=63072000; includeSubDomains; preload'
"""
import os
+import re
+from typing import NamedTuple
import cherrypy
-DEFAULT_VERSION = '1.0'
+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
# pylint: disable=wrong-import-position
import cherrypy
# pylint: disable=import-error
-from ceph_argparse import ArgumentFormat # type: ignore
+from ceph_argparse import ArgumentFormat
-from .. import DEFAULT_VERSION
+from .. import DEFAULT_API_VERSION, APIVersion
from ..api.doc import SchemaInput, SchemaType
from ..exceptions import DashboardException, PermissionNotValid, ScopeNotValid
from ..plugins import PLUGIN_MANAGER
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_VERSION):
+ json_response=True, proxy=False, xml=False, version=DEFAULT_API_VERSION):
if method is None:
method = 'GET'
elif not isinstance(method, str) or \
version):
@wraps(func)
def inner(*args, **kwargs):
- req_version = None
+ client_version = None
for key, value in kwargs.items():
if isinstance(value, str):
kwargs[key] = unquote(value)
kwargs.update(params)
if version is not None:
+ server_version = APIVersion.to_tuple(version)
accept_header = cherrypy.request.headers.get('Accept')
- if accept_header and accept_header.startswith('application/vnd.ceph.api.v'):
- req_match = re.search(r"\d\.\d", accept_header)
- if req_match:
- req_version = req_match[0]
+ 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")
- if req_version and req_version == version:
+ if server_version.supports(client_version):
ret = func(*args, **kwargs)
else:
raise cherrypy.HTTPError(415,
"Incorrect version: "
"{} requested but {} is expected"
- "".format(req_version, version))
+ "".format(
+ f'{client_version.major}.{client_version.minor}',
+ version))
else:
ret = func(*args, **kwargs)
if isinstance(ret, bytes):
}
_method_mapping = collections.OrderedDict([
- ('list', {'method': 'GET', 'resource': False, 'status': 200, 'version': DEFAULT_VERSION}),
- ('create', {'method': 'POST', 'resource': False, 'status': 201, 'version': DEFAULT_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('bulk_set', {'method': 'PUT', 'resource': False, 'status': 200, 'version': DEFAULT_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('bulk_delete', {'method': 'DELETE', 'resource': False, 'status': 204, 'version': DEFAULT_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('get', {'method': 'GET', 'resource': True, 'status': 200, 'version': DEFAULT_VERSION}),
- ('delete', {'method': 'DELETE', 'resource': True, 'status': 204, 'version': DEFAULT_VERSION}), # noqa E501 #pylint: disable=line-too-long
- ('set', {'method': 'PUT', 'resource': True, 'status': 200, 'version': DEFAULT_VERSION}),
- ('singleton_set', {'method': 'PUT', 'resource': False, 'status': 200, 'version': DEFAULT_VERSION}) # noqa E501 #pylint: disable=line-too-long
+ ('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
])
@classmethod
'method': None,
'query_params': None,
'path': '',
- 'version': DEFAULT_VERSION,
+ 'version': DEFAULT_API_VERSION,
'sec_permissions': hasattr(func, '_security_permissions'),
'permission': None,
}
@staticmethod
def Resource(method=None, path=None, status=None, query_params=None, # noqa: N802
- version=DEFAULT_VERSION):
+ version=DEFAULT_API_VERSION):
if not method:
method = 'GET'
return _wrapper
@staticmethod
- def MethodMap(resource=False, status=None, version=DEFAULT_VERSION): # noqa: N802
+ def MethodMap(resource=False, status=None, version=DEFAULT_API_VERSION): # noqa: N802
if status is None:
status = 200
@staticmethod
def Collection(method=None, path=None, status=None, query_params=None, # noqa: N802
- version=DEFAULT_VERSION):
+ version=DEFAULT_API_VERSION):
if not method:
method = 'GET'
import cherrypy
-from .. import DEFAULT_VERSION, mgr
+from .. import DEFAULT_API_VERSION, mgr
from ..api.doc import Schema, SchemaInput, SchemaType
from . import ENDPOINT_MAP, BaseController, Controller, Endpoint
}
if not version:
- version = DEFAULT_VERSION
+ version = DEFAULT_API_VERSION
if method.lower() == 'get':
resp['200'] = {'description': "OK",
}
}
+import { CdHelperClass } from '~/app/shared/classes/cd-helper.class';
import { Permissions } from '~/app/shared/models/permissions';
let auth: any;
cy.request({
method: 'POST',
url: 'api/auth',
- headers: { Accept: 'application/vnd.ceph.api.v1.0+json' },
+ headers: { Accept: CdHelperClass.cdVersionHeader('1', '0') },
body: { username: username, password: password }
}).then((resp) => {
auth = resp.body;
import { InventoryDevice } from '~/app/ceph/cluster/inventory/inventory-devices/inventory-device.model';
import { InventoryHost } from '~/app/ceph/cluster/inventory/inventory-host.model';
+import { CdHelperClass } from '~/app/shared/classes/cd-helper.class';
import { Daemon } from '../models/daemon.interface';
import { CdDevice } from '../models/devices';
import { SmartDataResponseV1 } from '../models/smart';
return this.http.post(
this.baseURL,
{ hostname: hostname, addr: addr, labels: labels, status: status },
- { observe: 'response', headers: { Accept: 'application/vnd.ceph.api.v0.1+json' } }
+ { observe: 'response', headers: { Accept: CdHelperClass.cdVersionHeader('0', '1') } }
);
}
return hasChanges;
}
+
+ static cdVersionHeader(major_ver: string, minor_ver: string) {
+ return `application/vnd.ceph.api.v${major_ver}.${minor_ver}+json`;
+ }
}
import { Observable, throwError as observableThrowError } from 'rxjs';
import { catchError } from 'rxjs/operators';
+import { CdHelperClass } from '~/app/shared/classes/cd-helper.class';
import { NotificationType } from '../enum/notification-type.enum';
import { CdNotificationConfig } from '../models/cd-notification';
import { FinishedTask } from '../models/finished-task';
) {}
intercept(request: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
- const defaultVersion = '1.0';
const acceptHeader = request.headers.get('Accept');
let reqWithVersion: HttpRequest<any>;
if (acceptHeader && acceptHeader.startsWith('application/vnd.ceph.api.v')) {
} else {
reqWithVersion = request.clone({
setHeaders: {
- Accept: `application/vnd.ceph.api.v${defaultVersion}+json`
+ Accept: CdHelperClass.cdVersionHeader('1', '0')
}
});
}
from mgr_module import HandleCommandResult
from pyfakefs import fake_filesystem
-from .. import DEFAULT_VERSION, mgr
+from .. import DEFAULT_API_VERSION, mgr
from ..controllers import generate_controller_routes, json_error_page
from ..module import Module
from ..plugins import PLUGIN_MANAGER, debug, feature_toggles # noqa
if cls._request_logging:
cherrypy.config.update({'tools.request_logging.on': False})
- def _request(self, url, method, data=None, headers=None, version=DEFAULT_VERSION):
+ def _request(self, url, method, data=None, headers=None, version=DEFAULT_API_VERSION):
if not data:
b = None
if version:
h = headers
self.getPage(url, method=method, body=b, headers=h)
- def _get(self, url, headers=None, version=DEFAULT_VERSION):
+ def _get(self, url, headers=None, version=DEFAULT_API_VERSION):
self._request(url, 'GET', headers=headers, version=version)
- def _post(self, url, data=None, version=DEFAULT_VERSION):
+ def _post(self, url, data=None, version=DEFAULT_API_VERSION):
self._request(url, 'POST', data, version=version)
- def _delete(self, url, data=None, version=DEFAULT_VERSION):
+ def _delete(self, url, data=None, version=DEFAULT_API_VERSION):
self._request(url, 'DELETE', data, version=version)
- def _put(self, url, data=None, version=DEFAULT_VERSION):
+ def _put(self, url, data=None, version=DEFAULT_API_VERSION):
self._request(url, 'PUT', data, version=version)
- def _task_request(self, method, url, data, timeout, version=DEFAULT_VERSION):
+ def _task_request(self, method, url, data, timeout, version=DEFAULT_API_VERSION):
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_VERSION):
+ def _task_post(self, url, data=None, timeout=60, version=DEFAULT_API_VERSION):
self._task_request('POST', url, data, timeout, version=version)
- def _task_delete(self, url, timeout=60, version=DEFAULT_VERSION):
+ def _task_delete(self, url, timeout=60, version=DEFAULT_API_VERSION):
self._task_request('DELETE', url, None, timeout, version=version)
- def _task_put(self, url, data=None, timeout=60, version=DEFAULT_VERSION):
+ def _task_put(self, url, data=None, timeout=60, version=DEFAULT_API_VERSION):
self._task_request('PUT', url, data, timeout, version=version)
def json_body(self):
except ImportError:
from unittest.mock import patch
-from .. import DEFAULT_VERSION
+from .. import DEFAULT_API_VERSION
from ..controllers import ApiController, 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
self._get("/foo")
self.assertStatus('200 OK')
self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_VERSION))
+ 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
self.assertBody('[]')
def test_fill(self):
self.assertJsonBody(data)
self.assertStatus(201)
self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_VERSION))
+ 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
self._get("/foo")
self.assertStatus('200 OK')
self.assertHeader('Content-Type',
- 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_VERSION))
+ 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
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_VERSION))
+ 'application/vnd.ceph.api.v{}+json'.format(DEFAULT_API_VERSION))
self.assertJsonBody({'newdata': 'newdata', 'key': '0'})
def test_not_implemented(self):
def vmethod(self):
return {'version': '1.0'}
+ @RESTController.Collection('GET', version="1.1")
+ def vmethodv1_1(self):
+ return {'version': '1.1'}
+
@RESTController.Collection('GET', version="2.0")
def vmethodv2(self):
return {'version': '2.0'}
with self.subTest(version=version):
self._get('/test/api/vtest/vmethodv2', version=version)
self.assertStatus(expected_status)
+
+ def test_backward_compatibility(self):
+ for (version, expected_status) in [
+ ("1.1", 200),
+ ("1.0", 200),
+ ("2.0", 415)
+ ]:
+ with self.subTest(version=version):
+ self._get('/test/api/vtest/vmethodv1_1', version=version)
+ self.assertStatus(expected_status)
projects / ceph.git / commitdiff