Function Alias API: delete and update

This patch supports function alias delete and update API.

- Function alias delete
- Function alias update
- Unit test for the API

Change-Id: I8341ef2a60ef0696ffd2095281cbb7fae3b15f5e
story: #2002143
task: #19987

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)

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)

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)

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)

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)