File: base.py

package info (click to toggle)
python-botocore 1.12.103%2Brepack-1
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 41,552 kB
  • sloc: python: 43,119; xml: 15,052; makefile: 131
file content (104 lines) | stat: -rw-r--r-- 3,596 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import json

from botocore import xform_name
from botocore.exceptions import ClientError

from behave import when, then
from nose.tools import assert_equal


def _params_from_table(table):
    # Unfortunately the way we're using table is not quite how
    # behave expects tables to be used:
    # They expect:
    #
    #     | name      | department  |
    #     | Barry     | foo         |
    #     | Pudey     | bar         |
    #     | Two-Lumps | bar         |
    #
    # Where the first row are headings that indicate the
    # key name you can use to retrieve row values,
    # e.g row['name'] -> Barry.
    #
    #
    # We just use:
    #      | LaunchConfigurationName | hello, world |
    #      | ImageId                 | ami-12345678 |
    #      | InstanceType            | m1.small     |
    #
    # So we have to grab the headings before iterating over
    # the table rows.
    params = {table.headings[0]: table.headings[1]}
    for row in table:
        params[row[0]] = row[1]
    return params


@when(u'I call the "{}" API')
def api_call_no_args(context, operation):
    context.response = getattr(context.client, xform_name(operation))()


@when(u'I call the "{}" API with')
def api_call_with_args(context, operation):
    params = _params_from_table(context.table)
    context.response = getattr(context.client, xform_name(operation))(**params)


@when(u'I call the "{}" API with JSON')
def api_call_with_json(context, operation):
    params = json.loads(context.text)
    context.response = getattr(context.client, xform_name(operation))(**params)


@when(u'I attempt to call the "{}" API with')
def api_call_with_error(context, operation):
    params = _params_from_table(context.table)
    try:
        getattr(context.client, xform_name(operation))(**params)
    except ClientError as e:
        context.error_response = e


@when(u'I attempt to call the "{}" API with JSON')
def api_call_with_json_and_error(context, operation):
    params = json.loads(context.text)
    try:
        getattr(context.client, xform_name(operation))(**params)
    except ClientError as e:
        context.error_response = e


@then(u'I expect the response error code to be "{}"')
def then_expected_error(context, code):
    assert_equal(context.error_response.response['Error']['Code'], code)


@then(u'the value at "{}" should be a list')
def then_expected_type_is_list(context, expression):
    # In botocore, if there are no values with an element,
    # it will not appear in the response dict, so it's actually
    # ok if the element does not exist (and is not a list).
    # If an exception happened the test will have already failed,
    # which makes this step a noop.  We'll just verify
    # the response is a dict to ensure it made it through
    # our response parser properly.
    if not isinstance(context.response, dict):
        raise AssertionError("Response is not a dict: %s" % context.response)


@then(u'the response should contain a "{}"')
def then_should_contain_key(context, key):
    # See then_expected_type_is_a_list for more background info.
    # We really just care that the request succeeded for these
    # smoke tests.
    if not isinstance(context.response, dict):
        raise AssertionError("Response is not a dict: %s" % context.response)


@then(u'I expect the response error to contain a message')
def then_error_has_message(context):
    if 'Message' not in context.error_response.response['Error']:
        raise AssertionError("Message key missing from error response: %s" %
                             context.error_response.response)