/
test_fargate.py
68 lines (56 loc) · 2.1 KB
/
test_fargate.py
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
import pytest
import json
import requests
from aws_py.ecs import Fargate
from unittest.mock import patch
@patch("requests.get")
def test_get_container_metadata_v4_successful(requests_get_function):
requests_get_function.return_value = type(
"fake_request", (), {"json": lambda: json.dumps({"A": "B"}), "status_code": 200}
)
result = Fargate().get_container_metadata_v4()
assert result == {"A": "B"}
@patch("requests.get")
def test_get_container_metadata_v4_failed_bad_json_raise_errors(requests_get_function):
requests_get_function.return_value = type(
"fake_request", (), {"json": lambda: "abc", "status_code": 200}
)
with pytest.raises(json.decoder.JSONDecodeError):
Fargate(raise_errors=True).get_container_metadata_v4()
@patch("requests.get")
def test_get_container_metadata_v4_failed_bad_json_swallow_errors(
requests_get_function,
):
requests_get_function.return_value = type(
"fake_request", (), {"json": lambda: "abc", "status_code": 200}
)
result = Fargate(raise_errors=False).get_container_metadata_v4()
assert result == {}
@patch("requests.get")
def test_get_container_metadata_v4_raises_invalid_status_code(requests_get_function):
requests_get_function.return_value = type(
"fake_request",
(),
{"json": lambda: "abc", "status_code": 401, "text": lambda: "abc"},
)
with pytest.raises(RuntimeError):
Fargate(raise_errors=True).get_container_metadata_v4()
@pytest.mark.parametrize(
"exception",
[
requests.exceptions.BaseHTTPError,
requests.exceptions.ReadTimeout,
requests.exceptions.RequestException,
requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
],
)
@patch("requests.get")
def test_get_container_metadata_v4_handles_http_exceptions(
requests_get_function, exception
):
requests_get_function.side_effect = exception
with pytest.raises(exception):
Fargate(raise_errors=True).get_container_metadata_v4()
result_no_raise = Fargate(raise_errors=False).get_container_metadata_v4()
assert result_no_raise == {}