test_core.py 文件源码

def test_traced_rollback(self):
        tracer = DummyTracer()
        sqlalchemy_opentracing.init_tracing(tracer, False, False)
        sqlalchemy_opentracing.register_engine(self.engine)

        creat = CreateTable(self.users_table)
        ins = self.users_table.insert().values(name='John Doe')

        # Don't trace this.
        self.engine.execute(creat)

        parent_span = DummySpan('parent span')
        conn = self.engine.connect()
        try:
            with conn.begin() as tx:
                sqlalchemy_opentracing.set_parent_span(conn, parent_span)
                conn.execute(ins)
                conn.execute(creat)
        except OperationalError:
            pass

        self.assertEqual(2, len(tracer.spans))
        self.assertEqual(True, all(map(lambda x: x.is_finished, tracer.spans)))
        self.assertEqual(True, all(map(lambda x: x.child_of == parent_span, tracer.spans)))
        self.assertEqual(['insert', 'create_table'],
                         map(lambda x: x.operation_name, tracer.spans))
        self.assertEqual(['false', 'true'],
                         map(lambda x: x.tags.get('error', 'false'), tracer.spans))