is_server_side=False,
_asyncio_internal=False):
if not _asyncio_internal:
- warnings.warn(f"{self.__class__} should be instaniated "
- "by asyncio internals only, "
- "please avoid its creation from user code",
- DeprecationWarning)
+ raise RuntimeError(f"{self.__class__} should be instantiated "
+ "by asyncio internals only")
self._mode = mode
self._transport = transport
self._protocol = protocol
self.loop.run_until_complete(test())
+ def test_stream_ctor_forbidden(self):
+ with self.assertRaisesRegex(RuntimeError,
+ "should be instantiated "
+ "by asyncio internals only"):
+ asyncio.Stream(asyncio.StreamMode.READWRITE)
+
if __name__ == '__main__':
unittest.main()