openstack
/
qinling
Code Issues Proposed changes

Filter executions by status 

Partially implements: blueprint qinling-execution-filter

Change-Id: Ia5ef99559044d5f8cd3803caba417e8def0872c8
This commit is contained in:
Lingxian Kong 2017-12-18 23:26:58 +13:00
parent 072605fa63
commit 96eb2a1416
2 changed files with 14 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
qinling/api/controllers/v1/execution.py
View File

@ -64,14 +64,17 @@ class ExecutionsController(rest.RestController):
return resources.Execution.from_dict(db_model.to_dict())
@rest_utils.wrap_wsme_controller_exception
@wsme_pecan.wsexpose(resources.Executions, wtypes.text, bool, wtypes.text)
def get_all(self, function_id=None, all_projects=False, project_id=None):
@wsme_pecan.wsexpose(resources.Executions, wtypes.text, bool, wtypes.text,
wtypes.text)
def get_all(self, function_id=None, all_projects=False, project_id=None,
status=None):
"""Return a list of executions.
:param function_id: Optional. Filtering executions by function_id.
:param project_id: Optional. Admin user can query other projects
resources, the param is ignored for normal user.
:param all_projects: Optional. Get resources of all projects.
:param status: Optional. Filter by execution status.
"""
ctx = context.get_ctx()
if project_id and not ctx.is_admin:
@ -85,6 +88,7 @@ class ExecutionsController(rest.RestController):
filters = rest_utils.get_filters(
function_id=function_id,
project_id=project_id,
status=status,
)
LOG.info("Get all %ss. filters=%s", self.type, filters)

10
qinling/tests/unit/api/controllers/v1/test_execution.py
View File

@ -77,17 +77,23 @@ class TestExecutionController(base.APITest):
}
resp = self.app.post_json('/v1/executions', body)
exec_id = resp.json.get('id')
self.assertEqual(201, resp.status_int)
# Test filtering by 'function_id'
resp = self.app.get('/v1/executions?function_id=%s' % self.func_id)
self.assertEqual(200, resp.status_int)
actual = self._assert_single_item(
resp.json['executions'], id=exec_id
)
self._assertDictContainsSubset(actual, body)
# Test filtering by 'status'
resp = self.app.get(
'/v1/executions?function_id=%s&status=running' % self.func_id
)
self.assertEqual(200, resp.status_int)
self._assert_single_item(resp.json['executions'], id=exec_id)
@mock.patch('qinling.rpc.EngineClient.create_execution')
def test_delete(self, mock_create_execution):
body = {