Recognize distro i2pd daemon process

This commit is contained in:
jwinterm
2026-07-08 09:14:13 -04:00
parent c38eab9c28
commit e272ec1afb
2 changed files with 35 additions and 2 deletions
+9 -1
View File
@@ -17,6 +17,9 @@ from .transports import (
VALID_DAEMON_POLICIES = {"external", "adopt", "manage", "auto"} VALID_DAEMON_POLICIES = {"external", "adopt", "manage", "auto"}
SUPPORTED_RENDERERS = {"clearnet", "privacy-html", "gemtext", "micron"} SUPPORTED_RENDERERS = {"clearnet", "privacy-html", "gemtext", "micron"}
PROCESS_NAME_ALIASES = {
"i2pd": ("i2pd", "i2pd-daemon"),
}
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -277,9 +280,14 @@ def _process_running(executable: str) -> bool:
pgrep = shutil.which("pgrep") pgrep = shutil.which("pgrep")
if not pgrep: if not pgrep:
return False return False
names = PROCESS_NAME_ALIASES.get(executable, (executable,))
return any(_pgrep_exact(pgrep, name) for name in names)
def _pgrep_exact(pgrep: str, name: str) -> bool:
try: try:
result = subprocess.run( result = subprocess.run(
[pgrep, "-x", executable], [pgrep, "-x", name],
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
timeout=1, timeout=1,
+26 -1
View File
@@ -3,11 +3,12 @@ import io
import tempfile import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch
from ampg.cli import main from ampg.cli import main
from ampg.config import load_config from ampg.config import load_config
from ampg.platforms import platform_by_name from ampg.platforms import platform_by_name
from ampg.status import DaemonProbe, doctor_gateway, gateway_status from ampg.status import DaemonProbe, _process_running, doctor_gateway, gateway_status
def _probe(*, installed: bool = False, running: bool = False): def _probe(*, installed: bool = False, running: bool = False):
@@ -197,6 +198,30 @@ daemon_policy = "external"
self.assertIn("warnings=1", output) self.assertIn("warnings=1", output)
self.assertIn("errors=0", output) self.assertIn("errors=0", output)
def test_i2pd_probe_accepts_distro_daemon_process_name(self):
calls: list[tuple[str, ...]] = []
def run(command, **_kwargs):
calls.append(tuple(command))
class Result:
returncode = 0 if command[-1] == "i2pd-daemon" else 1
return Result()
with patch("ampg.status.shutil.which", return_value="/usr/bin/pgrep"):
with patch("ampg.status.subprocess.run", side_effect=run):
running = _process_running("i2pd")
self.assertTrue(running)
self.assertEqual(
[
("/usr/bin/pgrep", "-x", "i2pd"),
("/usr/bin/pgrep", "-x", "i2pd-daemon"),
],
calls,
)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()