Repository: hikari-py/hikari
Pull request #1073 (2022-03-12 12:32:10)

Use a separate strategy for process and thread based file readers
### Summary
This allows the thread file reader strategy to keep the file open as long as the context manager is active giving similar behaviour to the url reader

### 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/1073.feature.md b/changes/1073.feature.md
new file mode 100644
index 00000000..b8f7b5ec
--- /dev/null
+++ b/changes/1073.feature.md
@@ -0,0 +1,1 @@
+The threaded file reader now persists the open file pointer while the context manager is active.
diff --git a/changes/1074.breaking.md b/changes/1074.breaking.md
new file mode 100644
index 00000000..c8be934b
--- /dev/null
+++ b/changes/1074.breaking.md
@@ -0,0 +1,1 @@
+Hide the entity factory's component deserialize methods.
diff --git a/hikari/api/entity_factory.py b/hikari/api/entity_factory.py
index 3f2688e2..8d127524 100644
--- a/hikari/api/entity_factory.py
+++ b/hikari/api/entity_factory.py
@@ -1296,68 +1296,6 @@ def deserialize_guild_sticker(self, payload: data_binding.JSONObject) -> sticker
     # MESSAGE MODELS #
     ##################
 
-    @abc.abstractmethod
-    def deserialize_action_row(self, payload: data_binding.JSONObject) -> message_models.ActionRowComponent:
-        """Parse a raw payload from Discord into an action row component object.
-
-        Parameters
-        ----------
-        payload : hikari.internal.data_binding.JSONObject
-            The JSON payload to deserialize.
-
-        Returns
-        -------
-        hikari.messages.ActionRowComponent
-            The deserialized action row component.
-        """
-
-    @abc.abstractmethod
-    def deserialize_button(self, payload: data_binding.JSONObject) -> message_models.ButtonComponent:
-        """Parse a raw payload from Discord into a button component object.
-
-        Parameters
-        ----------
-        payload : hikari.internal.data_binding.JSONObject
-            The JSON payload to deserialize.
-
-        Returns
-        -------
-        hikari.messages.ButtonComponent
-            The deserialized button component.
-        """
-
-    @abc.abstractmethod
-    def deserialize_select_menu(self, payload: data_binding.JSONObject) -> message_models.SelectMenuComponent:
-        """Parse a raw payload from Discord into a select menu component object.
-
-        Parameters
-        ----------
-        payload : hikari.internal.data_binding.JSONObject
-            The JSON payload to deserialize.
-
-        Returns
-        -------
-        hikari.messages.SelectMenuComponent
-            The deserialized button component.
-        """
-
-    @abc.abstractmethod
-    def deserialize_component(self, payload: data_binding.JSONObject) -> message_models.PartialComponent:
-        """Parse a raw payload from Discord into a message component object.
-
-        Parameters
-        ----------
-        payload : hikari.internal.data_binding.JSONObject
-            The JSON payload to deserialize.
-
-        Returns
-        -------
-        hikari.messages.PartialComponent
-            The deserialized message component.
-        hikari.errors.UnrecognisedEntityError
-            If the message component type isn't recognised.
-        """
-
     @abc.abstractmethod
     def deserialize_partial_message(self, payload: data_binding.JSONObject) -> message_models.PartialMessage:
         """Parse a raw payload from Discord into a partial message object.
diff --git a/hikari/files.py b/hikari/files.py
index 10b6dbac..178d624e 100644
--- a/hikari/files.py
+++ b/hikari/files.py
@@ -48,6 +48,7 @@
 import asyncio
 import base64
 import concurrent.futures
+import errno
 import inspect
 import io
 import mimetypes
@@ -771,7 +772,7 @@ class FileReader(AsyncReader, abc.ABC):
     """The path to the resource to read."""
 
 
-def _stat(path: pathlib.Path) -> os.stat_result:
+def _stat(path: pathlib.Path) -> pathlib.Path:
     # While paths will be implicitly resolved, we still need to explicitly
     # call expanduser to deal with a ~ base.
     try:
@@ -779,25 +780,30 @@ def _stat(path: pathlib.Path) -> os.stat_result:
     except RuntimeError:
         pass  # A home directory couldn't be resolved, so we'll just use the path as-is.
 
-    return path.stat()
+    # path.stat() will raise FileNotFoundError if the file doesn't exist
+    # (unlike is_dir) which is what we want here.
+    if stat.S_ISDIR(path.stat().st_mode):
+        raise IsADirectoryError(errno.EISDIR, "Cannot open the path specified as it is a directory", str(path))
+
+    return path
 
 
 @attr.define(weakref_slot=False)
 @typing.final
-class _FileAsyncReaderContextManagerImpl(AsyncReaderContextManager[FileReader]):
-    impl: FileReader = attr.field()
-
-    async def __aenter__(self) -> FileReader:
-        loop = asyncio.get_running_loop()
+class _ThreadedFileReaderContextManagerImpl(AsyncReaderContextManager[FileReader]):
+    executor: typing.Optional[concurrent.futures.ThreadPoolExecutor] = attr.field()
+    file: typing.Optional[typing.BinaryIO] = attr.field(default=None, init=False)
+    filename: str = attr.field()
+    path: pathlib.Path = attr.field()
 
-        # Will raise FileNotFoundError if the file doesn't exist (unlike is_dir),
-        # which is what we want here.
-        file_stats = await loop.run_in_executor(self.impl.executor, _stat, self.impl.path)
+    async def __aenter__(self) -> ThreadedFileReader:
+        if self.file:
+            raise RuntimeError("File is already open")
 
-        if stat.S_ISDIR(file_stats.st_mode):
-            raise IsADirectoryError(self.impl.path)
-
-        return self.impl
+        loop = asyncio.get_running_loop()
+        self.path = await loop.run_in_executor(self.executor, _stat, self.path)
+        self.file = typing.cast(io.BufferedReader, await loop.run_in_executor(self.executor, self.path.open, "rb"))
+        return ThreadedFileReader(self.filename, None, self.executor, self.path, self.file)
 
     async def __aexit__(
         self,
@@ -805,18 +811,13 @@ async def __aexit__(
         exc: typing.Optional[BaseException],
         exc_tb: typing.Optional[types.TracebackType],
     ) -> None:
-        pass
-
+        if not self.file:
+            raise RuntimeError("File isn't open")
 
-def _open_file(path: pathlib.Path) -> typing.BinaryIO:
-    # While paths will be implicitly resolved, we still need to explicitly
-    # call expanduser to deal with a ~ base.
-    try:
-        path = path.expanduser()
-    except RuntimeError:
-        pass  # A home directory couldn't be resolved, so we'll just use the path as-is.
-
-    return path.open("rb")
+        loop = asyncio.get_running_loop()
+        file = self.file
+        self.file = None
+        await loop.run_in_executor(self.executor, file.close)
 
 
 @attr.define(weakref_slot=False)
@@ -828,32 +829,44 @@ class ThreadedFileReader(FileReader):
     do not need to be pickled to be communicated.
     """
 
