Repository: abhinavsingh/proxy.py
Pull request #1049 (2022-01-23 18:54:42)

[Work] Refactor as FD work
The specific implementation of work framework has been carved out into a `fd` module which contains code dealing with file descriptors as work objects.

codecov[bot]: # [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) Report
> Merging [#1049](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (9350d07) into [develop](https://codecov.io/gh/abhinavsingh/proxy.py/commit/ac840aec78d6435ae85388bbccfc27ec51f0ce49?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (ac840ae) will **decrease** coverage by `0.39%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/graphs/tree.svg?width=650&height=150&src=pr&token=Zh9J7b4la2&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)

```diff
@@             Coverage Diff             @@
##           develop    #1049      +/-   ##
===========================================
- Coverage    87.15%   86.75%   -0.40%     
===========================================
  Files          157      159       +2     
  Lines         6608     6628      +20     
  Branches       669      620      -49     
===========================================
- Hits          5759     5750       -9     
- Misses         738      759      +21     
- Partials       111      119       +8     
```

| Flag | Coverage Δ | |
|---|---|---|
| pytest | `86.63% <100.00%> (-0.34%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) | Coverage Δ | |
|---|---|---|
| [proxy/core/work/work.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL3dvcmsucHk=) | `90.00% <ø> (-0.63%)` | :arrow_down: |
| [proxy/core/acceptor/acceptor.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9hY2NlcHRvci9hY2NlcHRvci5weQ==) | `97.32% <100.00%> (+0.02%)` | :arrow_up: |
| [proxy/core/work/\_\_init\_\_.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL19faW5pdF9fLnB5) | `100.00% <100.00%> (ø)` | |
| [proxy/core/work/fd/\_\_init\_\_.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL2ZkL19faW5pdF9fLnB5) | `100.00% <100.00%> (ø)` | |
| [proxy/core/work/fd/fd.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL2ZkL2ZkLnB5) | `87.50% <100.00%> (ø)` | |
| [proxy/core/work/fd/local.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL2ZkL2xvY2FsLnB5) | `100.00% <100.00%> (ø)` | |
| [proxy/core/work/fd/remote.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL2ZkL3JlbW90ZS5weQ==) | `96.15% <100.00%> (ø)` | |
| [proxy/core/work/local.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL2xvY2FsLnB5) | `62.50% <100.00%> (-37.50%)` | :arrow_down: |
| [proxy/core/work/pool.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL3Bvb2wucHk=) | `91.66% <100.00%> (+0.14%)` | :arrow_up: |
| [proxy/core/work/threadless.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS93b3JrL3RocmVhZGxlc3MucHk=) | `81.00% <100.00%> (-2.00%)` | :arrow_down: |
| ... and [10 more](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Last update [ac840ae...9350d07](https://codecov.io/gh/abhinavsingh/proxy.py/pull/1049?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh).


Patch:
diff --git a/docs/conf.py b/docs/conf.py
index a370246..cfa55f2 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -322,5 +322,5 @@ nitpick_ignore = [
     (_py_obj_role, 'proxy.core.work.threadless.T'),
     (_py_obj_role, 'proxy.core.work.work.T'),
     (_py_obj_role, 'proxy.core.base.tcp_server.T'),
-    (_py_obj_role, 'proxy.core.work.fd.T'),
+    (_py_obj_role, 'proxy.core.work.fd.fd.T'),
 ]
diff --git a/proxy/core/acceptor/acceptor.py b/proxy/core/acceptor/acceptor.py
index 640595f..0df4c2a 100644
--- a/proxy/core/acceptor/acceptor.py
+++ b/proxy/core/acceptor/acceptor.py
@@ -23,8 +23,9 @@ from typing import Dict, List, Tuple, Optional
 from multiprocessing import connection
 from multiprocessing.reduction import recv_handle
 
-from ..work import LocalExecutor, start_threaded_work, delegate_work_to_pool
+from ..work import start_threaded_work, delegate_work_to_pool
 from ..event import EventQueue
+from ..work.fd import LocalFdExecutor
 from ...common.flag import flags
 from ...common.types import HostPort
 from ...common.logger import Logger
@@ -99,7 +100,7 @@ class Acceptor(multiprocessing.Process):
         # Internals
         self._total: Optional[int] = None
         self._local_work_queue: Optional['NonBlockingQueue'] = None
-        self._local: Optional[LocalExecutor] = None
+        self._local: Optional[LocalFdExecutor] = None
         self._lthread: Optional[threading.Thread] = None
 
     def accept(
@@ -195,7 +196,7 @@ class Acceptor(multiprocessing.Process):
     def _start_local(self) -> None:
         assert self.socks
         self._local_work_queue = NonBlockingQueue()
-        self._local = LocalExecutor(
+        self._local = LocalFdExecutor(
             iid=self.idd,
             work_queue=self._local_work_queue,
             flags=self.flags,
diff --git a/proxy/core/work/__init__.py b/proxy/core/work/__init__.py
index a89add7..d09f7d3 100644
--- a/proxy/core/work/__init__.py
+++ b/proxy/core/work/__init__.py
@@ -14,8 +14,7 @@
 """
 from .pool import ThreadlessPool
 from .work import Work
-from .local import LocalExecutor
-from .remote import RemoteExecutor
+from .local import BaseLocalExecutor
 from .delegate import delegate_work_to_pool
 from .threaded import start_threaded_work
 from .threadless import Threadless
@@ -24,9 +23,8 @@ from .threadless import Threadless
 __all__ = [
     'Work',
     'Threadless',
-    'RemoteExecutor',
-    'LocalExecutor',
     'ThreadlessPool',
     'delegate_work_to_pool',
     'start_threaded_work',
+    'BaseLocalExecutor',
 ]
diff --git a/proxy/core/work/fd/__init__.py b/proxy/core/work/fd/__init__.py
new file mode 100644
index 0000000..f277bd5
--- /dev/null
+++ b/proxy/core/work/fd/__init__.py
@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+from .fd import ThreadlessFdExecutor
+from .local import LocalFdExecutor
+from .remote import RemoteFdExecutor
+
+
+__all__ = [
+    'ThreadlessFdExecutor',
+    'LocalFdExecutor',
+    'RemoteFdExecutor',
+]
diff --git a/proxy/core/work/fd.py b/proxy/core/work/fd/fd.py
similarity index 86%
rename from proxy/core/work/fd.py
rename to proxy/core/work/fd/fd.py
index 797d478..71a52af 100644
--- a/proxy/core/work/fd.py
+++ b/proxy/core/work/fd/fd.py
@@ -12,9 +12,9 @@ import socket
 import logging
 from typing import Any, TypeVar, Optional
 
-from ..event import eventNames
-from .threadless import Threadless
-from ...common.types import HostPort, TcpOrTlsSocket
+from ...event import eventNames
+from ..threadless import Threadless
+from ....common.types import HostPort, TcpOrTlsSocket
 
 
 T = TypeVar('T')
@@ -23,12 +23,10 @@ logger = logging.getLogger(__name__)
 
 
 class ThreadlessFdExecutor(Threadless[T]):
+    """A threadless executor which handles file descriptors
+    and works with read/write events over a socket."""
 
-    def work(
-            self,
-            *args: Any,
-            **kwargs: Any,
-    ) -> None:
+    def work(self, **kwargs: Any) -> None:
         fileno: int = kwargs['fileno']
         addr: Optional[HostPort] = kwargs.get('addr', None)
         conn: Optional[TcpOrTlsSocket] = \
diff --git a/proxy/core/work/fd/local.py b/proxy/core/work/fd/local.py
new file mode 100644
index 0000000..0c62d30
--- /dev/null
+++ b/proxy/core/work/fd/local.py
@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+"""
+    proxy.py
+    ~~~~~~~~
+    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
+    Network monitoring, controls & Application development, testing, debugging.
+
+    :copyright: (c) 2013-present by Abhinav Singh and contributors.
+    :license: BSD, see LICENSE for more details.
+"""
+import queue
+import asyncio
+import contextlib
+from typing import Any, Optional
+
+from .fd import ThreadlessFdExecutor
+from ....common.backports import NonBlockingQueue
+
+
+class LocalFdExecutor(ThreadlessFdExecutor[NonBlockingQueue]):
+    """A threadless executor implementation which uses a queue to receive new work."""
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
+
+    @property
+    def loop(self) -> Optional[asyncio.AbstractEventLoop]:
+        if self._loop is None:
+            self._loop = asyncio.get_event_loop_policy().new_event_loop()
+        return self._loop
+
+    def work_queue_fileno(self) -> Optional[int]:
+        return None
+
+    def receive_from_work_queue(self) -> bool:
+        with contextlib.suppress(queue.Empty):
+            work = self.work_queue.get()
+            if isinstance(work, bool) and work is False:
+                return True
+            self.initialize(work)
+        return False
+
+    def initialize(self, work: Any) -> None:
+        assert isinstance(work, tuple)
+        conn, addr = work
+        # NOTE: Here we are assuming to receive a connection object
+        # and not a fileno because we are a LocalExecutor.
+        fileno = conn.fileno()
+        self.work(fileno=fileno, addr=addr, conn=conn)
diff --git a/proxy/core/work/remote.py b/proxy/core/work/fd/remote.py
similarity index 93%
rename from proxy/core/work/remote.py
rename to proxy/core/work/fd/remote.py
index 6168bb3..f7ad87a 100644
--- a/proxy/core/work/remote.py
+++ b/proxy/core/work/fd/remote.py
@@ -9,7 +9,6 @@
     :license: BSD, see LICENSE for more details.
 """
 import asyncio
-import logging
 from typing import Any, Optional
 from multiprocessing import connection
 from multiprocessing.reduction import recv_handle
@@ -17,10 +16,7 @@ from multiprocessing.reduction import recv_handle
 from .fd import ThreadlessFdExecutor
 
 
-logger = logging.getLogger(__name__)
-
-
-class RemoteExecutor(ThreadlessFdExecutor[connection.Connection]):
+class RemoteFdExecutor(ThreadlessFdExecutor[connection.Connection]):
     """A threadless executor implementation which receives work over a connection.
 
     NOTE: RemoteExecutor uses ``recv_handle`` to accept file descriptors.
@@ -40,12 +36,6 @@ class RemoteExecutor(ThreadlessFdExecutor[connection.Connection]):
             self._loop = asyncio.get_event_loop_policy().get_event_loop()
         return self._loop
 
-    def work_queue_fileno(self) -> Optional[int]:
-        return self.work_queue.fileno()
-
-    def close_work_queue(self) -> None:
-        self.work_queue.close()
-
     def receive_from_work_queue(self) -> bool:
         # Acceptor will not send address for
         # unix socket domain environments.
@@ -55,3 +45,9 @@ class RemoteExecutor(ThreadlessFdExecutor[connection.Connection]):
         fileno = recv_handle(self.work_queue)
         self.work(fileno=fileno, addr=addr)
         return False
+
+    def work_queue_fileno(self) -> Optional[int]:
+        return self.work_queue.fileno()
+
+    def close_work_queue(self) -> None:
+        self.work_queue.close()
diff --git a/proxy/core/work/local.py b/proxy/core/work/local.py
index cdb26bd..9a72fab 100644
--- a/proxy/core/work/local.py
+++ b/proxy/core/work/local.py
@@ -8,22 +8,14 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
-import queue
 import asyncio
-import logging
-import contextlib
 from typing import Any, Optional
 
-from .fd import ThreadlessFdExecutor
-from ...common.backports import (  # noqa: W0611, F401   pylint: disable=unused-import
-    NonBlockingQueue,
-)
+from .threadless import Threadless
+from ...common.backports import NonBlockingQueue
 
 
-logger = logging.getLogger(__name__)
-
-
-class LocalExecutor(ThreadlessFdExecutor['NonBlockingQueue']):
+class BaseLocalExecutor(Threadless[NonBlockingQueue]):
     """A threadless executor implementation which uses a queue to receive new work."""
 
     def __init__(self, *args: Any, **kwargs: Any) -> None:
@@ -38,16 +30,3 @@ class LocalExecutor(ThreadlessFdExecutor['NonBlockingQueue']):
 
     def work_queue_fileno(self) -> Optional[int]:
         return None
-
-    def receive_from_work_queue(self) -> bool:
-        with contextlib.suppress(queue.Empty):
-            work = self.work_queue.get()
-            if isinstance(work, bool) and work is False:
-                return True
-            assert isinstance(work, tuple)
-            conn, addr = work
-            # NOTE: Here we are assuming to receive a connection object
-            # and not a fileno because we are a LocalExecutor.
-            fileno = conn.fileno()
-            self.work(fileno=fileno, addr=addr, conn=conn)
-        return False
diff --git a/proxy/core/work/pool.py b/proxy/core/work/pool.py
index 96c43a7..5458f0a 100644
--- a/proxy/core/work/pool.py
+++ b/proxy/core/work/pool.py
@@ -11,16 +11,18 @@
 import logging
 import argparse
 import multiprocessing
-from typing import TYPE_CHECKING, Any, List, Optional
+from typing import TYPE_CHECKING, Any, List, Type, TypeVar, Optional
 from multiprocessing import connection
 
-from .remote import RemoteExecutor
 from ...common.flag import flags
 from ...common.constants import DEFAULT_THREADLESS, DEFAULT_NUM_WORKERS
 
 
 if TYPE_CHECKING:   # pragma: no cover
     from ..event import EventQueue
+    from .threadless import Threadless
+
+T = TypeVar('T', bound='Threadless[Any]')
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +72,7 @@ class ThreadlessPool:
     def __init__(
         self,
         flags: argparse.Namespace,
+        executor_klass: Type['T'],
         event_queue: Optional['EventQueue'] = None,
     ) -> None:
         self.flags = flags
@@ -79,7 +82,9 @@ class ThreadlessPool:
         self.work_pids: List[int] = []
         self.work_locks: List['multiprocessing.synchronize.Lock'] = []
         # List of threadless workers
-        self._workers: List[RemoteExecutor] = []
+        self._executor_klass = executor_klass
+        # FIXME: Instead of Any type must be the executor klass
+        self._workers: List[Any] = []
         self._processes: List[multiprocessing.Process] = []
 
     def __enter__(self) -> 'ThreadlessPool':
@@ -115,8 +120,8 @@ class ThreadlessPool:
         self.work_locks.append(multiprocessing.Lock())
         pipe = multiprocessing.Pipe()
         self.work_queues.append(pipe[0])
-        w = RemoteExecutor(
-            iid=index,
+        w = self._executor_klass(
+            iid=str(index),
             work_queue=pipe[1],
             flags=self.flags,
             event_queue=self.event_queue,
diff --git a/proxy/core/work/threadless.py b/proxy/core/work/threadless.py
index ac76c7e..216a5bd 100644
--- a/proxy/core/work/threadless.py
+++ b/proxy/core/work/threadless.py
@@ -120,7 +120,7 @@ class Threadless(ABC, Generic[T]):
         raise NotImplementedError()
 
     @abstractmethod
-    def work(self, *args: Any, **kwargs: Any) -> None:
+    def work(self, **kwargs: Any) -> None:
         raise NotImplementedError()
 
     def close_work_queue(self) -> None:
diff --git a/proxy/core/work/work.py b/proxy/core/work/work.py
index 0856010..0cb4b67 100644
--- a/proxy/core/work/work.py
+++ b/proxy/core/work/work.py
@@ -55,16 +55,14 @@ class Work(ABC, Generic[T]):
         creation of externally defined work class objects."""
         raise NotImplementedError()
 
-    @abstractmethod
     async def get_events(self) -> SelectableEvents:
         """Return sockets and events (read or write) that we are interested in."""
         return {}   # pragma: no cover
 
-    @abstractmethod
     async def handle_events(
             self,
-            readables: Readables,
-            writables: Writables,
+            _readables: Readables,
+            _writables: Writables,
     ) -> bool:
         """Handle readable and writable sockets.
 
diff --git a/proxy/proxy.py b/proxy/proxy.py
index fcf8f79..522ce41 100644
--- a/proxy/proxy.py
+++ b/proxy/proxy.py
@@ -21,6 +21,7 @@ from .core.work import ThreadlessPool
 from .core.event import EventManager
 from .common.flag import FlagParser, flags
 from .common.utils import bytes_
+from .core.work.fd import RemoteFdExecutor
 from .core.acceptor import AcceptorPool
 from .core.listener import ListenerPool
 from .common.constants import (
@@ -213,6 +214,7 @@ class Proxy:
             self.executors = ThreadlessPool(
                 flags=self.flags,
                 event_queue=event_queue,
+                executor_klass=RemoteFdExecutor,
             )
             self.executors.setup()
         # Setup acceptors


Test patch:
diff --git a/tests/test_main.py b/tests/test_main.py
index ce341c8..1d4c115 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -16,6 +16,7 @@ from unittest import mock
 
 from proxy.proxy import main, entry_point
 from proxy.common.utils import bytes_
+from proxy.core.work.fd import RemoteFdExecutor
 from proxy.common.constants import (  # noqa: WPS450
     DEFAULT_PORT, DEFAULT_PLUGINS, DEFAULT_TIMEOUT, DEFAULT_KEY_FILE,
     DEFAULT_LOG_FILE, DEFAULT_PAC_FILE, DEFAULT_PID_FILE, PLUGIN_DASHBOARD,
@@ -119,6 +120,7 @@ class TestMain(unittest.TestCase):
         mock_executor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
             event_queue=None,
+            executor_klass=RemoteFdExecutor,
         )
         mock_acceptor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
@@ -169,6 +171,7 @@ class TestMain(unittest.TestCase):
         mock_executor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
             event_queue=None,
+            executor_klass=RemoteFdExecutor,
         )
         mock_acceptor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
@@ -214,6 +217,7 @@ class TestMain(unittest.TestCase):
         mock_executor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,
             event_queue=mock_event_manager.return_value.queue,
+            executor_klass=RemoteFdExecutor,
         )
         mock_acceptor_pool.assert_called_once_with(
             flags=mock_initialize.return_value,