Repository: hikari-py/hikari
Pull request #864 (2021-10-21 15:00:36)

Make EventStream a sync context manager + simplify filter behaviour
### Summary
<!-- Small summary of the merge request -->

### Checklist
<!-- Make sure to tick all the following boxes by putting an `x` in between (like this `[x]`) -->
- [ ] I have run `nox` and all the pipelines have passed.
- [ ] I have made unittests according to the code I have added/modified/deleted.

### Related issues
<!--
To mention an issue use `#issue-id` and to mention a merge request use `!merge-request-id`
To close/fix an issue use `Close #issue-id` or `Fix #issue-id` (depending on the merge request)
-->



Patch:
diff --git a/changes/864.deprecation.md b/changes/864.deprecation.md
new file mode 100644
index 00000000..5e274d53
--- /dev/null
+++ b/changes/864.deprecation.md
@@ -0,0 +1,1 @@
+EventStream is now a sync context menu not async.
diff --git a/changes/864.feature.md b/changes/864.feature.md
new file mode 100644
index 00000000..6e7fd063
--- /dev/null
+++ b/changes/864.feature.md
@@ -0,0 +1,1 @@
+`EventStream.filter` now always returns `EventStream`.
diff --git a/hikari/api/event_manager.py b/hikari/api/event_manager.py
index 90ec286c..8cd02c76 100644
--- a/hikari/api/event_manager.py
+++ b/hikari/api/event_manager.py
@@ -57,12 +57,12 @@ class EventStream(iterators.LazyIterator[EventT], abc.ABC):
 
     Examples
     --------
-    A streamer may either be started and closed using `async with` syntax
+    A streamer may either be started and closed using `with` syntax
     where `EventStream.open` and `EventStream.close` are implicitly called based on
     context.
 
     ```py
-    async with EventStream(app, EventType, timeout=50) as stream:
+    with EventStream(app, EventType, timeout=50) as stream:
         async for entry in stream:
             ...
     ```
@@ -89,33 +89,33 @@ class EventStream(iterators.LazyIterator[EventT], abc.ABC):
     __slots__: typing.Sequence[str] = ()
 
     @abc.abstractmethod
-    async def close(self) -> None:
+    def close(self) -> None:
         """Mark this streamer as closed to stop it from queueing and receiving events.
 
         If called on an already closed streamer then this will do nothing.
 
         !!! note
-            `async with streamer` may be used as a short-cut for opening and
+            `with streamer` may be used as a short-cut for opening and
             closing a streamer.
         """
 
     @abc.abstractmethod
-    async def open(self) -> None:
+    def open(self) -> None:
         """Mark this streamer as opened to let it start receiving and queueing events.
 
         If called on an already started streamer then this will do nothing.
 
         !!! note
-            `async with streamer` may be used as a short-cut for opening and
+            `with streamer` may be used as a short-cut for opening and
             closing a stream.
         """
 
     @abc.abstractmethod
-    async def __aenter__(self) -> EventStream[EventT]:
+    def __enter__(self) -> EventStream[EventT]:
         raise NotImplementedError
 
     @abc.abstractmethod