+    _pointer: typing.BinaryIO = attr.field()
+
     async def __aiter__(self) -> typing.AsyncGenerator[typing.Any, bytes]:
         loop = asyncio.get_running_loop()
 
-        fp = await loop.run_in_executor(self.executor, _open_file, self.path)
+        while True:
+            chunk = await loop.run_in_executor(self.executor, self._pointer.read, _MAGIC)
+            yield chunk
+            if len(chunk) < _MAGIC:
+                break
 
-        try:
-            while True:
-                chunk = await loop.run_in_executor(self.executor, fp.read, _MAGIC)
-                yield chunk
-                if len(chunk) < _MAGIC:
-                    break
 
-        finally:
-            await loop.run_in_executor(self.executor, fp.close)
+@attr.define(weakref_slot=False)
+@typing.final
+class _MultiProcessingFileReaderContextManagerImpl(AsyncReaderContextManager[FileReader]):
+    executor: concurrent.futures.ProcessPoolExecutor = attr.field()
+    file: typing.Optional[typing.BinaryIO] = attr.field(default=None, init=False)
+    filename: str = attr.field()
+    path: pathlib.Path = attr.field()
+
+    async def __aenter__(self) -> MultiprocessingFileReader:
+        loop = asyncio.get_running_loop()
 
+        path = await loop.run_in_executor(self.executor, _stat, self.path)
+        return MultiprocessingFileReader(self.filename, None, self.executor, path)
+
+    async def __aexit__(
+        self,
+        exc_type: typing.Optional[typing.Type[BaseException]],
+        exc: typing.Optional[BaseException],
+        exc_tb: typing.Optional[types.TracebackType],
+    ) -> None:
+        pass
 
