move interop tests tools into tests folder

tests/utils/interop/constants.py

@@ -0,0 +1 @@
+LOCALHOST_IP = "127.0.0.1"

tests/utils/interop/daemon.py

@@ -0,0 +1,149 @@
+from contextlib import (
+    asynccontextmanager,
+)
+from typing import (
+    AsyncIterator,
+)
+
+import multiaddr
+from multiaddr import (
+    Multiaddr,
+)
+from p2pclient import (
+    Client,
+)
+import trio
+
+from libp2p.peer.id import (
+    ID,
+)
+from libp2p.peer.peerinfo import (
+    PeerInfo,
+    info_from_p2p_addr,
+)
+from libp2p.security.noise.transport import PROTOCOL_ID as NOISE_PROTOCOL_ID
+from libp2p.security.secio.transport import ID as SECIO_PROTOCOL_ID
+from libp2p.typing import (
+    TProtocol,
+)
+
+from .constants import (
+    LOCALHOST_IP,
+)
+from .envs import (
+    GO_BIN_PATH,
+)
+from .process import (
+    BaseInteractiveProcess,
+)
+
+P2PD_PATH = GO_BIN_PATH / "p2pd"
+
+
+class P2PDProcess(BaseInteractiveProcess):
+    def __init__(
+        self,
+        control_maddr: Multiaddr,
+        security_protocol: TProtocol,
+        is_pubsub_enabled: bool = True,
+        is_gossipsub: bool = True,
+        is_pubsub_signing: bool = False,
+        is_pubsub_signing_strict: bool = False,
+    ) -> None:
+        args = [f"-listen={control_maddr!s}"]
+        if security_protocol == SECIO_PROTOCOL_ID:
+            args.append("-secio")
+        if security_protocol == NOISE_PROTOCOL_ID:
+            args.append("-noise")
+        if is_pubsub_enabled:
+            args.append("-pubsub")
+            if is_gossipsub:
+                args.append("-pubsubRouter=gossipsub")
+            else:
+                args.append("-pubsubRouter=floodsub")
+            if not is_pubsub_signing:
+                args.append("-pubsubSign=false")
+            if not is_pubsub_signing_strict:
+                args.append("-pubsubSignStrict=false")
+            # NOTE:
+            #   Two other params are possibly what we want to configure:
+            #   - gossipsubHeartbeatInterval: GossipSubHeartbeatInitialDelay = 100 * time.Millisecond  # noqa: E501
+            #   - gossipsubHeartbeatInitialDelay: GossipSubHeartbeatInterval = 1 * time.Second  # noqa: E501
+            #   Referece: https://github.com/libp2p/go-libp2p-daemon/blob/b95e77dbfcd186ccf817f51e95f73f9fd5982600/p2pd/main.go#L348-L353  # noqa: E501
+        self.proc = None
+        self.cmd = str(P2PD_PATH)
+        self.args = args
+        self.patterns = (b"Control socket:", b"Peer ID:", b"Peer Addrs:")
+        self.bytes_read = bytearray()
+        self.event_ready = trio.Event()
+
+
+class Daemon:
+    p2pd_proc: BaseInteractiveProcess
+    control: Client
+    peer_info: PeerInfo
+
+    def __init__(
+        self, p2pd_proc: BaseInteractiveProcess, control: Client, peer_info: PeerInfo
+    ) -> None:
+        self.p2pd_proc = p2pd_proc
+        self.control = control
+        self.peer_info = peer_info
+
+    def __repr__(self) -> str:
+        return f"<Daemon {self.peer_id.to_string()[2:8]}>"
+
+    @property
+    def peer_id(self) -> ID:
+        return self.peer_info.peer_id
+
+    @property
+    def listen_maddr(self) -> Multiaddr:
+        return self.peer_info.addrs[0]
+
+    async def close(self) -> None:
+        await self.p2pd_proc.close()
+        await self.control.close()
+
+
+@asynccontextmanager
+async def make_p2pd(
+    daemon_control_port: int,
+    client_callback_port: int,
+    security_protocol: TProtocol,
+    is_pubsub_enabled: bool = True,
+    is_gossipsub: bool = True,
+    is_pubsub_signing: bool = False,
+    is_pubsub_signing_strict: bool = False,
+) -> AsyncIterator[Daemon]:
+    control_maddr = Multiaddr(f"/ip4/{LOCALHOST_IP}/tcp/{daemon_control_port}")
+    p2pd_proc = P2PDProcess(
+        control_maddr,
+        security_protocol,
+        is_pubsub_enabled,
+        is_gossipsub,
+        is_pubsub_signing,
+        is_pubsub_signing_strict,
+    )
+    await p2pd_proc.start()
+    client_callback_maddr = Multiaddr(f"/ip4/{LOCALHOST_IP}/tcp/{client_callback_port}")
+    p2pc = Client(control_maddr, client_callback_maddr)
+
+    async with p2pc.listen():
+        peer_id, maddrs = await p2pc.identify()
+        listen_maddr: Multiaddr = None
+        for maddr in maddrs:
+            try:
+                ip = maddr.value_for_protocol(multiaddr.protocols.P_IP4)
+                # NOTE: Check if this `maddr` uses `tcp`.
+                maddr.value_for_protocol(multiaddr.protocols.P_TCP)
+            except multiaddr.exceptions.ProtocolLookupError:
+                continue
+            if ip == LOCALHOST_IP:
+                listen_maddr = maddr
+                break
+        assert listen_maddr is not None, "no loopback maddr is found"
+        peer_info = info_from_p2p_addr(
+            listen_maddr.encapsulate(Multiaddr(f"/p2p/{peer_id.to_string()}"))
+        )
+        yield Daemon(p2pd_proc, p2pc, peer_info)

