Python contextlib 模块，_GeneratorContextManager() 实例源码

def __call__(self, func):
        if isgeneratorfunction(func):
            def inner(*args, **kwds):
                with self._recreate_cm():
                    yield from func(*args, **kwds)
        elif is_contextmanager(func):
            @contextmanager
            def inner(*args, **kwds):
                with self._recreate_cm():
                    with func(*args, **kwds) as ret:
                        yield ret
        else:
            def inner(*args, **kwds):
                with self._recreate_cm():
                    return func(*args, **kwds)
        return wraps(func)(inner)


# Some python version have a different signature for '_GeneratorContextManager.__init__', so we must adapt:

def test_connection_is_context_manager(self):
        import contextlib
        import six

        queue_return = _Connection()
        pool = self._makeOneWithMockQueue(queue_return)
        cnxn_context = pool.connection()
        if six.PY3:  # pragma: NO COVER Python 3
            self.assertTrue(isinstance(cnxn_context,
                                       contextlib._GeneratorContextManager))
        else:
            self.assertTrue(isinstance(cnxn_context,
                                       contextlib.GeneratorContextManager))

def test_open_and_close(self):
        # ProfileConfig.open() returns a context manager that closes the
        # database on exit.
        config_file = os.path.join(self.make_dir(), "config")
        config = api.ProfileConfig.open(config_file)
        self.assertIsInstance(config, contextlib._GeneratorContextManager)
        with config as config:
            self.assertIsInstance(config, api.ProfileConfig)
            with config.cursor() as cursor:
                self.assertEqual(
                    (1,), cursor.execute("SELECT 1").fetchone())
        self.assertRaises(sqlite3.ProgrammingError, config.cursor)

def test_open_and_close(self):
        # ConfigurationDatabase.open() returns a context manager that closes
        # the database on exit.
        config_file = os.path.join(self.make_dir(), "config")
        config = ConfigurationDatabase.open_for_update(config_file)
        self.assertIsInstance(config, contextlib._GeneratorContextManager)
        with config as config:
            self.assertIsInstance(config, ConfigurationDatabase)
            with config.cursor() as cursor:
                self.assertEqual(
                    (1,), cursor.execute("SELECT 1").fetchone())
        self.assertRaises(sqlite3.ProgrammingError, config.cursor)

def test_open_and_close(self):
        # ConfigurationFile.open() returns a context manager.
        config_file = os.path.join(self.make_dir(), "config")
        config_ctx = ConfigurationFile.open(config_file)
        self.assertIsInstance(config_ctx, contextlib._GeneratorContextManager)
        with config_ctx as config:
            self.assertIsInstance(config, ConfigurationFile)
            self.assertThat(config_file, FileExists())
            self.assertEqual({}, config.config)
            self.assertFalse(config.dirty)
        self.assertThat(config_file, FileContains(""))