test_get_target_region.py 7.08 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
#
# Copyright 2017-2018 Amazon.com, Inc. and its affiliates. All Rights Reserved.
#
# Licensed under the MIT License. See the LICENSE accompanying this file
# for the specific language governing permissions and limitations under
# the License.
#

import mount_efs
import json

import pytest

14
15
from .. import utils

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
try:
    import ConfigParser
except ImportError:
    from configparser import ConfigParser

try:
    from urllib2 import URLError, HTTPError
except ImportError:
    from urllib.error import URLError, HTTPError

INSTANCE_DATA = {
  'devpayProductCodes': None,
  'privateIp': '192.168.1.1',
  'availabilityZone': 'us-east-1a',
  'version': '2010-08-31',
  'instanceId': 'i-deadbeefdeadbeef0',
  'billingProducts': None,
  'pendingTime': '2017-06-20T18:32:00Z',
  'instanceType': 'm3.xlarge',
  'accountId': '123412341234',
  'architecture': 'x86_64',
  'kernelId': None,
  'ramdiskId': None,
  'imageId': 'ami-deadbeef',
  'region': 'us-east-1'
}

TARGET_REGION = 'us-east-1'
INSTANCE_DOCUMENT = json.dumps(INSTANCE_DATA)
DNS_NAME_SUFFIX = 'amazonaws.com'


class MockHeaders(object):
    def __init__(self, content_charset=None):
        self.content_charset = content_charset

    def get_content_charset(self):
        return self.content_charset


class MockUrlLibResponse(object):
    def __init__(self, code=200, data=INSTANCE_DOCUMENT, headers=MockHeaders()):
        self.code = code
        self.data = data
        self.headers = headers

    def getcode(self):
        return self.code

    def read(self):
        return self.data


def get_config(dns_name_format, region=None):
    try:
        config = ConfigParser.SafeConfigParser()
    except AttributeError:
        config = ConfigParser()
    config.add_section(mount_efs.CONFIG_SECTION)
    config.set(mount_efs.CONFIG_SECTION, 'dns_name_format', dns_name_format)
    if region:
        config.set(mount_efs.CONFIG_SECTION, 'region', region)
    return config


def get_target_region_helper():
82
    config = get_config('{az}.{fs_id}.efs.{region}.{dns_name_suffix}')
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
    return mount_efs.get_target_region(config)


"""
Get target region from ec2 instance metadata
"""
def test_get_target_region_with_token(mocker):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value='ABCDEFG==')
    mocker.patch('mount_efs.urlopen', return_value=MockUrlLibResponse())
    assert 'us-east-1' == get_target_region_helper()


def test_get_target_region_without_token(mocker):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    mocker.patch('mount_efs.urlopen', return_value=MockUrlLibResponse())
    assert 'us-east-1' == get_target_region_helper()


101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def test_get_target_region_metadata_endpoint_unauthorized(mocker):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value='ABCDEFG==')
    mocker.patch('mount_efs.urlopen', side_effect=[HTTPError('url', 401, 'Unauthorized', None, None), MockUrlLibResponse()])
    assert 'us-east-1' == get_target_region_helper()


# Reproduce https://github.com/aws/efs-utils/issues/46
def test_get_target_region_token_endpoint_not_allowed(mocker):
    get_aws_ec2_metadata_token_mock = mocker.patch('mount_efs.get_aws_ec2_metadata_token',
                                                   side_effect=HTTPError('url', 405, 'Not allowed', None, None))
    mocker.patch('mount_efs.urlopen', return_value=MockUrlLibResponse())
    assert 'us-east-1' == get_target_region_helper()
    utils.assert_not_called(get_aws_ec2_metadata_token_mock)


116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def test_get_target_region_py3_no_charset(mocker):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    mocker.patch('mount_efs.urlopen', return_value=MockUrlLibResponse(data=bytearray(INSTANCE_DOCUMENT, 'us-ascii')))
    assert 'us-east-1' == get_target_region_helper()


def test_get_target_region_py3_utf8_charset(mocker):
    charset = 'utf-8'
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    mocker.patch('mount_efs.urlopen', return_value=MockUrlLibResponse(data=bytearray(INSTANCE_DOCUMENT, charset)),
                 headers=MockHeaders(content_charset=charset))
    assert 'us-east-1' == get_target_region_helper()


def test_get_target_region_from_metadata(mocker):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    mocker.patch('mount_efs.urlopen', return_value=MockUrlLibResponse())
    config = get_config('{fs_id}.efs.{region}.{dns_name_suffix}', None)
    assert TARGET_REGION == mount_efs.get_target_region(config)


def test_get_target_region_config_metadata_unavailable(mocker, capsys):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    mocker.patch('mount_efs.urlopen', side_effect=URLError('test error'))
    config = get_config('{fs_id}.efs.{region}.{dns_name_suffix}')
    with pytest.raises(SystemExit) as ex:
      mount_efs.get_target_region(config)

    assert 0 != ex.value.code
    out, err = capsys.readouterr()
    assert 'Error retrieving region' in err

148

149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def _test_get_target_region_error(mocker, capsys, response=None, error=None):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    if (response and error) or (not response and not error):
        raise ValueError('Invalid arguments')
    elif response:
        mocker.patch('mount_efs.urlopen', return_value=response)
    elif error:
        mocker.patch('mount_efs.urlopen', side_effect=error)

    with pytest.raises(SystemExit) as ex:
        get_target_region_helper()

    assert 0 != ex.value.code

    out, err = capsys.readouterr()
    assert 'Error retrieving region' in err

166

167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def test_get_target_region_bad_response(mocker, capsys):
    _test_get_target_region_error(mocker, capsys, error=HTTPError('url', 400, 'Bad Request Error', None, None))


def test_get_target_region_error_response(mocker, capsys):
    _test_get_target_region_error(mocker, capsys, error=URLError('test error'))


def test_get_target_region_bad_json(mocker, capsys):
    _test_get_target_region_error(mocker, capsys, response=MockUrlLibResponse(data='not json'))


def test_get_target_region_missing_region(mocker, capsys):
    _test_get_target_region_error(mocker, capsys, response=MockUrlLibResponse(data=json.dumps({})))


"""
Get target region from configuration file
"""
def test_get_target_region_from_config_variable(mocker):
187
    config = get_config('{az}.{fs_id}.efs.us-east-2.{dns_name_suffix}', TARGET_REGION)
188
189
190
191
192
193
194
195
196
197
    assert TARGET_REGION == mount_efs.get_target_region(config)


def _test_get_target_region_from_dns_format(mocker, config):
    mocker.patch('mount_efs.get_aws_ec2_metadata_token', return_value=None)
    mocker.patch('mount_efs.urlopen', side_effect=URLError('test error'))
    assert TARGET_REGION == mount_efs.get_target_region(config)


def test_get_target_region_from_legacy_dns_name_format(mocker):
198
    config = get_config('{az}.{fs_id}.efs.us-east-1.amazonaws.com')
199
200
201
202
    _test_get_target_region_from_dns_format(mocker, config)


def test_get_target_region_from_suffixed_dns_name_format(mocker):
203
    config = get_config('{az}.{fs_id}.efs.us-east-1.{dns_name_suffix}')
204
    config.set(mount_efs.CONFIG_SECTION, 'dns_name_suffix', DNS_NAME_SUFFIX)
205
    _test_get_target_region_from_dns_format(mocker, config)