-    async def __aexit__(
+    def __exit__(
         self,
         exc_type: typing.Optional[typing.Type[BaseException]],
         exc: typing.Optional[BaseException],
@@ -416,33 +416,33 @@ def stream(
         -------
         EventStream[hikari.events.base_events.Event]
             The async iterator to handle streamed events. This must be started
-            with `async with stream:` or `await stream.open()` before
+            with `with stream:` or `stream.open()` before
             asynchronously iterating over it.
 
         !!! warning
-            If you use `await stream.open()` to start the stream then you must
-            also close it with `await stream.close()` otherwise it may queue
+            If you use `stream.open()` to start the stream then you must
+            also close it with `stream.close()` otherwise it may queue
             events in memory indefinitely.
 
         Examples
         --------
 
         ```py
-        async with bot.stream(events.ReactionAddEvent, timeout=30).filter(("message_id", message.id)) as stream:
+        with bot.stream(events.ReactionAddEvent, timeout=30).filter(("message_id", message.id)) as stream:
             async for user_id in stream.map("user_id").limit(50):
                 ...
         ```
 
-        or using await `open()` and await `close()`
+        or using `open()` and `close()`
 
         ```py
         stream = bot.stream(events.ReactionAddEvent, timeout=30).filter(("message_id", message.id))
-        await stream.open()
+        stream.open()
 
         async for user_id in stream.map("user_id").limit(50)
             ...
 
-        await stream.close()
+        stream.close()
         ```
 
         See Also
diff --git a/hikari/impl/event_manager_base.py b/hikari/impl/event_manager_base.py
index 0c66ed64..3e3290a3 100644
--- a/hikari/impl/event_manager_base.py
+++ b/hikari/impl/event_manager_base.py
@@ -64,6 +64,7 @@
     WaiterMapT = typing.MutableMapping[
         typing.Type[event_manager_.EventT_co], typing.MutableSet[WaiterT[event_manager_.EventT_co]]
     ]
+    _EventStreamT = typing.TypeVar("_EventStreamT", bound="EventStream[typing.Any]")
 
 
 def _generate_weak_listener(
@@ -94,9 +95,11 @@ class EventStream(event_manager_.EventStream[event_manager_.EventT]):
     __slots__: typing.Sequence[str] = (
         "__weakref__",
         "_active",
+        "_event",
         "_event_manager",
         "_event_type",
         "_filters",
+        "_limit",
         "_queue",
         "_registered_listener",
         "_timeout",
@@ -110,55 +113,70 @@ def __init__(
         timeout: typing.Union[float, int, None],
         limit: typing.Optional[int] = None,
     ) -> None:
-        self._event_manager = event_manager
         self._active = False
+        self._event: typing.Optional[asyncio.Event] = None
+        self._event_manager = event_manager
         self._event_type = event_type
         self._filters: iterators.All[event_manager_.EventT] = iterators.All(())
-        # We accept `None` to represent unlimited here to be consistent with how `None` is already used to represent
-        # unlimited for timeout in other places.
-        self._queue: asyncio.Queue[event_manager_.EventT] = asyncio.Queue(limit or 0)
+        self._limit = limit
+        self._queue: typing.List[event_manager_.EventT] = []
         self._registered_listener: typing.Optional[
             typing.Callable[[event_manager_.EventT], typing.Coroutine[typing.Any, typing.Any, None]]
         ] = None
         # The registered wrapping function for the weak ref to this class's _listener method.
         self._timeout = timeout
 
-    async def __aenter__(self) -> EventStream[event_manager_.EventT]:
-        await self.open()
+    # These are only included at runtime in-order to avoid the model being typed as an asynchronous context manager.
+    if not typing.TYPE_CHECKING:
+
+        async def __aenter__(self: _EventStreamT) -> _EventStreamT:
+            # This is sync only.
+            warnings.warn(
+                "Using EventStream as an async context manager has been deprecated since 2.0.0.dev104. "
+                "Please use it as a sycnrhonous context manager (e.g. with bot.stream(...)) instead.",
+                category=DeprecationWarning,
+                stacklevel=2,
+            )
+
+            self.open()
+            return self
+
+        async def __aexit__(
+            self,
+            exc_type: typing.Optional[typing.Type[BaseException]],
+            exc: typing.Optional[BaseException],
+            exc_tb: typing.Optional[types.TracebackType],
+        ) -> None:
+            self.close()
+
+    def __enter__(self: _EventStreamT) -> _EventStreamT:
+        self.open()
         return self
 
-    async def __aexit__(
+    def __exit__(
         self,
         exc_type: typing.Optional[typing.Type[BaseException]],
-        exc: typing.Optional[BaseException],
+        exc_val: typing.Optional[BaseException],
         exc_tb: typing.Optional[types.TracebackType],
     ) -> None:
-        await self.close()
+        self.close()
 
-    # These are only included at runtime in-order to avoid the model being typed as a synchronous context manager.
-    if not typing.TYPE_CHECKING:
+    async def __anext__(self) -> event_manager_.EventT:
+        if not self._active:
+            raise TypeError("stream must be started with before entering it")
 
-        def __enter__(self) -> typing.NoReturn:
-            # This is async only.
-            cls = type(self)
-            raise TypeError(f"{cls.__module__}.{cls.__qualname__} is async-only, did you mean 'async with'?") from None
+        while not self._queue:
+            if not self._event:
+                self._event = asyncio.Event()
 
-        def __exit__(
-            self,
-            exc_type: typing.Optional[typing.Type[Exception]],
-            exc_val: typing.Optional[Exception],
-            exc_tb: typing.Optional[types.TracebackType],
-        ) -> None:
-            return None
+            try:
+                await asyncio.wait_for(self._event.wait(), timeout=self._timeout)
+            except asyncio.TimeoutError:
+                raise StopAsyncIteration from None
 
-    async def __anext__(self) -> event_manager_.EventT:
-        if not self._active:
-            raise TypeError("stream must be started with `async with` before entering it")
+            self._event.clear()
 
-        try:
-            return await asyncio.wait_for(self._queue.get(), timeout=self._timeout)
-        except asyncio.TimeoutError:
-            raise StopAsyncIteration from None
+        return self._queue.pop(0)
 
     def __await__(self) -> typing.Generator[None, None, typing.Sequence[event_manager_.EventT]]:
         return self._await_all().__await__()
@@ -170,27 +188,23 @@ def __del__(self) -> None:
         # ominous memory leak.
         if self._active:
             _LOGGER.warning("active %r streamer fell out of scope before being closed", self._event_type.__name__)
-            try:
-                asyncio.ensure_future(self.close())
-            except RuntimeError:
-                pass
+            self.close()
 
     async def _await_all(self) -> typing.Sequence[event_manager_.EventT]:
-        await self.open()
+        self.open()
         result = [event async for event in self]
-        await self.close()
+        self.close()
         return result
 
     async def _listener(self, event: event_manager_.EventT) -> None:
-        if not self._filters(event):
+        if not self._filters(event) or (self._limit is not None and len(self._queue) >= self._limit):
             return
 
-        try:
-            self._queue.put_nowait(event)
-        except asyncio.QueueFull:
-            pass
+        self._queue.append(event)
+        if self._event:
+            self._event.set()
 
-    async def close(self) -> None:
+    def close(self) -> None:
         if self._active and self._registered_listener is not None:
             try:
                 self._event_manager.unsubscribe(self._event_type, self._registered_listener)
@@ -202,17 +216,18 @@ async def close(self) -> None:
         self._active = False
 
     def filter(
-        self,
+        self: _EventStreamT,
         *predicates: typing.Union[typing.Tuple[str, typing.Any], typing.Callable[[event_manager_.EventT], bool]],
         **attrs: typing.Any,
-    ) -> typing.Union[EventStream[event_manager_.EventT], iterators.LazyIterator[event_manager_.EventT]]:
+    ) -> _EventStreamT:
+        filter_ = self._map_predicates_and_attr_getters("filter", *predicates, **attrs)
         if self._active:
-            return super().filter(*predicates, **attrs)
+            self._queue = [entry for entry in self._queue if filter_(entry)]
 
-        self._filters |= self._map_predicates_and_attr_getters("filter", *predicates, **attrs)
+        self._filters |= filter_
         return self
 
-    async def open(self) -> None:
+    def open(self) -> None:
         if not self._active:
             # For the sake of protecting highly intelligent people who forget to close this, we register the event
             # listener with a weakref then try to close this on deletion. While this may lead to their consoles being
diff --git a/hikari/impl/rest.py b/hikari/impl/rest.py
index 4385659e..20806553 100644
--- a/hikari/impl/rest.py
+++ b/hikari/impl/rest.py
@@ -670,8 +670,8 @@ def __enter__(self) -> typing.NoReturn:
 
         def __exit__(
             self,
-            exc_type: typing.Optional[typing.Type[Exception]],
-            exc_val: typing.Optional[Exception],
+            exc_type: typing.Optional[typing.Type[BaseException]],
+            exc_val: typing.Optional[BaseException],
             exc_tb: typing.Optional[types.TracebackType],
         ) -> None:
             return None


Test patch:
diff --git a/tests/hikari/impl/test_event_manager_base.py b/tests/hikari/impl/test_event_manager_base.py
index 8ff38beb..43e6ce2f 100644
--- a/tests/hikari/impl/test_event_manager_base.py
+++ b/tests/hikari/impl/test_event_manager_base.py
@@ -74,37 +74,45 @@ def mock_app():
 
 
 class TestEventStream:
-    @pytest.mark.asyncio()
-    async def test___aenter___and___aexit__(
+    def test___enter___and___exit__(
         self,
     ):
         stub_stream = hikari_test_helpers.mock_class_namespace(
-            event_manager_base.EventStream, open=mock.AsyncMock(), close=mock.AsyncMock()
+            event_manager_base.EventStream, open=mock.Mock(), close=mock.Mock()
         )(mock_app, base_events.Event, timeout=None)
 
-        async with stub_stream:
-            stub_stream.open.assert_awaited_once()
+        with stub_stream:
+            stub_stream.open.assert_called_once_with()
             stub_stream.close.assert_not_called()
 
-        stub_stream.open.assert_awaited_once()
-        stub_stream.close.assert_awaited_once()
+        stub_stream.open.assert_called_once_with()
+        stub_stream.close.assert_called_once_with()
 
-    def test___enter__(self):
+    @pytest.mark.asyncio()
+    async def test___aenter__(self):
         # flake8 gets annoyed if we use "with" here so here's a hacky alternative
-        stub_stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
-            mock_app, base_events.Event, timeout=None
-        )
+        stub_stream = hikari_test_helpers.mock_class_namespace(
+            event_manager_base.EventStream, open=mock.Mock(), close=mock.Mock()
+        )(mock_app, base_events.Event, timeout=None)
+
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", category=DeprecationWarning)
 
-        with pytest.raises(TypeError, match=" is async-only, did you mean 'async with'?"):
-            stub_stream.__enter__()
+            async with stub_stream:
+                stub_stream.open.assert_called_once_with()
+                stub_stream.close.assert_not_called()
 
-    def test___exit__(self):
+        stub_stream.open.assert_called_once_with()
+        stub_stream.close.assert_called_once_with()
+
+    @pytest.mark.asyncio()
+    async def test___exit__(self):
         stub_stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             mock_app, base_events.Event, timeout=None
         )
 
         try:
-            stub_stream.__exit__(None, None, None)
+            await stub_stream.__aexit__(None, None, None)
         except AttributeError as exc:
             pytest.fail(exc)
 
@@ -115,34 +123,37 @@ async def test__listener_when_filter_returns_false(self, mock_app):
         mock_event = object()
 
         assert await stream._listener(mock_event) is None
-        assert stream._queue.qsize() == 0
+        assert not stream._queue
 
+    @hikari_test_helpers.timeout()
     @pytest.mark.asyncio()
     async def test__listener_when_filter_passes_and_queue_full(self, mock_app):
         stream = event_manager_base.EventStream(mock_app, base_events.Event, timeout=None, limit=2)
-        stream._queue.put_nowait(object())
-        stream._queue.put_nowait(object())
+        stream._queue.append(object())
+        stream._queue.append(object())
         stream.filter(lambda _: True)
         mock_event = object()
 
-        assert await stream._listener(mock_event) is None
-        assert stream._queue.qsize() == 2
-        assert stream._queue.get_nowait() is not mock_event
-        assert stream._queue.get_nowait() is not mock_event
+        with stream:
+            assert await stream._listener(mock_event) is None
+            assert await stream.next() is not mock_event
+            assert await stream.next() is not mock_event
+            assert not stream._queue
 
+    @hikari_test_helpers.timeout()
     @pytest.mark.asyncio()
     async def test__listener_when_filter_passes_and_queue_not_full(self, mock_app):
         stream = event_manager_base.EventStream(mock_app, base_events.Event, timeout=None, limit=None)
-        stream._queue.put_nowait(object())
-        stream._queue.put_nowait(object())
+        stream._queue.append(object())
+        stream._queue.append(object())
         stream.filter(lambda _: True)
         mock_event = object()
 
-        assert await stream._listener(mock_event) is None
-        assert stream._queue.qsize() == 3
-        assert stream._queue.get_nowait() is not mock_event
-        assert stream._queue.get_nowait() is not mock_event
-        assert stream._queue.get_nowait() is mock_event
+        with stream:
+            assert await stream._listener(mock_event) is None
+            assert await stream.next() is not mock_event
+            assert await stream.next() is not mock_event
+            assert await stream.next() is mock_event
 
     @pytest.mark.asyncio()
     @hikari_test_helpers.timeout()
@@ -160,9 +171,9 @@ async def test___anext___times_out(self):
             mock.Mock(), event_type=base_events.Event, timeout=hikari_test_helpers.REASONABLE_QUICK_RESPONSE_TIME
         )
 
-        async with streamer:
-            async for _ in streamer:
-                pytest.fail("streamer shouldn't have yielded anything")
+        with streamer:
+            with pytest.raises(LookupError):
+                await streamer.next()
 
     @pytest.mark.asyncio()
     @hikari_test_helpers.timeout()
@@ -174,16 +185,12 @@ async def test___anext___waits_for_next_event(self):
 
         async def add_event():
             await asyncio.sleep(hikari_test_helpers.REASONABLE_SLEEP_TIME)
-            streamer._queue.put_nowait(mock_event)
+            await streamer._listener(mock_event)
 
         asyncio.create_task(add_event())
 
-        async with streamer:
-            async for event in streamer:
-                assert event is mock_event
-                return
-
-            pytest.fail("streamer should've yielded something")
+        with streamer:
+            assert await streamer.next() is mock_event
 
     @pytest.mark.asyncio()
     @hikari_test_helpers.timeout()
@@ -194,14 +201,10 @@ async def test___anext__(self):
             event_type=base_events.Event,
             timeout=hikari_test_helpers.REASONABLE_QUICK_RESPONSE_TIME,
         )
-        streamer._queue.put_nowait(mock_event)
+        streamer._queue.append(mock_event)
 
-        async with streamer:
-            async for event in streamer:
-                assert event is mock_event
-                return
-
-        pytest.fail("streamer should've yielded something")
+        with streamer:
+            assert await streamer.next() is mock_event
 
     @pytest.mark.asyncio()
     async def test___await__(self):
@@ -210,16 +213,16 @@ async def test___await__(self):
         mock_event_2 = object()
         streamer = hikari_test_helpers.mock_class_namespace(
             event_manager_base.EventStream,
-            close=mock.AsyncMock(),
-            open=mock.AsyncMock(),
+            close=mock.Mock(),
+            open=mock.Mock(),
             init_=False,
             _active=False,
             __anext__=mock.AsyncMock(side_effect=[mock_event_0, mock_event_1, mock_event_2]),
         )()
 
         assert await streamer == [mock_event_0, mock_event_1, mock_event_2]
-        streamer.open.assert_awaited_once()
-        streamer.close.assert_awaited_once()
+        streamer.open.assert_called_once_with()
+        streamer.close.assert_called_once_with()
 
     def test___del___for_active_stream(self):
         mock_coroutine = object()
@@ -230,14 +233,12 @@ def test___del___for_active_stream(self):
         streamer._event_type = base_events.Event
         streamer._active = True
 
-        with mock.patch.object(asyncio, "ensure_future", side_effect=RuntimeError) as ensure_future:
-            with unittest.TestCase().assertLogs("hikari.event_manager", level=logging.WARNING) as logging_watcher:
-                del streamer
+        with unittest.TestCase().assertLogs("hikari.event_manager", level=logging.WARNING) as logging_watcher:
+            del streamer
 
         assert logging_watcher.output == [
             "WARNING:hikari.event_manager:active 'Event' streamer fell out of scope before being closed"
         ]
-        ensure_future.assert_called_once_with(mock_coroutine)
         close_method.assert_called_once_with()
 
     def test___del___for_inactive_stream(self):
@@ -248,37 +249,29 @@ def test___del___for_inactive_stream(self):
         streamer._event_type = base_events.Event
         streamer._active = False
 
-        with mock.patch.object(asyncio, "ensure_future"):
-            del streamer
-            asyncio.ensure_future.assert_not_called()
-
+        del streamer
         close_method.assert_not_called()
 
-    @pytest.mark.asyncio()
-    async def test_close_for_inactive_stream(self, mock_app):
+    def test_close_for_inactive_stream(self, mock_app):
         stream = event_manager_base.EventStream(mock_app, base_events.Event, timeout=None, limit=None)
-        await stream.close()
+        stream.close()
         mock_app.event_manager.unsubscribe.assert_not_called()
 
-    @pytest.mark.asyncio()
-    @hikari_test_helpers.timeout()
-    async def test_close_for_active_stream(self):
+    def test_close_for_active_stream(self):
         mock_registered_listener = object()
         mock_manager = mock.Mock()
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             event_manager=mock_manager, event_type=base_events.Event, timeout=float("inf")
         )
 
-        await stream.open()
+        stream.open()
         stream._registered_listener = mock_registered_listener
-        await stream.close()
+        stream.close()
         mock_manager.unsubscribe.assert_called_once_with(base_events.Event, mock_registered_listener)
         assert stream._active is False
         assert stream._registered_listener is None
 
-    @pytest.mark.asyncio()
-    @hikari_test_helpers.timeout()
-    async def test_close_for_active_stream_handles_value_error(self):
+    def test_close_for_active_stream_handles_value_error(self):
         mock_registered_listener = object()
         mock_manager = mock.Mock()
         mock_manager.unsubscribe.side_effect = ValueError
@@ -286,14 +279,15 @@ async def test_close_for_active_stream_handles_value_error(self):
             event_manager=mock_manager, event_type=base_events.Event, timeout=float("inf")
         )
 
