forked from instana/python-sensor
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_ot_span.py
More file actions
130 lines (101 loc) · 3.89 KB
/
Copy pathtest_ot_span.py
File metadata and controls
130 lines (101 loc) · 3.89 KB
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
import time
import opentracing
from nose.tools import assert_equals
class TestOTSpan:
def setUp(self):
""" Clear all spans before a test run """
recorder = opentracing.tracer.recorder
recorder.clear_spans()
def tearDown(self):
""" Do nothing for now """
return None
def test_span_interface(self):
span = opentracing.tracer.start_span("blah")
assert hasattr(span, "finish")
assert hasattr(span, "set_tag")
assert hasattr(span, "tags")
assert hasattr(span, "operation_name")
assert hasattr(span, "set_baggage_item")
assert hasattr(span, "get_baggage_item")
assert hasattr(span, "context")
assert hasattr(span, "log")
def test_span_ids(self):
count = 0
while count <= 1000:
count += 1
span = opentracing.tracer.start_span("test_span_ids")
context = span.context
assert -9223372036854775808 <= context.span_id <= 9223372036854775807
assert -9223372036854775808 <= context.trace_id <= 9223372036854775807
def test_span_fields(self):
span = opentracing.tracer.start_span("mycustom")
assert_equals("mycustom", span.operation_name)
assert span.context
span.set_tag("tagone", "string")
span.set_tag("tagtwo", 150)
assert_equals("string", span.tags['tagone'])
assert_equals(150, span.tags['tagtwo'])
def test_span_queueing(self):
recorder = opentracing.tracer.recorder
count = 1
while count <= 20:
count += 1
span = opentracing.tracer.start_span("queuethisplz")
span.set_tag("tagone", "string")
span.set_tag("tagtwo", 150)
span.finish()
assert_equals(20, recorder.queue_size())
def test_sdk_spans(self):
recorder = opentracing.tracer.recorder
span = opentracing.tracer.start_span("custom_sdk_span")
span.set_tag("tagone", "string")
span.set_tag("tagtwo", 150)
span.set_tag('span.kind', "entry")
time.sleep(0.5)
span.finish()
spans = recorder.queued_spans()
assert 1, len(spans)
sdk_span = spans[0]
assert_equals('sdk', sdk_span.n)
assert_equals(None, sdk_span.p)
assert_equals(sdk_span.s, sdk_span.t)
assert sdk_span.ts
assert sdk_span.ts > 0
assert sdk_span.d
assert sdk_span.d > 0
assert_equals("py", sdk_span.ta)
assert sdk_span.data
assert sdk_span.data.sdk
assert_equals('entry', sdk_span.data.sdk.Type)
assert_equals('custom_sdk_span', sdk_span.data.sdk.name)
assert sdk_span.data.sdk.custom
assert sdk_span.data.sdk.custom.tags
def test_span_kind(self):
recorder = opentracing.tracer.recorder
span = opentracing.tracer.start_span("custom_sdk_span")
span.set_tag('span.kind', "consumer")
span.finish()
span = opentracing.tracer.start_span("custom_sdk_span")
span.set_tag('span.kind', "server")
span.finish()
span = opentracing.tracer.start_span("custom_sdk_span")
span.set_tag('span.kind', "producer")
span.finish()
span = opentracing.tracer.start_span("custom_sdk_span")
span.set_tag('span.kind', "client")
span.finish()
span = opentracing.tracer.start_span("custom_sdk_span")
span.set_tag('span.kind', "blah")
span.finish()
spans = recorder.queued_spans()
assert 5, len(spans)
span = spans[0]
assert_equals('entry', span.data.sdk.Type)
span = spans[1]
assert_equals('entry', span.data.sdk.Type)
span = spans[2]
assert_equals('exit', span.data.sdk.Type)
span = spans[3]
assert_equals('exit', span.data.sdk.Type)
span = spans[4]
assert_equals('local', span.data.sdk.Type)