def test_thread_decorator_writer_connection(self):
        connections = set()

        @enginefacade.writer.connection
        def go_one(context):
            connections.add(context.connection)

            self.ident = 2
            go_two(context)

            self.ident = 1

            go_three(context)

        @enginefacade.reader.connection
        def go_two(context):
            assert context.connection not in connections

        @enginefacade.reader
        def go_three(context):
            assert context.connection in connections

        context = oslo_context.RequestContext()
        with self._patch_thread_ident():
            go_one(context)
dataset/ETHPy150Open openstack/oslo.db/oslo_db/tests/sqlalchemy/test_enginefacade.py/ThreadingTest.test_thread_decorator_writer_connection/original