test_core.py 文件源码

def test_traced_clear_connection(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')

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

            # Stop tracing from this point.
            sqlalchemy_opentracing.clear_traced(conn)

            conn.execute(ins)

        self.assertEqual(1, len(tracer.spans))
        self.assertEqual('create_table', tracer.spans[0].operation_name)