-        await stream.open()
+        stream.open()
         stream._registered_listener = mock_registered_listener
-        await stream.close()
+        stream.close()
         mock_manager.unsubscribe.assert_called_once_with(base_events.Event, mock_registered_listener)
         assert stream._active is False
         assert stream._registered_listener is None
 
-    def test_filter_for_inactive_stream(self):
+    @pytest.mark.asyncio()
+    async def test_filter(self):
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             event_manager=mock.Mock(), event_type=base_events.Event, timeout=1
         )
@@ -308,30 +302,37 @@ def predicate(obj):
 
         stream.filter(predicate, attr=True)
 
-        assert stream._filters(first_pass) is True
-        assert stream._filters(first_fails) is False
-        assert stream._filters(second_pass) is True
-        assert stream._filters(second_fail) is False
+        await stream._listener(first_pass)
+        await stream._listener(first_fails)
+        await stream._listener(second_pass)
+        await stream._listener(second_fail)
+
+        assert await stream == [first_pass, second_pass]
 
     @pytest.mark.asyncio()
-    async def test_filter_for_active_stream(self):
+    async def test_filter_handles_calls_while_active(self):
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
-            event_manager=mock.Mock(), event_type=base_events.Event, timeout=float("inf")
+            event_manager=mock.Mock(), event_type=base_events.Event, timeout=1
         )