tests/utils/interop/envs.py

@@ -0,0 +1,4 @@
+import os
+import pathlib
+
+GO_BIN_PATH = pathlib.Path(os.environ["GOPATH"]) / "bin"

tests/utils/interop/process.py

@@ -0,0 +1,72 @@
+from abc import (
+    ABC,
+    abstractmethod,
+)
+import subprocess
+from typing import (
+    Iterable,
+    List,
+)
+
+import trio
+
+TIMEOUT_DURATION = 30
+
+
+class AbstractInterativeProcess(ABC):
+    @abstractmethod
+    async def start(self) -> None:
+        ...
+
+    @abstractmethod
+    async def close(self) -> None:
+        ...
+
+
+class BaseInteractiveProcess(AbstractInterativeProcess):
+    proc: trio.Process = None
+    cmd: str
+    args: List[str]
+    bytes_read: bytearray
+    patterns: Iterable[bytes] = None
+    event_ready: trio.Event
+
+    async def wait_until_ready(self) -> None:
+        patterns_occurred = {pat: False for pat in self.patterns}
+
+        async def read_from_daemon_and_check() -> None:
+            async for data in self.proc.stdout:
+                # TODO: It takes O(n^2), which is quite bad.
+                # But it should succeed in a few seconds.
+                self.bytes_read.extend(data)
+                for pat, occurred in patterns_occurred.items():
+                    if occurred:
+                        continue
+                    if pat in self.bytes_read:
+                        patterns_occurred[pat] = True
+                if all([value for value in patterns_occurred.values()]):
+                    return
+
+        with trio.fail_after(TIMEOUT_DURATION):
+            await read_from_daemon_and_check()
+        self.event_ready.set()
+        # Sleep a little bit to ensure the listener is up after logs are emitted.
+        await trio.sleep(0.01)
+
+    async def start(self) -> None:
+        if self.proc is not None:
+            return
+        # mypy says that `open_process` is not an attribute of trio, suggests run_process instead.  # noqa: E501
+        self.proc = await trio.open_process(  # type: ignore[attr-defined]
+            [self.cmd] + self.args,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,  # Redirect stderr to stdout, which makes parsing easier  # noqa: E501
+            bufsize=0,
+        )
+        await self.wait_until_ready()
+
+    async def close(self) -> None:
+        if self.proc is None:
+            return
+        self.proc.terminate()
+        await self.proc.wait()

tests/utils/interop/utils.py

@@ -0,0 +1,70 @@
+from typing import (
+    Union,
+)
+
+from multiaddr import (
+    Multiaddr,
+)
+import trio
+
+from libp2p.host.host_interface import (
+    IHost,
+)
+from libp2p.peer.id import (
+    ID,
+)
+from libp2p.peer.peerinfo import (
+    PeerInfo,
+)
+
+from .daemon import (
+    Daemon,
+)
+
+TDaemonOrHost = Union[IHost, Daemon]
+
+
+def _get_peer_info(node: TDaemonOrHost) -> PeerInfo:
+    peer_info: PeerInfo
+    if isinstance(node, Daemon):
+        peer_info = node.peer_info
+    else:  # isinstance(node, IHost)
+        peer_id = node.get_id()
+        maddrs = [
+            node.get_addrs()[0].decapsulate(Multiaddr(f"/p2p/{peer_id.to_string()}"))
+        ]
+        peer_info = PeerInfo(peer_id, maddrs)
+    return peer_info
+
+
+async def _is_peer(peer_id: ID, node: TDaemonOrHost) -> bool:
+    if isinstance(node, Daemon):
+        pinfos = await node.control.list_peers()
+        peers = tuple([pinfo.peer_id for pinfo in pinfos])
+        return peer_id in peers
+    else:  # isinstance(node, IHost)
+        return peer_id in node.get_network().connections
+
+
+async def connect(a: TDaemonOrHost, b: TDaemonOrHost) -> None:
+    # Type check
+    err_msg = (
+        f"Type of a={type(a)} or type of b={type(b)} is wrong."
+        "Should be either `IHost` or `Daemon`"
+    )
+    assert all(
+        [isinstance(node, IHost) or isinstance(node, Daemon) for node in (a, b)]
+    ), err_msg
+
+    b_peer_info = _get_peer_info(b)
+    if isinstance(a, Daemon):
+        await a.control.connect(b_peer_info.peer_id, b_peer_info.addrs)
+    else:  # isinstance(b, IHost)
+        await a.connect(b_peer_info)
+    # Allow additional sleep for both side to establish the connection.
+    await trio.sleep(0.1)
+
+    a_peer_info = _get_peer_info(a)
+
+    assert await _is_peer(b_peer_info.peer_id, a)
+    assert await _is_peer(a_peer_info.peer_id, b)