Source code for

# -*- coding: utf-8 -*-
"""Helper functions for tests using REST API."""

from cfme.utils.wait import wait_for

[docs]def assert_response( rest_obj, success=None, http_status=None, results_num=None, task_wait=600): """Asserts that the response HTTP status code and content is as expected.""" # check if `rest_obj` is an object with attribute referencing rest_api instance rest_api = rest_obj.rest_api if hasattr(rest_obj, 'rest_api') else rest_obj last_response = rest_api.response if http_status: assert last_response.status_code == http_status,\ 'The status code {} doesn\'t match the expected status code {}'.format( last_response.status_code, http_status) else: assert last_response, 'The request failed with {}'.format(last_response.status_code) try: content = last_response.json() except Exception: if last_response.status_code == 204: # 204 == No Content: check that message-body is empty and return assert not last_response.text.strip(), 'No content expected' return else: raise AssertionError('No content returned') def _check_result(result): if success is not None: assert 'success' in result assert result['success'] is success elif 'success' in result and last_response: # expect True if 'success' is present and HTTP status is success assert result['success'], 'The response "success" is {}'.format(result['success']) # if the request succeeded and there is a 'task_id' present in the response, # check the corresponding resource in /api/task/:task_id if task_wait and 'task_id' in result and result.get('success') and last_response: task = rest_api.get_entity('tasks', result['task_id']) wait_for( lambda: task.state.lower() == 'finished', fail_func=task.reload, num_sec=task_wait, message='task state finished', ) assert task.status.lower() == 'ok', 'Task failed with status "{}"'.format(task.status) if 'results' in content: results = content['results'] results_len = len(results) if results_num is not None: assert results_len == results_num,\ 'The number of results {} doesn\'t match the expected number {}'.format( results_len, results_num) for result in results: _check_result(result) else: _check_result(content) # preserve the original response rest_api.response = last_response
[docs]def get_vms_in_service(rest_api, service): """Gets list of vm entities associated with the service.""" service.vms.reload() # return entities under /api/vms, not under /api/services/:id/vms subcollection # where "actions" are not available return [rest_api.get_entity('vms', vm['id']) for vm in service.vms.all]