-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'
from __future__ import absolute_import
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
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, Endpoint, Router
from ._version import APIVersion
}
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 mgr
+from .. import DEFAULT_API_VERSION, 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_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")
self.status = 500
self.body = json.dumps(thread.res_task['exception'])
- 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_API_VERSION
from ..controllers import APIRouter, BaseController, Proxy, RESTController, Router
from ..controllers._version import APIVersion
from ..services.exception import handle_rados_error
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