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
|
import pytest
import requests
from aws_xray_sdk.core import patch
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core.context import Context
patch(('requests',))
# httpbin.org is created by the same author of requests to make testing http easy.
BASE_URL = 'httpbin.org'
@pytest.fixture(autouse=True)
def construct_ctx():
"""
Clean up context storage on each test run and begin a segment
so that later subsegment can be attached. After each test run
it cleans up context storage again.
"""
xray_recorder.configure(service='test', sampling=False, context=Context())
xray_recorder.clear_trace_entities()
xray_recorder.begin_segment('name')
yield
xray_recorder.clear_trace_entities()
def test_ok():
status_code = 200
url = 'http://{}/status/{}'.format(BASE_URL, status_code)
requests.get(url)
subsegment = xray_recorder.current_segment().subsegments[0]
assert subsegment.name == url
http_meta = subsegment.http
assert http_meta['request']['url'] == url
assert http_meta['request']['method'].upper() == 'GET'
assert http_meta['response']['status'] == status_code
def test_error():
status_code = 400
url = 'http://{}/status/{}'.format(BASE_URL, status_code)
requests.post(url)
subsegment = xray_recorder.current_segment().subsegments[0]
assert subsegment.name == url
assert subsegment.error
http_meta = subsegment.http
assert http_meta['request']['url'] == url
assert http_meta['request']['method'].upper() == 'POST'
assert http_meta['response']['status'] == status_code
def test_throttle():
status_code = 429
url = 'http://{}/status/{}'.format(BASE_URL, status_code)
requests.head(url)
subsegment = xray_recorder.current_segment().subsegments[0]
assert subsegment.name == url
assert subsegment.error
assert subsegment.throttle
http_meta = subsegment.http
assert http_meta['request']['url'] == url
assert http_meta['request']['method'].upper() == 'HEAD'
assert http_meta['response']['status'] == status_code
def test_fault():
status_code = 500
url = 'http://{}/status/{}'.format(BASE_URL, status_code)
requests.put(url)
subsegment = xray_recorder.current_segment().subsegments[0]
assert subsegment.name == url
assert subsegment.fault
http_meta = subsegment.http
assert http_meta['request']['url'] == url
assert http_meta['request']['method'].upper() == 'PUT'
assert http_meta['response']['status'] == status_code
def test_invalid_url():
try:
requests.get('http://doesnt.exist')
except Exception:
# prevent uncatch exception from breaking test run
pass
subsegment = xray_recorder.current_segment().subsegments[0]
assert subsegment.fault
exception = subsegment.cause['exceptions'][0]
assert exception.type == 'ConnectionError'
|
