From 4f620324e55e54c8188533d3471bf84a997723ee Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Wed, 27 May 2026 14:47:09 +0800 Subject: [PATCH] up --- reme4/steps/__init__.py | 2 - reme4/steps/graph/__init__.py | 7 -- reme4/steps/graph/traverse.py | 119 ------------------ tests4/unittest/test_common_steps.py | 182 +++++++++++---------------- 4 files changed, 75 insertions(+), 235 deletions(-) delete mode 100644 reme4/steps/graph/__init__.py delete mode 100644 reme4/steps/graph/traverse.py diff --git a/reme4/steps/__init__.py b/reme4/steps/__init__.py index 614fa2b3..5407633d 100644 --- a/reme4/steps/__init__.py +++ b/reme4/steps/__init__.py @@ -38,14 +38,12 @@ from . import index # noqa: F401 -- registers update_catalog_step / update_ind # from . import jobs # noqa: F401 -- registers synchronizer / digester from .base_step import BaseStep -from . import graph # noqa: F401 __all__ = [ "background", "common", "crud", "transfer", - "graph", "index", "frontmatter", "daily", diff --git a/reme4/steps/graph/__init__.py b/reme4/steps/graph/__init__.py deleted file mode 100644 index c5147dce..00000000 --- a/reme4/steps/graph/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -"""Graph steps.""" - -from .traverse import GraphTraverseStep - -__all__ = [ - "GraphTraverseStep", -] diff --git a/reme4/steps/graph/traverse.py b/reme4/steps/graph/traverse.py deleted file mode 100644 index 25326f56..00000000 --- a/reme4/steps/graph/traverse.py +++ /dev/null @@ -1,119 +0,0 @@ -"""``graph_traverse_step`` — BFS over wikilink edges from a seed file. - -Single tool for relationship browsing. ``depth=1`` covers the trivial -"what does this link to / what links here" lookups (set ``direction`` -accordingly); higher depth opens up multi-hop exploration. - -Output is one record per edge traversed (not per node), so the same -target can appear multiple times if reached via different predicates -or paths — agents dedupe at the call site if they want a flat node -set. Each record carries ``via`` (the predecessor) and the link's -``predicate`` / ``anchor`` so the agent can reconstruct the path. - -Adjacency is loaded once via ``file_graph.get_nodes(None)`` — every -real node arrives with its full ``links`` payload, and we build both -the outbound and the inbound index in a single pass. The BFS then -runs purely in memory: no per-frontier-node graph round-trips, no -filesystem walk. The ``get_inlinks`` / ``get_outlinks`` contract -methods stay unused here because they'd add network round-trips for -data we already have. - -Direction vocabulary accepts both the standard convention -(``forward`` / ``backward`` / ``both``) and the engine convention -(``out`` / ``in`` / ``both``). - -The seed ``path`` is taken as-is (vault-relative). A seed that doesn't -match any graph node yields an empty result (no error). -""" - -from collections import deque - -from ..base_step import BaseStep -from ...components import R -from ...schema import FileLink - - -_FORWARD = {"out", "forward"} -_BACKWARD = {"in", "backward"} -_BOTH = {"both"} -_VALID_DIRECTIONS = _FORWARD | _BACKWARD | _BOTH - - -@R.register("graph_traverse_step") -class GraphTraverseStep(BaseStep): - """BFS from a seed file to explore wikilink relationships. - - Parameters: - path — seed path (vault-relative). - direction — ``forward`` / ``backward`` / ``both`` (or ``out`` / ``in`` / ``both``). - depth — hop limit (default 1 = immediate neighbors). - predicate — optional edge-type filter; ``None`` = no filter. - """ - - async def execute(self): - """BFS from ``path`` and emit one record per traversed edge.""" - assert self.context is not None - seed = str(self.context.get("path") or "").strip() - assert seed, "path is required" - max_depth = int(self.context.get("depth") or 1) - direction = (self.context.get("direction") or "both").lower() - predicate = self.context.get("predicate") - assert ( - direction in _VALID_DIRECTIONS - ), f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}" - - # Build outbound / inbound adjacency in one pass over all nodes. - outbound: dict[str, list[tuple[str, FileLink]]] = {} - inbound: dict[str, list[tuple[str, FileLink]]] = {} - if self.file_store.file_graph: - for node in await self.file_store.file_graph.get_nodes(): - for link in node.links: - if not link.target_path: - continue - outbound.setdefault(node.path, []).append((link.target_path, link)) - inbound.setdefault(link.target_path, []).append((node.path, link)) - - walk_out = direction in _FORWARD or direction in _BOTH - walk_in = direction in _BACKWARD or direction in _BOTH - - visited_edges: set[tuple[str, str, str | None]] = set() - results: list[dict] = [] - queue: deque[tuple[str, int]] = deque([(seed, 0)]) - - while queue: - current, depth = queue.popleft() - if depth >= max_depth: - continue - - edges: list[tuple[str, str | None, str | None]] = [] - if walk_out: - for tgt, link in outbound.get(current, ()): - if predicate is not None and link.predicate != predicate: - continue - edges.append((tgt, link.predicate, link.target_anchor)) - if walk_in: - for src, link in inbound.get(current, ()): - if predicate is not None and link.predicate != predicate: - continue - edges.append((src, link.predicate, link.target_anchor)) - - for next_path, pred, anchor in edges: - edge_key = (current, next_path, pred) - if edge_key in visited_edges: - continue - visited_edges.add(edge_key) - results.append( - { - "path": next_path, - "depth": depth + 1, - "via": current, - "predicate": pred, - "anchor": anchor, - }, - ) - if depth + 1 < max_depth: - queue.append((next_path, depth + 1)) - - self.context.response.success = True - self.context.response.answer = f"Traversed {len(results)} edge(s) from {seed}" - self.context.response.metadata.update({"edges": results, "count": len(results)}) diff --git a/tests4/unittest/test_common_steps.py b/tests4/unittest/test_common_steps.py index b0d48a81..52c4d270 100644 --- a/tests4/unittest/test_common_steps.py +++ b/tests4/unittest/test_common_steps.py @@ -2,10 +2,10 @@ Two surfaces share this file: -* **HTTP / MCP E2E tests** (top half) spawn ``reme4 start`` via - ``mock_reme_server`` and drive ``version`` / ``help`` / ``search`` / - ``init`` / ``demo`` over the wire. Each test uses an isolated cwd so - the vault (``.reme`` by default) does not collide. +* **In-process job tests** (top half) build an ``Application`` from the + default config and call ``run_job`` directly — no subprocess, no HTTP. + Each test uses an isolated cwd so the vault (``.reme`` by default) + does not collide. * **Direct unit tests** (bottom half) exercise ``TraverseStep`` (registered as ``traverse_step``) — BFS over wikilink edges from a seed file, forward / backward / both — against a freshly built @@ -19,11 +19,12 @@ import os import tempfile import warnings -from reme4 import __version__ as REME_VERSION +from reme4 import Application, __version__ as REME_VERSION from reme4.components.file_store import LocalFileStore +from reme4.config import resolve_app_config from reme4.schema import FileLink, FileNode from reme4.steps.common import traverse as traverse_mod -from reme4.utils import call_action, call_and_check, mock_reme_server +from reme4.utils import load_env warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba") warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources") @@ -50,6 +51,15 @@ def _run(coro): asyncio.run(coro) +async def _make_app() -> Application: + """Build and start an Application with the default config, logging silenced.""" + load_env() + cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False) + app = Application(**cfg) + await app.start() + return app + + def _node(path: str, links: list[tuple[str, str | None, str | None]] | None = None) -> FileNode: """Build a FileNode with (target_path, target_anchor, predicate) outgoing edges.""" return FileNode( @@ -73,7 +83,7 @@ def _edges(step) -> list[dict]: # =========================================================================== -# HTTP / MCP E2E tests: version / help / search / init / demo +# In-process job tests: version / help / health_check / search / reindex # =========================================================================== @@ -82,18 +92,14 @@ def test_version_job(): async def run(): with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - await call_and_check( - "version", - host=host, - port=port, - validator=lambda r: ( - isinstance(r, dict) - and r.get("success") is True - and r.get("answer") == REME_VERSION - and r.get("metadata", {}).get("version") == REME_VERSION - ), - ) + app = await _make_app() + try: + resp = await app.run_job("version") + assert resp.success is True + assert resp.answer == REME_VERSION + assert resp.metadata.get("version") == REME_VERSION + finally: + await app.close() print("✓ test_version_job passed") _run(run()) @@ -104,24 +110,17 @@ def test_help_job(): async def run(): with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - result = await call_and_check( - "help", - host=host, - port=port, - validator=lambda r: ( - isinstance(r, dict) - and r.get("success") is True - and isinstance(r.get("answer"), str) - and r.get("metadata", {}).get("job_count", 0) > 0 - and "`help`" not in r["answer"] - ), - ) - # Spot-check that a couple of known jobs appear in the listing. - answer = result["answer"] + app = await _make_app() + try: + resp = await app.run_job("help") + assert resp.success is True + assert isinstance(resp.answer, str) + assert resp.metadata.get("job_count", 0) > 0 + assert "`help`" not in resp.answer for expected_job in ("version", "health_check", "search"): - if expected_job not in answer: - raise AssertionError(f"help output missing job {expected_job!r}: {answer!r}") + assert expected_job in resp.answer, f"help missing {expected_job!r}: {resp.answer!r}" + finally: + await app.close() print("✓ test_help_job passed") _run(run()) @@ -132,92 +131,61 @@ def test_search_job_empty_store(): async def run(): with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - await call_and_check( - "search", - host=host, - port=port, - query="hello world", - limit=5, - validator=lambda r: ( - isinstance(r, dict) - and r.get("success") is True - and isinstance(r.get("metadata"), dict) - and isinstance(r["metadata"].get("counts"), dict) - and r["metadata"]["counts"].get("returned", -1) == 0 - ), - ) + app = await _make_app() + try: + resp = await app.run_job("search", query="hello world", limit=5) + assert resp.success is True + counts = resp.metadata.get("counts", {}) + assert isinstance(counts, dict) + assert counts.get("returned", -1) == 0 + finally: + await app.close() print("✓ test_search_job_empty_store passed") _run(run()) def test_search_job_missing_query(): - """search without a query should surface the assertion error in `answer`.""" + """search with empty query returns success=False and a query-related error in answer.""" async def run(): with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - result = await call_action("search", host=host, port=port, query="") - if not isinstance(result, dict): - raise AssertionError(f"expected dict response, got {result!r}") - if "query" not in str(result.get("answer", "")).lower(): - raise AssertionError(f"expected query-related error in answer, got {result!r}") + app = await _make_app() + try: + resp = await app.run_job("search", query="") + assert resp.success is False + assert "query" in str(resp.answer).lower() + finally: + await app.close() print("✓ test_search_job_missing_query passed") _run(run()) -# -- aggregate: reuse one server instance for all jobs ------------------- - - -def test_all_jobs_one_server(): - """Run every common job against a single shared server for efficiency.""" +def test_all_jobs_single_app(): + """Run every common job against one shared in-process Application for efficiency.""" async def run(): with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp): - async with mock_reme_server() as (host, port): - # version - await call_and_check( - "version", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) and r.get("answer") == REME_VERSION, - ) - # help - await call_and_check( - "help", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) and r.get("metadata", {}).get("job_count", 0) > 0, - ) - # health_check - await call_and_check( - "health_check", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) - and isinstance( - r.get("metadata", {}).get("health"), - dict, - ), - ) - # search (empty store) - await call_and_check( - "search", - host=host, - port=port, - query="anything", - validator=lambda r: isinstance(r, dict) and r.get("success") is True, - ) - # reindex - await call_and_check( - "reindex", - host=host, - port=port, - validator=lambda r: isinstance(r, dict) and isinstance(r.get("metadata", {}).get("counts"), dict), - ) - print("✓ test_all_jobs_one_server passed") + app = await _make_app() + try: + resp = await app.run_job("version") + assert resp.answer == REME_VERSION + + resp = await app.run_job("help") + assert resp.metadata.get("job_count", 0) > 0 + + resp = await app.run_job("health_check") + assert isinstance(resp.metadata.get("health"), dict) + + resp = await app.run_job("search", query="anything") + assert resp.success is True + + resp = await app.run_job("reindex") + assert isinstance(resp.metadata.get("counts"), dict) + finally: + await app.close() + print("✓ test_all_jobs_single_app passed") _run(run()) @@ -365,12 +333,12 @@ def test_traverse_both_directions(): if __name__ == "__main__": - print("\n=== reme4 common steps E2E tests ===") + print("\n=== reme4 common steps in-process tests ===") test_version_job() test_help_job() test_search_job_empty_store() test_search_job_missing_query() - test_all_jobs_one_server() + test_all_jobs_single_app() print("\n=== traverse step tests ===") test_traverse_forward_depth_1() test_traverse_backward_returns_inlinks()