test service shutdown response to signals (#13576)

* minimal fixup for daemon signal handling regression

* test daemon shutdown response to signals

* positional

* test datalayer shutdown response to signals

* one parametrized test for the services

* colocate the daemon test

* drop sig

* add sendable_termination_signals

* oops, it was SIGTERM

* wait for the daemon

chia/util/misc.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import dataclasses
+import signal
+import sys
 from pathlib import Path
 from typing import Any, Dict, Sequence, Union
 
@@ -107,3 +109,11 @@ def validate_directory_writable(path: Path) -> None:
         raise InvalidPathError(path, "Directory doesn't exist")
     except OSError:
         raise InvalidPathError(path, "Directory not writable")
+
+
+if sys.platform == "win32" or sys.platform == "cygwin":
+    termination_signals = [signal.SIGBREAK, signal.SIGINT, signal.SIGTERM]
+    sendable_termination_signals = [signal.SIGTERM]
+else:
+    termination_signals = [signal.SIGINT, signal.SIGTERM]
+    sendable_termination_signals = termination_signals

tests/conftest.py

@@ -2,6 +2,7 @@
 import aiohttp
 import multiprocessing
 import os
+import sysconfig
 from typing import Any, AsyncIterator, Dict, List, Tuple, Union
 
 import pytest
@@ -27,6 +28,7 @@ from chia.types.peer_info import PeerInfo
 from chia.util.config import create_default_chia_config, lock_and_load_config
 from chia.util.ints import uint16
 from chia.wallet.wallet import Wallet
+from tests.core.data_layer.util import ChiaRoot
 from tests.core.node_height import node_height_at_least
 from tests.setup_nodes import (
     setup_simulators_and_wallets,
@@ -688,3 +690,21 @@ def config_with_address_prefix(root_path_populated_with_config: Path, prefix: st
         if prefix is not None:
             config["network_overrides"]["config"][config["selected_network"]]["address_prefix"] = prefix
     return config
+
+
+@pytest.fixture(name="scripts_path", scope="session")
+def scripts_path_fixture() -> Path:
+    scripts_string = sysconfig.get_path("scripts")
+    if scripts_string is None:
+        raise Exception("These tests depend on the scripts path existing")
+
+    return Path(scripts_string)
+
+
+@pytest.fixture(name="chia_root", scope="function")
+def chia_root_fixture(tmp_path: Path, scripts_path: Path) -> ChiaRoot:
+    root = ChiaRoot(path=tmp_path.joinpath("chia_root"), scripts_path=scripts_path)
+    root.run(args=["init"])
+    root.run(args=["configure", "--set-log-level", "INFO"])
+
+    return root

tests/core/data_layer/conftest.py

@@ -1,14 +1,11 @@
 from __future__ import annotations
 
-import contextlib
 import os
 import pathlib
 import random
-import subprocess
 import sys
-import sysconfig
 import time
-from typing import Any, AsyncIterable, Awaitable, Callable, Dict, Iterator, List
+from typing import Any, AsyncIterable, Awaitable, Callable, Dict, Iterator
 
 import pytest
 import pytest_asyncio
@@ -26,45 +23,13 @@ from tests.core.data_layer.util import (
     add_01234567_example,
     create_valid_node_values,
 )
+from tests.util.misc import closing_chia_root_popen
 
 # TODO: These are more general than the data layer and should either move elsewhere or
 #       be replaced with an existing common approach.  For now they can at least be
 #       shared among the data layer test files.
 
 
-@pytest.fixture(name="scripts_path", scope="session")
-def scripts_path_fixture() -> pathlib.Path:
-    scripts_string = sysconfig.get_path("scripts")
-    if scripts_string is None:
-        raise Exception("These tests depend on the scripts path existing")
-
-    return pathlib.Path(scripts_string)
-
-
-@pytest.fixture(name="chia_root", scope="function")
-def chia_root_fixture(tmp_path: pathlib.Path, scripts_path: pathlib.Path) -> ChiaRoot:
-    root = ChiaRoot(path=tmp_path.joinpath("chia_root"), scripts_path=scripts_path)
-    root.run(args=["init"])
-    root.run(args=["configure", "--set-log-level", "INFO"])
-
-    return root
-
-
-@contextlib.contextmanager
-def closing_chia_root_popen(chia_root: ChiaRoot, args: List[str]) -> Iterator[None]:
-    environment = {**os.environ, "CHIA_ROOT": os.fspath(chia_root.path)}
-
-    with subprocess.Popen(args=args, env=environment) as process:
-        try:
-            yield
-        finally:
-            process.terminate()
-            try:
-                process.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                process.kill()
-
-
 @pytest.fixture(name="chia_daemon", scope="function")
 def chia_daemon_fixture(chia_root: ChiaRoot) -> Iterator[None]:
     with closing_chia_root_popen(chia_root=chia_root, args=[sys.executable, "-m", "chia.daemon.server"]):

tests/core/test_services.py

@@ -0,0 +1,155 @@
+from __future__ import annotations
+
+import asyncio
+import signal
+import sys
+import time
+from pathlib import Path
+from typing import Any, Dict
+
+import aiohttp.client_exceptions
+import pytest
+from typing_extensions import Protocol
+
+from chia.daemon.client import DaemonProxy, connect_to_daemon_and_validate
+from chia.rpc.data_layer_rpc_client import DataLayerRpcClient
+from chia.rpc.farmer_rpc_client import FarmerRpcClient
+from chia.rpc.full_node_rpc_client import FullNodeRpcClient
+from chia.rpc.harvester_rpc_client import HarvesterRpcClient
+from chia.rpc.rpc_client import RpcClient
+from chia.rpc.wallet_rpc_client import WalletRpcClient
+from chia.simulator.socket import find_available_listen_port
+from chia.util.config import lock_and_load_config, save_config
+from chia.util.ints import uint16
+from chia.util.misc import sendable_termination_signals
+from tests.core.data_layer.util import ChiaRoot
+from tests.util.misc import closing_chia_root_popen
+
+
+class CreateServiceProtocol(Protocol):
+    async def __call__(
+        self,
+        self_hostname: str,
+        port: uint16,
+        root_path: Path,
+        net_config: Dict[str, Any],
+    ) -> RpcClient:
+        ...
+
+
+async def wait_for_daemon_connection(root_path: Path, config: Dict[str, Any], timeout: int = 15) -> DaemonProxy:
+    start = time.monotonic()
+    while time.monotonic() - start < timeout:
+        client = await connect_to_daemon_and_validate(root_path=root_path, config=config, quiet=True)
+        if client is not None:
+            break
+        await asyncio.sleep(0.1)
+    else:
+        raise Exception(f"unable to connect within {timeout} seconds")
+    return client
+
+
+@pytest.mark.parametrize(argnames="signal_number", argvalues=sendable_termination_signals)
+@pytest.mark.asyncio
+async def test_daemon_terminates(signal_number: signal.Signals, chia_root: ChiaRoot) -> None:
+    port = find_available_listen_port()
+    with lock_and_load_config(root_path=chia_root.path, filename="config.yaml") as config:
+        config["daemon_port"] = port
+        save_config(root_path=chia_root.path, filename="config.yaml", config_data=config)
+
+    with closing_chia_root_popen(chia_root=chia_root, args=[sys.executable, "-m", "chia.daemon.server"]) as process:
+        client = await wait_for_daemon_connection(root_path=chia_root.path, config=config)
+
+        try:
+            return_code = process.poll()
+            assert return_code is None
+
+            process.send_signal(signal_number)
+            process.communicate(timeout=5)
+        finally:
+            await client.close()
+
+
+@pytest.mark.parametrize(argnames="signal_number", argvalues=sendable_termination_signals)
+@pytest.mark.parametrize(
+    argnames=["create_service", "module_path", "service_config_name"],
+    argvalues=[
+        [DataLayerRpcClient.create, "chia.server.start_data_layer", "data_layer"],
+        [FarmerRpcClient.create, "chia.server.start_farmer", "farmer"],
+        [FullNodeRpcClient.create, "chia.server.start_full_node", "full_node"],
+        [HarvesterRpcClient.create, "chia.server.start_harvester", "harvester"],
+        [WalletRpcClient.create, "chia.server.start_wallet", "wallet"],
+        # TODO: review and somehow test the other services too
+        # [, "chia.server.start_introducer", "introducer"],
+        # [, "chia.seeder.start_crawler", ""],
+        # [, "chia.server.start_timelord", "timelord"],
+        # [, "chia.timelord.timelord_launcher", ],
+        # [, "chia.simulator.start_simulator", ],
+        # [, "chia.data_layer.data_layer_server", "data_layer"],
+    ],
+)
+@pytest.mark.asyncio
+async def test_services_terminate(
+    signal_number: signal.Signals,
+    chia_root: ChiaRoot,
+    create_service: CreateServiceProtocol,
+    module_path: str,
+    service_config_name: str,
+) -> None:
+    with lock_and_load_config(root_path=chia_root.path, filename="config.yaml") as config:
+        config["daemon_port"] = find_available_listen_port(name="daemon")
+        service_config = config[service_config_name]
+        if "port" in service_config:
+            port = find_available_listen_port(name="service")
+            service_config["port"] = port
+        rpc_port = find_available_listen_port(name="rpc")
+        service_config["rpc_port"] = rpc_port
+        save_config(root_path=chia_root.path, filename="config.yaml", config_data=config)
+
+    # TODO: make the wallet start up regardless so this isn't needed
+    with closing_chia_root_popen(
+        chia_root=chia_root,
+        args=[sys.executable, "-m", "chia.daemon.server"],
+    ):
+        # Make sure the daemon is running and responsive before starting other services.
+        # This probably shouldn't be required.  For now, it helps at least with the
+        # farmer.
+        daemon_client = await wait_for_daemon_connection(root_path=chia_root.path, config=config)
+        await daemon_client.close()
+
+        with closing_chia_root_popen(
+            chia_root=chia_root,
+            args=[sys.executable, "-m", module_path],
+        ) as process:
+            client = await create_service(
+                self_hostname=config["self_hostname"],
+                port=uint16(rpc_port),
+                root_path=chia_root.path,
+                net_config=config,
+            )
+            try:
+                start = time.monotonic()
+                while time.monotonic() - start < 50:
+                    return_code = process.poll()
+                    assert return_code is None
+
+                    try:
+                        result = await client.healthz()
+                    except aiohttp.client_exceptions.ClientConnectorError:
+                        pass
+                    else:
+                        if result.get("success", False):
+                            break
+
+                    await asyncio.sleep(0.1)
+                else:
+                    raise Exception("unable to connect")
+
+                return_code = process.poll()
+                assert return_code is None
+
+                process.send_signal(signal_number)
+                process.communicate(timeout=30)
+            finally:
+                client.close()
+                await client.await_closed()

tests/util/misc.py

@@ -5,17 +5,21 @@ import dataclasses
 import enum
 import gc
 import math
+import os
+import subprocess
 from concurrent.futures import Future
 from inspect import getframeinfo, stack
 from statistics import mean
 from textwrap import dedent
 from time import thread_time
 from types import TracebackType
-from typing import Callable, Iterator, List, Optional, Type, Union
+from typing import Any, Callable, Iterator, List, Optional, Type, Union
 
 import pytest
 from typing_extensions import final
 
+from tests.core.data_layer.util import ChiaRoot
+
 
 class GcMode(enum.Enum):
     nothing = enum.auto
@@ -284,3 +288,18 @@ def assert_rpc_error(error: str) -> Iterator[None]:
     with pytest.raises(ValueError) as exception_info:
         yield
     assert error in exception_info.value.args[0]["error"]
+
+
+@contextlib.contextmanager
+def closing_chia_root_popen(chia_root: ChiaRoot, args: List[str]) -> Iterator[subprocess.Popen[Any]]:
+    environment = {**os.environ, "CHIA_ROOT": os.fspath(chia_root.path)}
+
+    with subprocess.Popen(args=args, env=environment) as process:
+        try:
+            yield process
+        finally:
+            process.terminate()
+            try:
+                process.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                process.kill()