(self)
| 1958 | self.loop.run_until_complete(coro) |
| 1959 | |
| 1960 | def test_close(self): |
| 1961 | self.loop.close() |
| 1962 | |
| 1963 | async def test(): |
| 1964 | pass |
| 1965 | |
| 1966 | func = lambda: False |
| 1967 | coro = test() |
| 1968 | self.addCleanup(coro.close) |
| 1969 | |
| 1970 | # operation blocked when the loop is closed |
| 1971 | with self.assertRaises(RuntimeError): |
| 1972 | self.loop.run_forever() |
| 1973 | with self.assertRaises(RuntimeError): |
| 1974 | fut = self.loop.create_future() |
| 1975 | self.loop.run_until_complete(fut) |
| 1976 | with self.assertRaises(RuntimeError): |
| 1977 | self.loop.call_soon(func) |
| 1978 | with self.assertRaises(RuntimeError): |
| 1979 | self.loop.call_soon_threadsafe(func) |
| 1980 | with self.assertRaises(RuntimeError): |
| 1981 | self.loop.call_later(1.0, func) |
| 1982 | with self.assertRaises(RuntimeError): |
| 1983 | self.loop.call_at(self.loop.time() + .0, func) |
| 1984 | with self.assertRaises(RuntimeError): |
| 1985 | self.loop.create_task(coro) |
| 1986 | with self.assertRaises(RuntimeError): |
| 1987 | self.loop.add_signal_handler(signal.SIGTERM, func) |
| 1988 | |
| 1989 | # run_in_executor test is tricky: the method is a coroutine, |
| 1990 | # but run_until_complete cannot be called on closed loop. |
| 1991 | # Thus iterate once explicitly. |
| 1992 | with self.assertRaises(RuntimeError): |
| 1993 | it = self.loop.run_in_executor(None, func).__await__() |
| 1994 | next(it) |
| 1995 | |
| 1996 | |
| 1997 | class SubprocessTestsMixin: |
nothing calls this directly
no test coverage detected