diff --git a/pyproject.toml b/pyproject.toml index 1e067337..d992b0f9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ dependencies = [ "mistletoe>=1.5.1", "numpy>=2.2.6", "openai>=2.26.0", + "psutil>=5.9", "pydantic>=2.12.5", "python-frontmatter>=1.1.0", "pyyaml>=6.0.3", diff --git a/reme/__init__.py b/reme/__init__.py index 5b352c2a..d30e7bf5 100644 --- a/reme/__init__.py +++ b/reme/__init__.py @@ -1,6 +1,6 @@ """ReMe CLI package.""" -__version__ = "0.4.0.4" +__version__ = "0.4.0.5" from . import config from . import constants diff --git a/reme/steps/transfer/__init__.py b/reme/steps/transfer/__init__.py index fcaa0b02..d71e8165 100644 --- a/reme/steps/transfer/__init__.py +++ b/reme/steps/transfer/__init__.py @@ -1,11 +1,9 @@ """Transfer steps.""" from .download import DownloadStep -from .ingest import IngestStep from .upload import UploadStep __all__ = [ "DownloadStep", - "IngestStep", "UploadStep", ] diff --git a/reme/steps/transfer/ingest.py b/reme/steps/transfer/ingest.py deleted file mode 100644 index b493b2ab..00000000 --- a/reme/steps/transfer/ingest.py +++ /dev/null @@ -1,452 +0,0 @@ -"""``ingest`` — capture an externally-received asset into ``resource//``. - -This step is the dedicated information-capture interface: when an -external channel (wechat group, email, browser save, API push, ...) -hands the agent a file, ``ingest`` lands it in ``resource//`` -keyed by the day it was received (always today, local time), -alongside a ``meta.json`` row recording provenance. - -This is the only ingest path through ``resource/``. Materials the -main agent *actively* fetches or generates during a daily task -belong inside the daily folder as sibling materials, not here. - -For generic file copy (local fs → arbitrary workspace path) use -``upload`` instead. - -Bucket layout:: - - resource// - meta.json # JSON array of FileNode rows - .md # derived markdown view - ____ # the asset itself - ... - -Naming convention — the file name is **derived**, never caller-supplied:: - - ____ - -* ```` — top-level identity (anchors provenance), lowercase - letters/digits/dashes only. -* ```` — receive time within the bucket day (the date is - already implicit in the bucket folder). -* ```` — the basename of the ``path`` argument, - after rejecting path separators, dot segments, and leading dots. - -This format is self-describing in directory listings + wikilinks -(``[[resource//wechat__153022__report.pdf]]``) and makes -cross-channel basename collisions structurally impossible. Two -genuine duplicates (same channel + same second + same basename) -are reported as an error — the step never silently dedupes, so -callers see the conflict and can decide whether to retry, rename -upstream, or skip. - -Each ``ingest`` call: - -1. Resolves the bucket date as today (local time). -2. Validates the inputs: ``path`` exists, ``channel`` matches the - allowed character class, ``description`` is non-empty, the - source basename has no path separators / dot segments / leading - dot. -3. Builds the final name from the format above. -4. Under a per-day file lock, checks the final name against - ``meta.json`` ∪ the on-disk listing. Any collision → error - (no silent suffixing). -5. Copies the asset into the bucket. -6. Appends a :class:`FileNode` row to ``meta.json`` (with provenance - on ``front_matter``) and re-renders ``.md``. - -Parameters: - -* ``path`` (required) — local filesystem path to the asset to ingest. -* ``channel`` (required) — inbound channel identifier (wechat / - email / browser / api / ...). Lowercase letters / digits / dashes - only. -* ``description`` (required) — analysis hint for downstream agents: - where the asset came from, what kind of content it carries, and how - it should be interpreted. The dreamer / auto_memory reads this - verbatim from ``meta.json`` to decide how to read the asset (skim - vs. deep parse, structured extraction vs. summarization, etc.), so - callers should write enough detail to drive that decision — not - just a title. Multi-line is fine; the ``.md`` bullet view - flattens for display while ``meta.json`` preserves the original. -* ``metadata`` (optional dict) — extras persisted on the meta row. - ``source`` (free-form origin within the channel) is conventional; - any other keys pass through verbatim. Keys ``name``, ``channel``, - ``received_at``, ``description`` are reserved. - -Returns ``{date, name, path}`` on success or ``{error}`` on failure. -""" - -import datetime -import fcntl -import json -import os -import re -import shutil -import tempfile -from pathlib import Path - -from ..base_step import BaseStep - -from ...components import R - -from ...schema import FileFrontMatter, FileNode - - -# Channel identifier character class — keeps the derived filename predictable -# and parseable (the `__` separator is also disjoint from this set). -_CHANNEL_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") - -# Keys the step manages itself — callers cannot smuggle them in via metadata. -_RESERVED_METADATA_KEYS = frozenset({"name", "channel", "received_at", "description"}) - - -@R.register("ingest_step") -class IngestStep(BaseStep): - """Capture an external asset into ``resource//`` and update the day's meta + index.""" - - async def execute(self): - assert self.context is not None - path: str = (self.context.get("path", "") or "").strip() - channel: str = (self.context.get("channel", "") or "").strip() - description: str = (self.context.get("description", "") or "").strip() - metadata_raw = self.context.get("metadata") or {} - - prepared, prep_error = _prepare_inputs(path, channel, description, metadata_raw) - if prep_error: - self._fail({"error": prep_error}) - self.logger.info(f"[{self.name}] ingest failed channel={channel!r} error={prep_error!r}") - return - - try: - outcome = self._land( - src=prepared["src"], - date=prepared["date"], - final_name=prepared["final_name"], - entry_fields={ - **prepared["metadata"], - "channel": channel, - "received_at": prepared["received_at"], - "description": description, - }, - ) - except _DuplicateIngest as e: - self._fail({"error": str(e)}) - self.logger.info(f"[{self.name}] ingest duplicate channel={channel!r} error={str(e)!r}") - return - except Exception as e: - self._fail({"error": f"{type(e).__name__}: {e}"}) - self.logger.info(f"[{self.name}] ingest crashed channel={channel!r} error={type(e).__name__}: {e}") - return - - self.context.response.success = True - self.context.response.answer = f"Ingested {outcome['name']} to {outcome['path']}" - self.context.response.metadata.update(outcome) - self.logger.info( - f"[{self.name}] channel={channel} date={outcome['date']} name={outcome['name']} path={outcome['path']}", - ) - - # ------------------------------------------------------------------ - - def _fail(self, payload: dict) -> None: - assert self.context is not None - self.context.response.success = False - self.context.response.answer = f"Error: {payload.get('error', 'ingest failed')}" - self.context.response.metadata.update(payload) - - def _resource_dir_name(self) -> str: - """Configured ``resource_dir`` subdir name.""" - return self.config_value("resource_dir") - - def _workspace_dir(self) -> Path: - vr = getattr(self.file_store, "workspace_path", None) - return Path(vr).resolve() if vr else Path.cwd().resolve() - - def _land(self, src: Path, date: str, final_name: str, entry_fields: dict) -> dict: - resource_dir = self._resource_dir_name() - bucket = self._workspace_dir() / resource_dir / date - bucket.mkdir(parents=True, exist_ok=True) - meta_path = bucket / "meta.json" - day_md = bucket / f"{date}.md" - lock_path = bucket / ".lock" - - rel_path = f"{resource_dir}/{date}/{final_name}" - - with _bucket_lock(lock_path): - existing_entries = _read_meta(meta_path) - # Collision check spans meta ∪ on-disk listing so a stray file - # (from a crashed earlier run) and case-insensitive filesystems - # both surface the conflict rather than getting silently clobbered. - on_disk = {p.name for p in bucket.iterdir() if p.is_file()} - existing_names = {Path(e.path).name for e in existing_entries} | on_disk - existing_names_folded = {name.casefold() for name in existing_names} - if final_name.casefold() in existing_names_folded: - raise _DuplicateIngest( - f"duplicate: {final_name!r} already exists in {resource_dir}/{date}/", - ) - - dst_path = bucket / final_name - shutil.copyfile(src, dst_path) - - # FileFrontMatter has first-class `name` / `description`; everything - # else (channel / source / received_at / passthrough metadata) rides - # the extras bag (model_config extra="allow"). - description = entry_fields.pop("description", "") - entry = FileNode( - path=rel_path, - st_mtime=dst_path.stat().st_mtime, - front_matter=FileFrontMatter(description=description, **entry_fields), - ) - updated = existing_entries + [entry] - _atomic_write_text( - meta_path, - json.dumps([e.model_dump() for e in updated], ensure_ascii=False, indent=2) + "\n", - ) - _atomic_write_text(day_md, _assemble_day_md(updated, date)) - - return { - "date": date, - "name": final_name, - "path": rel_path, - } - - -class _DuplicateIngest(Exception): - """Raised when the derived name already exists in the bucket.""" - - -# ---------------------------------------------------------------------- -# Input validation -# ---------------------------------------------------------------------- - - -def _prepare_inputs( - path: str, - channel: str, - description: str, - metadata_raw, -) -> tuple[dict, str]: - """Validate caller args and derive the bucket date / final name. - - Returns ``(prepared, error)``: on success ``prepared`` has - ``{src, date, received_at, final_name, metadata}`` and ``error`` is - empty; on failure ``prepared`` is ``{}`` and ``error`` carries the - first violation in user-facing order (path → channel → description - → metadata-shape → file existence → basename → metadata-keys). - """ - src = Path(path) if path else None - metadata, meta_error = _sanitize_metadata(metadata_raw) if isinstance(metadata_raw, dict) else ({}, "") - error = next( - ( - msg - for msg in ( - "path is required" if not path else "", - _validate_channel(channel), - "description is required" if not description else "", - "metadata must be a dict" if not isinstance(metadata_raw, dict) else "", - f"path not found: {path}" if src is not None and not src.is_file() else "", - _validate_basename(src.name) if src is not None else "", - meta_error, - ) - if msg - ), - "", - ) - if error: - return {}, error - - assert src is not None # narrowed by the "path is required" check - now = datetime.datetime.now() - return ( - { - "src": src, - "date": now.strftime("%Y-%m-%d"), - "received_at": now.isoformat(timespec="seconds"), - "final_name": f"{channel}__{now.strftime('%H%M%S')}__{src.name}", - "metadata": metadata, - }, - "", - ) - - -def _validate_channel(channel: str) -> str: - """Return an error string when ``channel`` is unsafe; empty when OK.""" - if not channel: - return "channel is required" - if not _CHANNEL_RE.match(channel): - return f"channel {channel!r} must be lowercase letters / digits / dashes " f"and start with a letter or digit" - return "" - - -def _validate_basename(name: str) -> str: - """Return an error string when the source basename is unsafe; empty when OK. - - The derived filename embeds this string after a ``__`` separator, so we - only need to block characters that would mangle the filesystem path — - path separators, dot segments, leading dots. Bookkeeping-name collisions - (``meta.json`` / ``.md``) are impossible by construction once the - channel + time prefix is prepended. - """ - if not name: - return "source basename is empty" - if "/" in name or "\\" in name: - return f"source basename must not contain path separators: {name!r}" - if name in {".", ".."}: - return f"source basename {name!r} is reserved" - if name.startswith("."): - return f"source basename may not start with '.': {name!r}" - return "" - - -def _sanitize_metadata(raw: dict) -> tuple[dict, str]: - """Return ``(cleaned_metadata, error)``. - - Rejects reserved keys (those the step manages itself) and coerces - ``source`` to a string. All other keys pass through verbatim so - callers can attach arbitrary tags that land on the entry's - ``front_matter`` extras. - """ - for key in _RESERVED_METADATA_KEYS: - if key in raw: - return {}, f"metadata key {key!r} is reserved" - - cleaned = dict(raw) - if "source" in cleaned: - src = cleaned["source"] - cleaned["source"] = src.strip() if isinstance(src, str) else "" - return cleaned, "" - - -# ---------------------------------------------------------------------- -# Per-day exclusive lock + atomic write helpers -# ---------------------------------------------------------------------- - - -class _bucket_lock: - """Exclusive ``flock`` on a per-day lock file; serializes meta+md writes.""" - - def __init__(self, lock_path: Path): - self.lock_path = lock_path - self._fp = None - - def __enter__(self): - self._fp = open(self.lock_path, "w", encoding="utf-8") - fcntl.flock(self._fp.fileno(), fcntl.LOCK_EX) - return self - - def __exit__(self, *exc): - if self._fp is not None: - try: - fcntl.flock(self._fp.fileno(), fcntl.LOCK_UN) - finally: - self._fp.close() - self._fp = None - - -def _read_meta(meta_path: Path) -> list[FileNode]: - """Read ``meta.json`` as a list of :class:`FileNode` rows; missing or malformed → [].""" - if not meta_path.is_file(): - return [] - try: - raw = json.loads(meta_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return [] - if not isinstance(raw, list): - return [] - out: list[FileNode] = [] - for row in raw: - if not isinstance(row, dict): - continue - try: - out.append(FileNode(**row)) - except Exception: - continue - return out - - -def _atomic_write_text(target: Path, text: str) -> None: - """Atomic text write via tempfile + os.replace in the same directory.""" - target.parent.mkdir(parents=True, exist_ok=True) - tmp_fd, tmp_path = tempfile.mkstemp(prefix=".ingest-", dir=target.parent) - try: - with os.fdopen(tmp_fd, "w", encoding="utf-8") as f: - f.write(text) - os.replace(tmp_path, target) - except Exception: - try: - os.unlink(tmp_path) - except OSError: - pass - raise - - -# ---------------------------------------------------------------------- -# Day-view markdown rendering (pure) -# ---------------------------------------------------------------------- - - -def _assemble_day_md(entries: list[FileNode], date: str) -> str: - """Render the day's bucket as a markdown view. - - Layout:: - - --- - name: - assets: [, , ...] - --- - - # resources - - - [[]] — from `` at — - - Provenance lives on each entry's ``front_matter`` (``channel`` / - ``source`` / ``received_at`` as extras, ``description`` as the - first-class field). ``received_at`` is rendered as ``HH:MM`` when it - parses as ISO 8601, else dropped silently — the asset list stays - readable even when upstream channels emit malformed timestamps. - ``source`` is dropped when empty. ``description`` is flattened - (newlines collapsed to spaces) so the bullet stays one line per - asset; ``meta.json`` preserves the verbatim multi-line text. - """ - names = [Path(e.path).name for e in entries] - lines: list[str] = [ - "---", - f"name: {date}", - f"assets: {json.dumps(names, ensure_ascii=False)}", - "---", - "", - f"# {date} resources", - "", - ] - for entry in entries: - fm = entry.front_matter - channel = getattr(fm, "channel", "") or "" - source = getattr(fm, "source", "") or "" - received_at = getattr(fm, "received_at", "") or "" - description = fm.description or "" - - bits: list[str] = [f"- [[{entry.path}]]"] - provenance = channel - if source: - provenance += f" from `{source}`" - time_part = _hhmm(received_at) - if time_part: - provenance += f" at {time_part}" - bits.append(provenance) - if description: - # Flatten so the bullet stays one line per asset; meta.json keeps - # the verbatim multi-line description for downstream consumers. - bits.append(" ".join(description.split())) - lines.append(" — ".join(bits)) - return "\n".join(lines) + "\n" - - -def _hhmm(received_at: str) -> str: - """Best-effort HH:MM extraction from an ISO 8601 timestamp.""" - if not received_at: - return "" - raw = received_at.replace("Z", "+00:00") - try: - return datetime.datetime.fromisoformat(raw).strftime("%H:%M") - except ValueError: - return "" diff --git a/reme/steps/transfer/upload.py b/reme/steps/transfer/upload.py index a86e8159..dceb0bc1 100644 --- a/reme/steps/transfer/upload.py +++ b/reme/steps/transfer/upload.py @@ -12,9 +12,9 @@ include a directory component so the caller is always explicit about where in the workspace the file lands. ``overwrite`` defaults to False — callers must opt in to clobber an existing destination. -For the resource-bucket ingest path (channel-tagged, dated under -``resource//`` with provenance metadata) use -``ingest`` instead. +To capture an external asset into the resource bucket, upload it under +``resource//`` (or loosely under ``resource/``); the resource +file watcher then interprets it into a source-linked daily note. """ import mimetypes diff --git a/reme/utils/service_utils.py b/reme/utils/service_utils.py index 35a077cd..0f5b0d07 100644 --- a/reme/utils/service_utils.py +++ b/reme/utils/service_utils.py @@ -2,9 +2,10 @@ import asyncio import socket -import subprocess import sys +import psutil + from ..constants import REME_DEFAULT_HOST, REME_DEFAULT_PORT @@ -28,34 +29,42 @@ async def find_reme(host: str, port: int) -> str: return "occupied" -def _sh(cmd: list[str]) -> str: - """Run cmd; return stdout, or '' on failure.""" - try: - return subprocess.check_output(cmd, stderr=subprocess.DEVNULL, text=True) - except (subprocess.CalledProcessError, FileNotFoundError): - return "" - - def _pid_on_port(port: int) -> int | None: - """PID listening on TCP port, or None.""" - out = _sh(["lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"]).strip() - return int(out.splitlines()[0]) if out else None + """PID listening on TCP ``port``, or None. Cross-platform via psutil. + + Iterates per-process rather than calling the system-wide + ``psutil.net_connections()`` — the latter needs root on macOS, while + per-process connection enumeration works without elevation for + processes the current user owns (which reme's own server always is). + """ + for proc in psutil.process_iter(["pid"]): + try: + for conn in proc.net_connections(kind="tcp"): + if conn.status == psutil.CONN_LISTEN and conn.laddr and conn.laddr.port == port: + return proc.info["pid"] + except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess): + continue + return None def _scan_reme_procs() -> list[tuple[int, str, int]]: """List running 'reme ... start' processes as (pid, host, port).""" procs: list[tuple[int, str, int]] = [] - for line in _sh(["pgrep", "-af", "reme.* start"]).splitlines(): - parts = line.split() - if not parts or not parts[0].isdigit(): + for proc in psutil.process_iter(["pid", "cmdline"]): + try: + cmdline = proc.info["cmdline"] or [] + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + # Match a `reme ... start` invocation (mirrors the old `pgrep -af`). + if "start" not in cmdline or not any("reme" in tok for tok in cmdline): continue host, port = REME_DEFAULT_HOST, REME_DEFAULT_PORT - for t in parts[1:]: + for t in cmdline: if t.startswith("service.host="): host = t.split("=", 1)[1] elif t.startswith("service.port=") and t.split("=", 1)[1].isdigit(): port = int(t.split("=", 1)[1]) - procs.append((int(parts[0]), host, port)) + procs.append((proc.info["pid"], host, port)) return procs diff --git a/tests/unit/test_service_utils.py b/tests/unit/test_service_utils.py new file mode 100644 index 00000000..d6107f78 --- /dev/null +++ b/tests/unit/test_service_utils.py @@ -0,0 +1,159 @@ +"""Tests for the psutil-based service discovery helpers in ``service_utils``. + +Covers the cross-platform replacements for the old ``lsof`` / ``pgrep`` +shell-outs: + +* ``_pid_on_port`` — real integration (open a listening socket, find its + PID) plus a mock test proving inaccessible processes are skipped, not + crashed on. +* ``_scan_reme_procs`` — mock ``psutil.process_iter`` to assert cmdline + parsing, defaulting, filtering, and AccessDenied tolerance. +""" + +# pylint: disable=protected-access,missing-function-docstring,unused-argument + +import os +import socket +from types import SimpleNamespace + +import psutil + +from reme.utils import service_utils as su + + +# ---------------------------------------------------------------------- +# Fakes for mocking psutil.process_iter +# ---------------------------------------------------------------------- + + +def _conn(port: int, status=psutil.CONN_LISTEN, has_laddr: bool = True): + return SimpleNamespace(status=status, laddr=SimpleNamespace(port=port) if has_laddr else None) + + +class _FakeProc: + """Stand-in for a psutil.Process yielded by process_iter.""" + + def __init__(self, pid: int, *, cmdline=None, conns=None, conn_exc=None): + self.info = {"pid": pid, "cmdline": cmdline if cmdline is not None else []} + self._conns = conns or [] + self._conn_exc = conn_exc + + def net_connections(self, kind="tcp"): + if self._conn_exc is not None: + raise self._conn_exc + return self._conns + + +class _RaisingInfoProc: + """Process whose `.info` access raises (simulates AccessDenied in iteration).""" + + def __init__(self, exc): + self._exc = exc + + @property + def info(self): + raise self._exc + + +def _patch_iter(monkeypatch, procs): + monkeypatch.setattr(su.psutil, "process_iter", lambda attrs=None: iter(procs)) + + +# ---------------------------------------------------------------------- +# _pid_on_port +# ---------------------------------------------------------------------- + + +def test_pid_on_port_finds_own_listening_socket(): + """Real integration: a listening socket is attributed to this process's PID.""" + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(("127.0.0.1", 0)) + srv.listen(1) + port = srv.getsockname()[1] + try: + assert su._pid_on_port(port) == os.getpid() + finally: + srv.close() + + +def test_pid_on_port_none_when_nobody_listening(): + """A port with no listener resolves to None.""" + # Grab then immediately release an ephemeral port to get a number that is + # (almost certainly) unbound right now. + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + free_port = s.getsockname()[1] + s.close() + assert su._pid_on_port(free_port) is None + + +def test_pid_on_port_skips_inaccessible_processes(monkeypatch): + """AccessDenied on one process must not abort the scan of the rest.""" + procs = [ + _FakeProc(111, conn_exc=psutil.AccessDenied(pid=111)), + _FakeProc(222, conn_exc=psutil.NoSuchProcess(pid=222)), + _FakeProc(333, conns=[_conn(9999)]), + ] + _patch_iter(monkeypatch, procs) + assert su._pid_on_port(9999) == 333 + + +def test_pid_on_port_ignores_non_listen_and_mismatched(monkeypatch): + """Only LISTEN sockets on the exact port match.""" + procs = [ + _FakeProc(1, conns=[_conn(9999, status=psutil.CONN_ESTABLISHED)]), # right port, wrong state + _FakeProc(2, conns=[_conn(8888)]), # listening, wrong port + _FakeProc(3, conns=[_conn(9999, has_laddr=False)]), # no laddr + ] + _patch_iter(monkeypatch, procs) + assert su._pid_on_port(9999) is None + + +# ---------------------------------------------------------------------- +# _scan_reme_procs +# ---------------------------------------------------------------------- + + +def test_scan_reme_procs_parses_host_and_port(monkeypatch): + procs = [ + _FakeProc( + 123, + cmdline=["python", "-m", "reme.reme", "start", "service.host=0.0.0.0", "service.port=8123"], + ), + ] + _patch_iter(monkeypatch, procs) + assert su._scan_reme_procs() == [(123, "0.0.0.0", 8123)] + + +def test_scan_reme_procs_defaults_when_args_absent(monkeypatch): + procs = [_FakeProc(7, cmdline=["reme", "start"])] + _patch_iter(monkeypatch, procs) + assert su._scan_reme_procs() == [(7, su.REME_DEFAULT_HOST, su.REME_DEFAULT_PORT)] + + +def test_scan_reme_procs_filters_unrelated(monkeypatch): + procs = [ + _FakeProc(1, cmdline=["reme", "find_reme"]), # no 'start' token + _FakeProc(2, cmdline=["python", "-m", "http.server", "start"]), # no reme token + _FakeProc(3, cmdline=[]), # empty cmdline + ] + _patch_iter(monkeypatch, procs) + assert not su._scan_reme_procs() + + +def test_scan_reme_procs_ignores_non_digit_port(monkeypatch): + """A malformed service.port= falls back to the default port, not a crash.""" + procs = [_FakeProc(9, cmdline=["reme", "start", "service.port=notaport"])] + _patch_iter(monkeypatch, procs) + assert su._scan_reme_procs() == [(9, su.REME_DEFAULT_HOST, su.REME_DEFAULT_PORT)] + + +def test_scan_reme_procs_skips_access_denied(monkeypatch): + """A process that denies cmdline access is skipped, others still scanned.""" + procs = [ + _RaisingInfoProc(psutil.AccessDenied(pid=99)), + _FakeProc(5, cmdline=["reme", "start"]), + ] + _patch_iter(monkeypatch, procs) + assert su._scan_reme_procs() == [(5, su.REME_DEFAULT_HOST, su.REME_DEFAULT_PORT)]