-        stream._active = True
-        mock_wrapping_iterator = object()
-        predicate = object()
+        stream._filters = iterators.All(())
+        first_pass = mock.Mock(attr=True)
+        second_pass = mock.Mock(attr=True)
+        first_fails = mock.Mock(attr=True)
+        second_fail = mock.Mock(attr=False)
+        await stream._listener(first_pass)
+        await stream._listener(first_fails)
+        await stream._listener(second_pass)
+        await stream._listener(second_fail)
 
-        with mock.patch.object(iterators.LazyIterator, "filter", return_value=mock_wrapping_iterator):
-            assert stream.filter(predicate, name="OK") is mock_wrapping_iterator
+        def predicate(obj):
+            return obj in (first_pass, second_pass)
 
-            iterators.LazyIterator.filter.assert_called_once_with(predicate, name="OK")
+        with stream:
+            stream.filter(predicate, attr=True)
 
-        # Ensure we don't get a warning or error on del
-        stream._active = False
+            assert await stream == [first_pass, second_pass]
 
-    @pytest.mark.asyncio()
-    async def test_open_for_inactive_stream(self):
+    def test_open_for_inactive_stream(self):
         mock_listener = object()
         mock_manager = mock.Mock()
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
@@ -345,7 +346,7 @@ async def test_open_for_inactive_stream(self):
 
         with mock.patch.object(event_manager_base, "_generate_weak_listener"):
             with mock.patch.object(weakref, "WeakMethod"):
-                await stream.open()
+                stream.open()
 
                 weakref.WeakMethod.assert_not_called()
             event_manager_base._generate_weak_listener.assert_not_called()
@@ -357,9 +358,7 @@ async def test_open_for_inactive_stream(self):
         # Ensure we don't get a warning or error on del
         stream._active = False
 
-    @pytest.mark.asyncio()
-    @hikari_test_helpers.timeout()
-    async def test_open_for_active_stream(self):
+    def test_open_for_active_stream(self):
         mock_manager = mock.Mock()
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             event_manager=mock_manager, event_type=base_events.Event, timeout=float("inf")
@@ -370,7 +369,7 @@ async def test_open_for_active_stream(self):
 
         with mock.patch.object(event_manager_base, "_generate_weak_listener", return_value=mock_listener):
             with mock.patch.object(weakref, "WeakMethod", return_value=mock_listener_ref):
-                await stream.open()
+                stream.open()
 
                 weakref.WeakMethod.assert_called_once_with(stream._listener)
             event_manager_base._generate_weak_listener.assert_called_once_with(mock_listener_ref)