Browse Source

Merge "Function Alias API: delete and update"

changes/69/580569/1
Zuul 3 years ago
committed by Gerrit Code Review
parent
commit
7dbef0f030
  1. 43
      qinling/api/controllers/v1/function_alias.py
  2. 2
      qinling/api/controllers/v1/resources.py
  3. 8
      qinling/db/api.py
  4. 14
      qinling/db/sqlalchemy/api.py
  5. 56
      qinling/tests/unit/api/controllers/v1/test_function_alias.py

43
qinling/api/controllers/v1/function_alias.py

@ -30,6 +30,7 @@ LOG = logging.getLogger(__name__)
CONF = cfg.CONF
POST_REQUIRED = set(['name', 'function_id'])
UPDATE_ALLOWED = set(['function_id', 'function_version', 'description'])
class FunctionAliasesController(rest.RestController):
@ -111,3 +112,45 @@ class FunctionAliasesController(rest.RestController):
for db_model in db_aliases]
return resources.FunctionAliases(function_aliases=aliases)
@wsme_pecan.wsexpose(None, wtypes.text, status_code=204)
def delete(self, alias_name):
"""Delete a specific alias.
"""
ctx = context.get_ctx()
acl.enforce('function_alias:delete', ctx)
LOG.info("Deleting alias %s.", alias_name)
db_api.delete_function_alias(alias_name)
LOG.info("Alias %s deleted.", alias_name)
@rest_utils.wrap_wsme_controller_exception
@wsme_pecan.wsexpose(
resources.FunctionAlias,
wtypes.text,
body=resources.FunctionAlias,
)
def put(self, alias_name, body):
"""Update alias for the specified function.
The supported body params:
- function_id: Optional. Function id the alias point to.
- function_version: Optional. Version number the alias point to.
- description: Optional. The description of the alias.
"""
ctx = context.get_ctx()
acl.enforce('function_alias:update', ctx)
params = body.to_dict()
values = {}
for key in UPDATE_ALLOWED:
if params.get(key) is not None:
values.update({key: params[key]})
LOG.info("Updating Alias %s, params: %s", alias_name, values)
alias = db_api.update_function_alias(alias_name, **values)
LOG.info("Alias updated.")
return resources.FunctionAlias.from_db_obj(alias)

2
qinling/api/controllers/v1/resources.py

@ -395,7 +395,7 @@ class FunctionAlias(Resource):
name = wtypes.text
description = wtypes.text
function_id = types.uuid
function_version = wsme.wsattr(int, default=0)
function_version = wsme.wsattr(int)
project_id = wsme.wsattr(wtypes.text, readonly=True)
created_at = wsme.wsattr(wtypes.text, readonly=True)
updated_at = wsme.wsattr(wtypes.text, readonly=True)

8
qinling/db/api.py

@ -238,6 +238,14 @@ def get_function_aliases(**kwargs):
return IMPL.get_function_aliases(**kwargs)
def update_function_alias(name, **kwargs):
return IMPL.update_function_alias(name, **kwargs)
def delete_function_alias(name, **kwargs):
return IMPL.delete_function_alias(name, **kwargs)
# For unit test
def delete_function_aliases(**kwargs):
return IMPL.delete_function_aliases(**kwargs)

14
qinling/db/sqlalchemy/api.py

@ -600,6 +600,20 @@ def get_function_aliases(session=None, **kwargs):
return _get_collection_sorted_by_time(models.FunctionAlias, **kwargs)
@db_base.session_aware()
def update_function_alias(name, session=None, **kwargs):
alias_db = get_function_alias(name, session=session)
alias_db.update(kwargs.copy())
return alias_db
@db_base.session_aware()
def delete_function_alias(name, session=None):
alias_db = get_function_alias(name)
session.delete(alias_db)
@db_base.session_aware()
def delete_function_aliases(session=None, **kwargs):
return _delete_all(models.FunctionAlias, **kwargs)

56
qinling/tests/unit/api/controllers/v1/test_function_alias.py

@ -14,6 +14,7 @@
from qinling import context
from qinling.db import api as db_api
from qinling import exceptions as exc
from qinling.tests.unit.api import base
from qinling.tests.unit import base as unit_base
@ -96,3 +97,58 @@ class TestFunctionAliasController(base.APITest):
actual = self._assert_single_item(resp.json['function_aliases'],
name=name)
self._assertDictContainsSubset(actual, expected)
def test_delete(self):
name = self.rand_name(name="alias", prefix=self.prefix)
function_version = 0
body = {'function_id': self.func_id,
'function_version': function_version,
'name': name,
'description': 'new alias'}
db_api.create_function_alias(**body)
resp = self.app.delete('/v1/aliases/%s' % name)
self.assertEqual(204, resp.status_int)
context.set_ctx(self.ctx)
self.assertRaises(exc.DBEntityNotFoundError,
db_api.get_function_alias,
name)
def test_put(self):
name = self.rand_name(name="alias", prefix=self.prefix)
function_version = 0
body = {'function_id': self.func_id,
'function_version': function_version,
'name': name,
'description': 'new alias'}
db_api.create_function_alias(**body)
body['function_version'] = 1
body['description'] = 'update alias'
resp = self.app.put_json('/v1/aliases/%s' % name, body)
self.assertEqual(200, resp.status_int)
self._assertDictContainsSubset(resp.json, body)
def test_put_without_optional_params(self):
name = self.rand_name(name="alias", prefix=self.prefix)
function_version = 1
body = {'function_id': self.func_id,
'function_version': function_version,
'name': name,
'description': 'new alias'}
db_api.create_function_alias(**body)
update_body = {}
resp = self.app.put_json('/v1/aliases/%s' % name, update_body)
self.assertEqual(200, resp.status_int)
self._assertDictContainsSubset(resp.json, body)
Loading…
Cancel
Save