-def _read_all(path: pathlib.Path) -> bytes:
-    # While paths will be implicitly resolved, we still need to explicitly
-    # call expanduser to deal with a ~ base.
-    try:
-        path = path.expanduser()
-    except RuntimeError:
-        pass  # A home directory couldn't be resolved, so we'll just use the path as-is.
 
-    with path.open("rb") as fp:
-        return fp.read()
+def _read_all(path: pathlib.Path) -> bytes:
+    with path.open("rb") as file:
+        return file.read()
 
 
 @attr.define(slots=False, weakref_slot=False)  # Do not slot (pickle)
@@ -952,12 +965,23 @@ def stream(
         AsyncReaderContextManager[FileReader]
             An async context manager that when entered, produces the
             data stream.
+
+        Raises
+        ------
+        IsADirectoryError
+            If the file's path leads to a directory.
+        FileNotFoundError
+            If the file doesn't exist.
         """
         # asyncio forces the default executor when this is None to always be a thread pool executor anyway,
         # so this is safe enough to do.:
-        is_threaded = executor is None or isinstance(executor, concurrent.futures.ThreadPoolExecutor)
-        impl = ThreadedFileReader if is_threaded else MultiprocessingFileReader
-        return _FileAsyncReaderContextManagerImpl(impl(self.filename, None, executor, self.path))
+        if executor is None or isinstance(executor, concurrent.futures.ThreadPoolExecutor):
+            return _ThreadedFileReaderContextManagerImpl(executor, self.filename, self.path)
+
+        if not isinstance(executor, concurrent.futures.ProcessPoolExecutor):
+            raise TypeError("The executor must be a ProcessPoolExecutor, ThreadPoolExecutor, or `builtins.None`.")
+
+        return _MultiProcessingFileReaderContextManagerImpl(executor, self.filename, self.path)
 
 
 ########################################################################
diff --git a/hikari/impl/entity_factory.py b/hikari/impl/entity_factory.py
index 5f1dfb30..45ad579e 100644
--- a/hikari/impl/entity_factory.py
+++ b/hikari/impl/entity_factory.py
@@ -261,9 +261,9 @@ def __init__(self, app: traits.RESTAware) -> None:
             commands.CommandType.MESSAGE: self.deserialize_context_menu_command,
         }
         self._component_type_mapping = {
-            message_models.ComponentType.ACTION_ROW: self.deserialize_action_row,
-            message_models.ComponentType.BUTTON: self.deserialize_button,
-            message_models.ComponentType.SELECT_MENU: self.deserialize_select_menu,
+            message_models.ComponentType.ACTION_ROW: self._deserialize_action_row,
+            message_models.ComponentType.BUTTON: self._deserialize_button,
+            message_models.ComponentType.SELECT_MENU: self._deserialize_select_menu,
         }
         self._dm_channel_type_mapping = {
             channel_models.ChannelType.DM: self.deserialize_dm,
@@ -2181,12 +2181,12 @@ def deserialize_guild_sticker(self, payload: data_binding.JSONObject) -> sticker
     # MESSAGE MODELS #
     ##################
 
-    def deserialize_action_row(self, payload: data_binding.JSONObject) -> message_models.ActionRowComponent:
+    def _deserialize_action_row(self, payload: data_binding.JSONObject) -> message_models.ActionRowComponent:
         components: typing.List[message_models.PartialComponent] = []
 
         for component_payload in payload["components"]:
             try:
-                components.append(self.deserialize_component(component_payload))
+                components.append(self._deserialize_component(component_payload))
 
             except errors.UnrecognisedEntityError:
                 pass
@@ -2195,7 +2195,7 @@ def deserialize_action_row(self, payload: data_binding.JSONObject) -> message_mo
             type=message_models.ComponentType(payload["type"]), components=components
         )
 
-    def deserialize_button(self, payload: data_binding.JSONObject) -> message_models.ButtonComponent:
+    def _deserialize_button(self, payload: data_binding.JSONObject) -> message_models.ButtonComponent:
         emoji_payload = payload.get("emoji")
         return message_models.ButtonComponent(
             type=message_models.ComponentType(payload["type"]),
@@ -2207,7 +2207,7 @@ def deserialize_button(self, payload: data_binding.JSONObject) -> message_models
             is_disabled=payload.get("disabled", False),
         )
 
-    def deserialize_select_menu(self, payload: data_binding.JSONObject) -> message_models.SelectMenuComponent:
+    def _deserialize_select_menu(self, payload: data_binding.JSONObject) -> message_models.SelectMenuComponent:
         options: typing.List[message_models.SelectMenuOption] = []
         for option_payload in payload["options"]:
             emoji = None
@@ -2234,7 +2234,7 @@ def deserialize_select_menu(self, payload: data_binding.JSONObject) -> message_m
             is_disabled=payload.get("disabled", False),
         )
 
-    def deserialize_component(self, payload: data_binding.JSONObject) -> message_models.PartialComponent:
+    def _deserialize_component(self, payload: data_binding.JSONObject) -> message_models.PartialComponent:
         component_type = message_models.ComponentType(payload["type"])
 
         if deserialize := self._component_type_mapping.get(component_type):
@@ -2390,7 +2390,7 @@ def deserialize_partial_message(  # noqa CFQ001 - Function too long
             components = []
             for component_payload in component_payloads:
                 try:
-                    components.append(self.deserialize_component(component_payload))
+                    components.append(self._deserialize_component(component_payload))
 
                 except errors.UnrecognisedEntityError:
                     pass
@@ -2509,7 +2509,7 @@ def deserialize_message(
         if component_payloads := payload.get("components"):
             for component_payload in component_payloads:
                 try:
-                    components.append(self.deserialize_component(component_payload))
+                    components.append(self._deserialize_component(component_payload))
 
                 except errors.UnrecognisedEntityError:
                     pass


Test patch:
diff --git a/tests/hikari/impl/test_entity_factory.py b/tests/hikari/impl/test_entity_factory.py
index 36158647..404df0fe 100644
--- a/tests/hikari/impl/test_entity_factory.py
+++ b/tests/hikari/impl/test_entity_factory.py
@@ -3939,14 +3939,14 @@ def test_max_age_when_zero(self, entity_factory_impl, invite_with_metadata_paylo
     def action_row_payload(self, button_payload):
         return {"type": 1, "components": [button_payload]}
 
-    def test_deserialize_action_row(self, entity_factory_impl, action_row_payload, button_payload):
-        action_row = entity_factory_impl.deserialize_action_row(action_row_payload)
+    def test__deserialize_action_row(self, entity_factory_impl, action_row_payload, button_payload):
+        action_row = entity_factory_impl._deserialize_action_row(action_row_payload)
 
         assert action_row.type is message_models.ComponentType.ACTION_ROW
-        assert action_row.components == [entity_factory_impl.deserialize_component(button_payload)]
+        assert action_row.components == [entity_factory_impl._deserialize_component(button_payload)]
 
-    def test_deserialize_action_row_handles_unknown_component_type(self, entity_factory_impl):
-        action_row = entity_factory_impl.deserialize_action_row(
+    def test__deserialize_action_row_handles_unknown_component_type(self, entity_factory_impl):
+        action_row = entity_factory_impl._deserialize_action_row(
             {"type": 1, "components": [{"type": "9494949"}, {"type": "9239292"}]}
         )
 
@@ -3964,8 +3964,8 @@ def button_payload(self, custom_emoji_payload):
             "disabled": True,
         }
 
-    def test_deserialize_deserialize_button(self, entity_factory_impl, button_payload, custom_emoji_payload):
-        button = entity_factory_impl.deserialize_button(button_payload)
+    def test_deserialize__deserialize_button(self, entity_factory_impl, button_payload, custom_emoji_payload):
+        button = entity_factory_impl._deserialize_button(button_payload)
 
         assert button.type is message_models.ComponentType.BUTTON
         assert button.style is message_models.ButtonStyle.PRIMARY
@@ -3975,10 +3975,10 @@ def test_deserialize_deserialize_button(self, entity_factory_impl, button_payloa
         assert button.is_disabled is True
         assert button.url == "okokok"
 
-    def test_deserialize_deserialize_button_with_unset_fields(
+    def test_deserialize__deserialize_button_with_unset_fields(
         self, entity_factory_impl, button_payload, custom_emoji_payload
     ):
-        button = entity_factory_impl.deserialize_button({"type": 2, "style": 5})
+        button = entity_factory_impl._deserialize_button({"type": 2, "style": 5})
 
         assert button.type is message_models.ComponentType.BUTTON
         assert button.style is message_models.ButtonStyle.LINK
@@ -4008,8 +4008,8 @@ def select_menu_payload(self, custom_emoji_payload):
             "disabled": True,
         }
 
-    def test_deserialize_select_menu(self, entity_factory_impl, select_menu_payload, custom_emoji_payload):
-        menu = entity_factory_impl.deserialize_select_menu(select_menu_payload)
+    def test__deserialize_select_menu(self, entity_factory_impl, select_menu_payload, custom_emoji_payload):
+        menu = entity_factory_impl._deserialize_select_menu(select_menu_payload)
 
         assert menu.type is message_models.ComponentType.SELECT_MENU
         assert menu.custom_id == "Not an ID"
@@ -4029,8 +4029,8 @@ def test_deserialize_select_menu(self, entity_factory_impl, select_menu_payload,
         assert menu.max_values == 420
         assert menu.is_disabled is True
 
-    def test_deserialize_select_menu_partial(self, entity_factory_impl):
-        menu = entity_factory_impl.deserialize_select_menu(
+    def test__deserialize_select_menu_partial(self, entity_factory_impl):
+        menu = entity_factory_impl._deserialize_select_menu(
             {
                 "type": 3,
                 "custom_id": "Not an ID",
@@ -4050,17 +4050,17 @@ def test_deserialize_select_menu_partial(self, entity_factory_impl):
         assert menu.max_values == 1
         assert menu.is_disabled is False
 
-    def test_deserialize_component(self, entity_factory_impl, action_row_payload, button_payload, select_menu_payload):
+    def test__deserialize_component(self, entity_factory_impl, action_row_payload, button_payload, select_menu_payload):
         for expected_type, payload in [
             (message_models.ActionRowComponent, action_row_payload),
             (message_models.ButtonComponent, button_payload),
             (message_models.SelectMenuComponent, select_menu_payload),
         ]:
-            assert type(entity_factory_impl.deserialize_component(payload)) is expected_type
+            assert type(entity_factory_impl._deserialize_component(payload)) is expected_type
 
-    def test_deserialize_component_handles_unknown_type(self, entity_factory_impl):
+    def test__deserialize_component_handles_unknown_type(self, entity_factory_impl):
         with pytest.raises(errors.UnrecognisedEntityError):
-            entity_factory_impl.deserialize_component({"type": -9434994})
+            entity_factory_impl._deserialize_component({"type": -9434994})
 
     @pytest.fixture()
     def partial_application_payload(self):
@@ -4300,7 +4300,7 @@ def test_deserialize_partial_message(
         assert partial_message.interaction.user == entity_factory_impl.deserialize_user(user_payload)
         assert isinstance(partial_message.interaction, message_models.MessageInteraction)
 
-        assert partial_message.components == [entity_factory_impl.deserialize_component(action_row_payload)]
+        assert partial_message.components == [entity_factory_impl._deserialize_component(action_row_payload)]
 
     def test_deserialize_partial_message_with_partial_fields(self, entity_factory_impl, message_payload):
         message_payload["content"] = ""
@@ -4482,7 +4482,7 @@ def test_deserialize_message(
         assert message.interaction.user == entity_factory_impl.deserialize_user(user_payload)
         assert isinstance(message.interaction, message_models.MessageInteraction)
 
-        assert message.components == [entity_factory_impl.deserialize_component(action_row_payload)]
+        assert message.components == [entity_factory_impl._deserialize_component(action_row_payload)]
 
     def test_deserialize_message_with_unset_sub_fields(self, entity_factory_impl, message_payload):
         del message_payload["application"]["cover_image"]
diff --git a/tests/hikari/test_files.py b/tests/hikari/test_files.py
index 59ee0c04..380db057 100644
--- a/tests/hikari/test_files.py
+++ b/tests/hikari/test_files.py
@@ -22,11 +22,9 @@
 
 import base64
 import concurrent.futures
-import contextlib
 import pathlib
 import random
 import tempfile
-import typing
 
 import mock
 import pytest
@@ -52,68 +50,154 @@ def test___exit__(self, reader):
             pytest.fail(exc)
 
 
-class Test_FileAsyncReaderContextManagerImpl:
-    @pytest.mark.parametrize(
-        "executor", [concurrent.futures.ThreadPoolExecutor, concurrent.futures.ProcessPoolExecutor]
-    )
+class Test_ThreadedFileReaderContextManagerImpl:
     @pytest.mark.asyncio()
-    async def test_context_manager(self, executor: typing.Callable[[], concurrent.futures.Executor]):
-        mock_reader = mock.Mock(executor=executor())
-        context_manager = files._FileAsyncReaderContextManagerImpl(mock_reader)
+    async def test_enter_dunder_method_when_already_open(self):
+        manager = files._ThreadedFileReaderContextManagerImpl(mock.Mock(), "ea", pathlib.Path("ea"))
+        manager.file = mock.Mock()
+        with pytest.raises(RuntimeError, match="File is already open"):
+            await manager.__aenter__()
 
-        with tempfile.NamedTemporaryFile() as file:
-            mock_reader.path = pathlib.Path(file.name)
+    @pytest.mark.asyncio()
+    async def test_exit_dunder_method_when_not_open(self):
+        manager = files._ThreadedFileReaderContextManagerImpl(mock.Mock(), "ea", pathlib.Path("ea"))
+
+        with pytest.raises(RuntimeError, match="File isn't open"):
+            await manager.__aexit__(None, None, None)
+
+    @pytest.mark.asyncio()
+    async def test_context_manager(self):
+        executor = concurrent.futures.ThreadPoolExecutor()
+        mock_data = b"meeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee" * 50
+
+        # A try, finally is used to delete the file rather than relying on delete=True behaviour
+        # as on Windows the file cannot be accessed by other processes if delete is True.
+        file = tempfile.NamedTemporaryFile("wb", delete=False)
+        path = pathlib.Path(file.name)
+        try:
+            with file:
+                file.write(mock_data)
+
+            context_manager = files._ThreadedFileReaderContextManagerImpl(executor, "meow.txt", path)
 
             async with context_manager as reader:
-                assert reader is mock_reader
+                data = await reader.read()
+
+                assert reader.filename == "meow.txt"
+                assert reader.path == path
+                assert reader.executor is executor
+                assert data == mock_data
+
+        finally:
+            path.unlink()
 
+    @mock.patch.object(pathlib.Path, "expanduser", side_effect=RuntimeError)
     @pytest.mark.asyncio()
-    async def test_context_manager_when_expandname_raises_runtime_error(self):
+    async def test_context_manager_when_expandname_raises_runtime_error(self, expanduser: mock.Mock):
         # We can't mock patch stuff in other processes easily (if at all) so
-        # for this test we only run it threaded.
-        mock_reader = mock.Mock(executor=concurrent.futures.ThreadPoolExecutor())
-        context_manager = files._FileAsyncReaderContextManagerImpl(mock_reader)
+        # for this test we have to cheat and use a thread pool executor.
+        executor = concurrent.futures.ThreadPoolExecutor()
 
-        stack = contextlib.ExitStack()
-        file = stack.enter_context(tempfile.NamedTemporaryFile())
-        expandname = stack.enter_context(mock.patch.object(pathlib.Path, "expanduser", side_effect=RuntimeError))
+        # A try, finally is used to delete the file rather than relying on delete=True behaviour
+        # as on Windows the file cannot be accessed by other processes if delete is True.
+        with tempfile.NamedTemporaryFile(delete=False) as file:
+            pass
 
-        with file:
-            mock_reader.path = pathlib.Path(file.name)
+        path = pathlib.Path(file.name)
+        try:
+            context_manager = files._ThreadedFileReaderContextManagerImpl(executor, "filename.txt", path)
 
             async with context_manager as reader:
-                assert reader is mock_reader
+                assert reader.path == path
+
+            expanduser.assert_called_once_with()
 
-        expandname.assert_called_once_with()
+        finally:
+            path.unlink()
 
-    @pytest.mark.parametrize(
-        "executor", [concurrent.futures.ThreadPoolExecutor, concurrent.futures.ProcessPoolExecutor]
-    )
     @pytest.mark.asyncio()
-    async def test_context_manager_for_unknown_file(self, executor: typing.Callable[[], concurrent.futures.Executor]):
-        mock_reader = mock.Mock(executor=executor())
-        context_manager = files._FileAsyncReaderContextManagerImpl(mock_reader)
+    async def test_context_manager_for_unknown_file(self):
+        executor = concurrent.futures.ThreadPoolExecutor()
+        path = pathlib.Path(base64.urlsafe_b64encode(random.getrandbits(512).to_bytes(64, "little")).decode())
+        context_manager = files._ThreadedFileReaderContextManagerImpl(executor, "ea.txt", path)
 
-        mock_reader.path = pathlib.Path(
-            base64.urlsafe_b64encode(random.getrandbits(512).to_bytes(64, "little")).decode()
-        )
+        with pytest.raises(FileNotFoundError):  # noqa:  PT012 - raises block should contain a single statement
+            async with context_manager:
+                ...
+
+    @pytest.mark.asyncio()
+    async def test_test_context_manager_when_target_is_dir(self):
+        executor = concurrent.futures.ThreadPoolExecutor()
+
+        with tempfile.TemporaryDirectory() as name:
+            path = pathlib.Path(name)
+            context_manager = files._ThreadedFileReaderContextManagerImpl(executor, "meow.txt", path)
+
+            with pytest.raises(IsADirectoryError):  # noqa:  PT012 - raises block should contain a single statement
+                async with context_manager:
+                    ...
+
+
+class Test_MultiProcessingFileReaderContextManagerImpl:
+    @pytest.mark.asyncio()
+    async def test_context_manager(self):
+        executor = concurrent.futures.ProcessPoolExecutor()
+        mock_data = b"kon'nichiwa i am yellow and blue da be meow da bayeet" * 50
+
+        # A try, finally is used to delete the file rather than relying on delete=True behaviour
+        # as on Windows the file cannot be accessed by other processes if delete is True.
+        file = tempfile.NamedTemporaryFile("wb", delete=False)
+        path = pathlib.Path(file.name)
+        try:
+            with file:
+                file.write(mock_data)
+
+            context_manager = files._MultiProcessingFileReaderContextManagerImpl(executor, "filename.txt", path)
+
+            async with context_manager as reader:
+                data = await reader.read()
+
+                assert reader.filename == "filename.txt"
+                assert reader.path == path
+                assert reader.executor is executor
+                assert data == mock_data
+
+        finally:
+            path.unlink()
+
+    @mock.patch.object(pathlib.Path, "expanduser", side_effect=RuntimeError)
+    @pytest.mark.asyncio()
+    async def test_context_manager_when_expandname_raises_runtime_error(self, expanduser: mock.Mock):
+        # We can't mock patch stuff in other processes easily (if at all) so
+        # for this test we have to cheat and use a thread pool executor.
+        executor = concurrent.futures.ThreadPoolExecutor()
+
+        with tempfile.NamedTemporaryFile() as file:
+            path = pathlib.Path(file.name)
+            context_manager = files._MultiProcessingFileReaderContextManagerImpl(executor, "filename.txt", path)
+
+            async with context_manager as reader:
+                assert reader.path == path
+
+        expanduser.assert_called_once_with()
+
+    @pytest.mark.asyncio()
+    async def test_context_manager_for_unknown_file(self):
+        executor = concurrent.futures.ProcessPoolExecutor()
+        path = pathlib.Path(base64.urlsafe_b64encode(random.getrandbits(512).to_bytes(64, "little")).decode())
+        context_manager = files._MultiProcessingFileReaderContextManagerImpl(executor, "ea.txt", path)
 
         with pytest.raises(FileNotFoundError):  # noqa:  PT012 - raises block should contain a single statement
             async with context_manager:
                 ...
 
-    @pytest.mark.parametrize(
-        "executor", [concurrent.futures.ThreadPoolExecutor, concurrent.futures.ProcessPoolExecutor]
-    )
     @pytest.mark.asyncio()
-    async def test_test_context_manager_when_target_is_dir(
-        self, executor: typing.Callable[[], concurrent.futures.Executor]
-    ):
-        mock_reader = mock.Mock(executor=executor())
-        context_manager = files._FileAsyncReaderContextManagerImpl(mock_reader)
+    async def test_test_context_manager_when_target_is_dir(self):
+        executor = concurrent.futures.ProcessPoolExecutor()
 
         with tempfile.TemporaryDirectory() as name:
-            mock_reader.path = pathlib.Path(name)
+            path = pathlib.Path(name)
+            context_manager = files._MultiProcessingFileReaderContextManagerImpl(executor, "meow.txt", path)
 
             with pytest.raises(IsADirectoryError):  # noqa:  PT012 - raises block should contain a single statement
                 async with context_manager: