mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-10-08 03:10:24 +00:00
refactor(transfer): drop orphaned ingest step, make service discovery cross-platform (#300)
* refactor(transfer): drop orphaned ingest step, make service discovery cross-platform - Remove ingest step: superseded by auto_resource (drop files under resource/ → watcher interprets them); its meta.json/<date>.md outputs had no consumers and tripped the auto_resource watcher. - Replace lsof/pgrep shell-outs in service_utils with psutil (per-process enumeration, no root needed on macOS) for Windows/macOS/Linux support. - Add cross-platform test coverage for _pid_on_port / _scan_reme_procs. - Deps: +psutil, -filelock (only used by the removed ingest lock). * chore(release): bump version to 0.4.0.5
This commit is contained in:
Sen Huang
GitHub
parent
3dee10d4f9
commit
be7d1c0cf2
7 changed files with 190 additions and 475 deletions
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
"""ReMe CLI package."""
|
||||
|
||||
__version__ = "0.4.0.4"
|
||||
__version__ = "0.4.0.5"
|
||||
|
||||
from . import config
|
||||
from . import constants
|
||||
|
|
|
|||
|
|
@ -1,11 +1,9 @@
|
|||
"""Transfer steps."""
|
||||
|
||||
from .download import DownloadStep
|
||||
from .ingest import IngestStep
|
||||
from .upload import UploadStep
|
||||
|
||||
__all__ = [
|
||||
"DownloadStep",
|
||||
"IngestStep",
|
||||
"UploadStep",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -1,452 +0,0 @@
|
|||
"""``ingest`` — capture an externally-received asset into ``resource/<date>/``.
|
||||
|
||||
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/<YYYY-MM-DD>/``
|
||||
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/<YYYY-MM-DD>/
|
||||
meta.json # JSON array of FileNode rows
|
||||
<date>.md # derived markdown view
|
||||
<channel>__<HHMMSS>__<source-basename> # the asset itself
|
||||
...
|
||||
|
||||
Naming convention — the file name is **derived**, never caller-supplied::
|
||||
|
||||
<channel>__<HHMMSS>__<source-basename>
|
||||
|
||||
* ``<channel>`` — top-level identity (anchors provenance), lowercase
|
||||
letters/digits/dashes only.
|
||||
* ``<HHMMSS>`` — receive time within the bucket day (the date is
|
||||
already implicit in the bucket folder).
|
||||
* ``<source-basename>`` — 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/<date>/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 ``<date>.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 ``<date>.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/<date>/`` 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`` / ``<date>.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: <date>
|
||||
assets: [<name1>, <name2>, ...]
|
||||
---
|
||||
|
||||
# <date> resources
|
||||
|
||||
- [[<path>]] — <channel> from `<source>` at <hh:mm> — <description>
|
||||
|
||||
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 ""
|
||||
|
|
@ -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/<YYYY-MM-DD>/`` with provenance metadata) use
|
||||
``ingest`` instead.
|
||||
To capture an external asset into the resource bucket, upload it under
|
||||
``resource/<YYYY-MM-DD>/`` (or loosely under ``resource/``); the resource
|
||||
file watcher then interprets it into a source-linked daily note.
|
||||
"""
|
||||
|
||||
import mimetypes
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
159
tests/unit/test_service_utils.py
Normal file
159
tests/unit/test_service_utils.py
Normal file
|
|
@ -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)]
|
||||
Loading…
Add table
Reference in a new issue