test_core.py 文件源码

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

        creat = CreateTable(self.users_table)

        conn = self.engine.connect()
        with conn.begin() as tx:
            sqlalchemy_opentracing.set_traced(conn)
            conn.execute(creat)

        self.assertEqual(1, len(tracer.spans))

        # Do something right after with this connection,
        # no tracing should happen.
        tracer.clear()
        ins = self.users_table.insert().values(name='John Doe')
        with conn.begin() as tx:
            conn.execute(ins)

        self.assertEqual(0, len(tracer.spans))