Source code for

# -*- coding: utf-8 -*-
"""Helper functions for tests using REST API."""
import pytest
from collections import namedtuple

from cfme.exceptions import OptionNotAvailable
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. If specific http_status is not given, we simply check that the status was a successful response code via requests.Response.__bool__() If response status code is '204', ensures there is no content. Example of verifying a success response: .. code-block:: python assert_response(appliance) Example of verifying a failure response: .. code-block:: python with error.expected('ActiveRecord::RecordNotFound'): collection.action.delete(some_stuff) assert_response(appliance, http_status=404) Note: For below args, 'results' refers to rest_obj.last_response.json()['results'] Args: rest_obj -- instance of cfme.utils.Appliance (with rest_api attr defined) or cfme.utils.appliance.MiqApi success -- if defined, checks each result in results to ensure that result['success'] is equal to the value defined here http_status (int or tuple of int) -- expected http response status codes, if None, we simply verify that the response was a success results_num (int) -- specifies expected number of results task_wait (int) -- if any result in results contains a 'task_id', this method will polls the API to ensure that task has moved to 'finished' and wait 'task_wait' seconds for that state change to occur """ # 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: # Convert single int to tuple if needed if isinstance(http_status, int): http_status = (http_status,) # Check the response codes assert last_response.status_code in http_status,\ 'The status code {} doesn\'t match the expected status code {}'.format( last_response.status_code, http_status) else: # No specific status_code specified, simply check if response was a success 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 JSON content returned') def _check_result(result): # check that result contains data to catch bugs like BZ 1414845 assert result, 'The result should not be empty' 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']) task.wait_exists(num_sec=5) wait_for( lambda: task.state.lower() == 'finished', fail_func=task.reload, num_sec=task_wait, message='task state finished', ) task_message = getattr(task, 'message', '') assert task.status.lower() == 'ok', ( 'Task failed with status "{}", message "{}"'.format(task.status, task_message)) 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(service): """Gets list of vm entities associated with the service.""" rest_api = service.collection._api 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]
[docs]def create_resource(rest_api, col_name, col_data, col_action='create', substr_search=False): """Creates new resource in collection.""" collection = getattr(rest_api.collections, col_name) try: action = getattr(collection.action, col_action) except AttributeError: raise OptionNotAvailable( "Action `{}` for {} is not implemented in this version".format(col_action, col_name)) entities = action(*col_data) action_response = rest_api.response search_str = '%{}%' if substr_search else '{}' for entity in col_data: if entity.get('name'): wait_for(lambda: collection.find_by( name=search_str.format(entity.get('name'))) or False, num_sec=180, delay=10) elif entity.get('description'): wait_for(lambda: collection.find_by( description=search_str.format(entity.get('description'))) or False, num_sec=180, delay=10) else: raise NotImplementedError # make sure action response is preserved rest_api.response = action_response return entities
[docs]def delete_resources_from_collection( resources, collection=None, not_found=None, num_sec=10, delay=2, check_response=True): """Checks that delete from collection works as expected.""" collection = collection or resources[0].collection def _assert_response(*args, **kwargs): if check_response: assert_response(collection._api, *args, **kwargs) collection.action.delete(*resources) _assert_response() for resource in resources: resource.wait_not_exists(num_sec=num_sec, delay=delay) def _response_success_false(): collection.action.delete(*resources) _assert_response(success=False) current_version = collection._api.server_info.get('version') if not_found is False: _response_success_false() # On version < 5.9 when one resource in a list of resources to delete is missing, # the whole request fails with 'RecordNotFound'. It's not the case only # for resources in the "configuration_script_sources" collection ATM. elif not_found or current_version < '5.9': with pytest.raises(Exception, match='ActiveRecord::RecordNotFound'): collection.action.delete(*resources) _assert_response(http_status=404) else: _response_success_false()
[docs]def delete_resources_from_detail( resources, method='POST', num_sec=10, delay=2, check_response=True): """Checks that delete from detail works as expected.""" method = method.upper() rest_api = resources[0].collection._api def _assert_response(*args, **kwargs): if check_response: assert_response(rest_api, *args, **kwargs) for resource in resources: getattr(resource.action.delete, method)() _assert_response() # Wait for resource non-existence in separate loop so the delete actions are # not delayed by waiting for the previously deleted resource to disappear. # This way the combined wait time is likely to be much shorter. for resource in resources: resource.wait_not_exists(num_sec=num_sec, delay=delay) with pytest.raises(Exception, match='ActiveRecord::RecordNotFound'): getattr(resource.action.delete, method)() _assert_response(http_status=404)
[docs]def query_resource_attributes(resource, soft_assert=None): """Checks that all available attributes/subcollections are really accessible.""" collection = resource.collection rest_api = collection._api options = rest_api.options(collection._href) attrs_to_query = options['virtual_attributes'] + options['relationships'] subcolls_to_check = options['subcollections'] FailedRecord = namedtuple('FailedRecord', ['name', 'type', 'error', 'response']) service_href = resource.href failed = [] missing = [] for attr in attrs_to_query: try: response = rest_api.get('{}?attributes={}'.format(service_href, attr)) assert rest_api.response, 'Failed response' except Exception as err: failed.append(FailedRecord(attr, 'attribute', err, rest_api.response)) continue if attr not in response: missing.append(attr) for subcol in subcolls_to_check: try: subcol_rest = getattr(resource, subcol) subcol_rest.reload() except Exception as err: failed.append(FailedRecord(subcol, 'subcollection', err, rest_api.response)) outcome = namedtuple('AttrCheck', ['failed', 'missing'])(failed, missing) if soft_assert: for failure in outcome.failed: soft_assert(False, '{0} "{1}": status: {2}, error: `{3}`'.format( failure.type,, failure.response.status_code, failure.error)) return outcome