# Copyright The OpenTracing Authors # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import absolute_import from unittest import mock import time import pytest import opentracing from opentracing import Format class APICompatibilityCheckMixin(object): """ A mixin class for validation that a given tracer implementation satisfies the requirements of the OpenTracing API. """ def tracer(self): raise NotImplementedError('Subclass must implement tracer()') def check_baggage_values(self): """If true, the test will validate Baggage items by storing and retrieving them from the trace context. If false, it will only attempt to store and retrieve the Baggage items to check the API compliance, but not actually validate stored values. The latter mode is only useful for no-op tracer. """ return True def check_scope_manager(self): """If true, the test suite will validate the `ScopeManager` propagation to ensure correct parenting. If false, it will only use the API without asserting. The latter mode is only useful for no-op tracer. """ return True def is_parent(self, parent, span): """Utility method that must be defined by Tracer implementers to define how the test suite can check when a `Span` is a parent of another one. It depends by the underlying implementation that is not part of the OpenTracing API. """ return False def test_active_span(self): tracer = self.tracer() span = tracer.start_span('Fry') if self.check_scope_manager(): assert tracer.active_span is None assert tracer.scope_manager.active is None with tracer.scope_manager.activate(span, True): assert tracer.active_span is span assert tracer.scope_manager.active.span is span def test_start_active_span(self): # the first usage returns a `Scope` that wraps a root `Span` tracer = self.tracer() with tracer.start_active_span('Fry') as scope: assert scope.span is not None if self.check_scope_manager(): assert self.is_parent(None, scope.span) def test_start_active_span_parent(self): # ensure the `ScopeManager` provides the right parenting tracer = self.tracer() with tracer.start_active_span('Fry') as parent: with tracer.start_active_span('Farnsworth') as child: if self.check_scope_manager(): assert self.is_parent(parent.span, child.span) def test_start_active_span_ignore_active_span(self): # ensure the `ScopeManager` ignores the active `Scope` # if the flag is set tracer = self.tracer() with tracer.start_active_span('Fry') as parent: with tracer.start_active_span('Farnsworth', ignore_active_span=True) as child: if self.check_scope_manager(): assert not self.is_parent(parent.span, child.span) def test_start_active_span_not_finish_on_close(self): # ensure a `Span` is finished when the `Scope` close tracer = self.tracer() scope = tracer.start_active_span('Fry', finish_on_close=False) with mock.patch.object(scope.span, 'finish') as finish: scope.close() assert finish.call_count == 0 def test_start_active_span_finish_on_close(self): # a `Span` is not finished when the flag is set tracer = self.tracer() scope = tracer.start_active_span('Fry', finish_on_close=True) with mock.patch.object(scope.span, 'finish') as finish: scope.close() if self.check_scope_manager(): assert finish.call_count == 1 def test_start_active_span_default_finish_on_close(self): # a `Span` is finished when no flag is set tracer = self.tracer() scope = tracer.start_active_span('Fry') with mock.patch.object(scope.span, 'finish') as finish: scope.close() if self.check_scope_manager(): assert finish.call_count == 1 def test_start_span(self): tracer = self.tracer() span = tracer.start_span(operation_name='Fry') span.finish() with tracer.start_span(operation_name='Fry', tags={'birthday': 'August 14 1974'}) as span: span.log_event('birthplace', payload={'hospital': 'Brooklyn Pre-Med Hospital', 'city': 'Old New York'}) def test_start_span_propagation(self): # `start_span` must inherit the current active `Scope` span tracer = self.tracer() with tracer.start_active_span('Fry') as parent: with tracer.start_span(operation_name='Farnsworth') as child: if self.check_scope_manager(): assert self.is_parent(parent.span, child) def test_start_span_propagation_ignore_active_span(self): # `start_span` doesn't inherit the current active `Scope` span # if the flag is set tracer = self.tracer() with tracer.start_active_span('Fry') as parent: with tracer.start_span(operation_name='Farnsworth', ignore_active_span=True) as child: if self.check_scope_manager(): assert not self.is_parent(parent.span, child) def test_start_span_with_parent(self): tracer = self.tracer() parent_span = tracer.start_span(operation_name='parent') assert parent_span is not None span = tracer.start_span( operation_name='Leela', child_of=parent_span) span.finish() span = tracer.start_span( operation_name='Leela', references=[opentracing.follows_from(parent_span.context)], tags={'birthplace': 'sewers'}) span.finish() parent_span.finish() def test_start_child_span(self): tracer = self.tracer() parent_span = tracer.start_span(operation_name='parent') assert parent_span is not None child_span = opentracing.start_child_span( parent_span, operation_name='Leela') child_span.finish() parent_span.finish() def test_set_operation_name(self): span = self.tracer().start_span().set_operation_name('Farnsworth') span.finish() def test_span_as_context_manager(self): tracer = self.tracer() finish = {'called': False} def mock_finish(*_): finish['called'] = True with tracer.start_span(operation_name='antiquing') as span: setattr(span, 'finish', mock_finish) assert finish['called'] is True # now try with exception finish['called'] = False try: with tracer.start_span(operation_name='antiquing') as span: setattr(span, 'finish', mock_finish) raise ValueError() except ValueError: assert finish['called'] is True else: raise AssertionError('Expected ValueError') # pragma: no cover def test_span_tag_value_types(self): with self.tracer().start_span(operation_name='ManyTypes') as span: span. \ set_tag('an_int', 9). \ set_tag('a_bool', True). \ set_tag('a_string', 'aoeuidhtns') def test_span_tags_with_chaining(self): span = self.tracer().start_span(operation_name='Farnsworth') span. \ set_tag('birthday', '9 April, 2841'). \ set_tag('loves', 'different lengths of wires') span. \ set_tag('unicode_val', u'non-ascii: \u200b'). \ set_tag(u'unicode_key_\u200b', 'ascii val') span.finish() def test_span_logs(self): span = self.tracer().start_span(operation_name='Fry') # Newer API span.log_kv( {'frozen.year': 1999, 'frozen.place': 'Cryogenics Labs'}) span.log_kv( {'defrosted.year': 2999, 'defrosted.place': 'Cryogenics Labs'}, time.time()) # Older API span.\ log_event('frozen', {'year': 1999, 'place': 'Cryogenics Labs'}). \ log_event('defrosted', {'year': 2999}). \ log_event('became his own grandfather', 1947) span.\ log(event='frozen'). \ log(payload={'year': 1999}). \ log(timestamp=time.time(), event='frozen', payload={'year': 1999}). \ log(timestamp=time.time(), event='unfrozen', payload={'year': 2999}) def test_span_baggage(self): with self.tracer().start_span(operation_name='Fry') as span: assert span.context.baggage == {} span_ref = span.set_baggage_item('Kiff-loves', 'Amy') assert span_ref is span val = span.get_baggage_item('Kiff-loves') if self.check_baggage_values(): assert 'Amy' == val pass def test_context_baggage(self): with self.tracer().start_span(operation_name='Fry') as span: assert span.context.baggage == {} span.set_baggage_item('Kiff-loves', 'Amy') if self.check_baggage_values(): assert span.context.baggage == {'Kiff-loves': 'Amy'} pass def test_text_propagation(self): with self.tracer().start_span(operation_name='Bender') as span: text_carrier = {} self.tracer().inject( span_context=span.context, format=opentracing.Format.TEXT_MAP, carrier=text_carrier) extracted_ctx = self.tracer().extract( format=opentracing.Format.TEXT_MAP, carrier=text_carrier) assert extracted_ctx.baggage == {} def test_binary_propagation(self): with self.tracer().start_span(operation_name='Bender') as span: bin_carrier = bytearray() self.tracer().inject( span_context=span.context, format=opentracing.Format.BINARY, carrier=bin_carrier) extracted_ctx = self.tracer().extract( format=opentracing.Format.BINARY, carrier=bin_carrier) assert extracted_ctx.baggage == {} def test_mandatory_formats(self): formats = [ (Format.TEXT_MAP, {}), (Format.HTTP_HEADERS, {}), (Format.BINARY, bytearray()), ] with self.tracer().start_span(operation_name='Bender') as span: for fmt, carrier in formats: # expecting no exceptions span.tracer.inject(span.context, fmt, carrier) span.tracer.extract(fmt, carrier) def test_unknown_format(self): custom_format = 'kiss my shiny metal ...' with self.tracer().start_span(operation_name='Bender') as span: with pytest.raises(opentracing.UnsupportedFormatException): span.tracer.inject(span.context, custom_format, {}) with pytest.raises(opentracing.UnsupportedFormatException): span.tracer.extract(custom_format, {}) def test_tracer_start_active_span_scope(self): # the Tracer ScopeManager should store the active Scope tracer = self.tracer() scope = tracer.start_active_span('Fry') if self.check_scope_manager(): assert tracer.scope_manager.active == scope scope.close() def test_tracer_start_span_scope(self): # the Tracer ScopeManager should not store the new Span tracer = self.tracer() span = tracer.start_span(operation_name='Fry') if self.check_scope_manager(): assert tracer.scope_manager.active is None span.finish()