napseer/nap-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d86610da8e0ac03f730fb7ebf9e5bd98bea7a873
nap-cli/tests/test_mcp_reliability_contracts.py
T
HumbleTurtleandClaude Opus 5.5 0938b3026e Keep the local index fresh from the change feed and improve ranking 
- Track a feed cursor in the local index: reindex takes the head cursor
  before listing and commits it with the snapshot; reads refresh from
  /changes (bounded, throttled) and apply creates, updates, archives,
  restores and hard deletes atomically with the cursor. Existing
  cursorless indexes bootstrap tracking on first read. Failures answer
  locally with explicit freshness warnings; 400 triggers a full
  reindex, 404 falls back to timestamp sync.
- Rank by a weighted score: exact identity (including aliases and
  paths), term coverage including body, intent, bm25, authority, a
  status penalty for superseded/completed records and mild recency.
- Return query-centred FTS5 snippets (never for E2E nodes).
- FTS schema v2: porter stemming, prefix index, title column, rowid
  deletes, two-character terms, domain words no longer stopwords;
  v1 indexes upgrade in place.
- Hyphenated alias/path literals match as adjacent-token phrases and
  rank as exact identity.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 22:09:12 -03:00

602 lines
25 KiB
Python
Raw Blame History

"""Regressions for write idempotency, stale local reads, atomic claim-next,
indexed link counts, and locator-declared tool profiles."""
import importlib.util
import io
import json
import pathlib
import sys
import urllib.error
import uuid
import pytest
def load_module():
script_path = (
pathlib.Path(__file__).resolve().parents[1]
/ "resources"
/ "scripts"
/ "napseer_mcp_server.py"
)
sys.path.insert(0, str(script_path.parent))
spec = importlib.util.spec_from_file_location("napseer_mcp_reliability_test", script_path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
@pytest.fixture
def mod(monkeypatch, tmp_path):
monkeypatch.delenv("NAPSEER_TOOL_PROFILES", raising=False)
module = load_module()
module.DEFAULT_PROJECT_ID = "project-1"
module.TOKEN = "synthetic-access"
module.REFRESH_TOKEN = "synthetic-refresh"
module.TOKEN_EXPIRES_AT = "2099-01-01T00:00:00Z"
module.REFRESH_EXPIRES_AT = None
module.AUTH = {**module.AUTH, "account_mode": "operator_account", "project_id": "project-1"}
module.refresh_public_auth_state = lambda: module.AUTH
module.project_memory_encryption_active = lambda _project_id: False
module.AUTH_DIR = tmp_path / "state"
module.INDEX_PATH = tmp_path / "state" / "index.sqlite"
module.INDEX_LOCK_PATH = tmp_path / "state" / "index.lock"
module.resolve_project_id = lambda _args: "project-1"
return module
class FakeResponse:
def __init__(self, payload):
self._body = json.dumps(payload).encode("utf-8")
def read(self, *_args):
return self._body
def __enter__(self):
return self
def __exit__(self, *_exc):
return False
def http_error(status, body):
return urllib.error.HTTPError(
"https://api.example.test/x", status, "error", {}, io.BytesIO(json.dumps(body).encode("utf-8"))
)
def stub_project_lock(mod):
mod.acquire_project_lock = lambda args=None, **kwargs: {"id": "lock-1", "lease_token": "lease-1"}
mod.cleanup_project_lock = lambda project_id, lock: None
# -- 1. Idempotency keys ----------------------------------------------------
def test_create_node_idempotency_key_survives_auth_retry_and_is_returned(mod):
stub_project_lock(mod)
seen = []
def fake_urlopen(request, timeout=None):
seen.append((request.get_method(), request.full_url, request.get_header("Idempotency-key"),
request.get_header("Authorization")))
if len(seen) == 1:
raise http_error(401, {"error": {"code": "unauthorized"}})
return FakeResponse({"id": "node-new", "project_id": "project-1", "full_path": "/notes/a",
"folder_path": "/notes", "name": "a", "type": "note", "content_text": "x"})
def fake_renew(stale_token=None):
mod.TOKEN = "renewed-access"
return {"status": "renewed"}
mod.api_urlopen = fake_urlopen
mod.renew_auth = fake_renew
result = mod.call_tool_impl("nap_create_node", {"folder_path": "/notes", "name": "a", "content_text": "x"})
assert len(seen) == 2
generated = seen[0][2]
assert generated and str(uuid.UUID(generated)) == generated
# One logical request: the internal auth replay reuses the same key.
assert seen[1][2] == generated
assert result["idempotency_key"] == generated
seen.clear()
mod.api_urlopen = lambda request, timeout=None: seen.append(request.get_header("Idempotency-key")) or FakeResponse(
{"id": "node-new", "project_id": "project-1", "full_path": "/notes/a"}
)
result = mod.call_tool_impl(
"nap_create_node", {"folder_path": "/notes", "name": "a", "content_text": "x", "idempotency_key": "caller-key-1"}
)
assert seen == ["caller-key-1"]
assert result["idempotency_key"] == "caller-key-1"
def test_node_patch_sends_header_not_body_and_bulk_items_get_keys(mod):
stub_project_lock(mod)
existing = {"id": "node-123", "project_id": "project-1", "full_path": "/notes/current",
"folder_path": "/notes", "name": "current", "type": "note", "content_text": "old"}
mod.get_node_by_id = lambda args: existing
mod.try_get_node_by_path = lambda path, allow_agent=False: None
mod.index_node = lambda node: None
writes = []
def fake_request(method, path, payload=None, **kwargs):
writes.append((method, path, payload, dict(kwargs.get("extra_headers") or {})))
if path.endswith("/patch"):
return {"ok": True, "data": {"node": existing, "revision": "r2", "read_fingerprint": "f2"}}
return {**existing, "id": "node-new" if method == "POST" else existing["id"]}
mod.request_json = fake_request
patched = mod.guarded_node_patch({
"node_id": "node-123",
"precondition": {"revision": "r1", "read_fingerprint": "f1"},
"merge_metadata": {"status": "active"},
"idempotency_key": "patch-key-1",
})
method, path, payload, headers = writes[-1]
assert (method, path) == ("PATCH", "/v1/projects/project-1/nodes/node-123/patch")
assert "idempotency_key" not in payload
assert headers["Idempotency-Key"] == "patch-key-1"
assert headers["X-Napseer-Lock-Id"] == "lock-1"
assert patched["idempotency_key"] == "patch-key-1"
generated = mod.guarded_node_patch({
"node_id": "node-123",
"precondition": {"revision": "r1", "read_fingerprint": "f1"},
"merge_metadata": {"status": "active"},
})
assert generated["idempotency_key"] == writes[-1][3]["Idempotency-Key"]
uuid.UUID(generated["idempotency_key"])
writes.clear()
bulk = mod.bulk_upsert_nodes({
"idempotency_key": "bulk-base",
"nodes": [
{"path": "/notes/new", "content_text": "new"},
{"node_id": "node-123", "content_text": "updated", "idempotency_key": "item-own"},
],
})
assert [(w[0], w[3]["Idempotency-Key"]) for w in writes] == [("POST", "bulk-base:0"), ("PATCH", "item-own")]
assert [item["idempotency_key"] for item in bulk["items"]] == ["bulk-base:0", "item-own"]
for name in ("nap_create_node", "nap_node_patch", "nap_bulk"):
assert mod.tool_contract_metadata(name)["idempotency"] == "same-idempotency-key"
tools = {tool["name"]: tool for tool in mod.raw_tools()}
assert "idempotency_key" in tools["nap_create_node"]["inputSchema"]["properties"]
assert "idempotency_key" in tools["nap_bulk"]["inputSchema"]["properties"]
assert "idempotency_key" in tools["nap_bulk"]["inputSchema"]["properties"]["nodes"]["items"]["properties"]
# -- 2. Stale local reads and whoami auth precedence -----------------------
def indexed(path, links=None):
folder, name = path.rsplit("/", 1)
return {
"id": f"id-{name}",
"project_id": "project-1",
"full_path": path,
"folder_path": folder,
"name": name,
"type": "note",
"tags": [],
"aliases": [],
"links": links or [],
"metadata": {"status": "active"},
"content_text": f"body of {name} gateway reliability",
"encryption_state": "legacy_plaintext",
"updated_at": "2026-08-22T00:00:00Z",
}
def build_complete_index(mod):
mod.index_node(indexed("/plans/alpha", links=[{"path": "/plans/beta", "relation": "references"}]))
mod.index_node(indexed("/plans/beta"))
mod.set_local_graph_index_complete("project-1", True)
def record_revoked_grant(mod):
def rejected(*_args, **_kwargs):
raise mod.SafeToolError("http_401", "rejected", status=401)
mod.request_form_json = rejected
with pytest.raises(mod.SafeToolError) as caught:
mod.renew_auth()
assert caught.value.code == "auth_required"
assert "expired or revoked" in caught.value.safe_message
def test_local_answers_warn_when_login_grant_is_revoked(mod):
build_complete_index(mod)
def empty_feed(_method, path, *_args, **_kwargs):
if "/changes?" in path:
return {"items": [], "next_cursor": "c0", "has_more": False}
return {"items": [], "next_cursor": None}
# A healthy read starts change-feed tracking for the cursorless index.
mod.request_json = empty_feed
healthy = mod.discover_memory({"folder_path": "/plans", "view": "paths"})
assert not healthy.get("warnings")
assert not (healthy.get("freshness") or {}).get("stale")
assert healthy["freshness"]["feed_sync"] == "tracking_initialized"
mod.request_json = lambda *a, **k: (_ for _ in ()).throw(AssertionError("no HTTP expected"))
record_revoked_grant(mod)
state_text = (mod.AUTH_DIR / "auth-failure.json").read_text()
assert "synthetic-refresh" not in state_text and "synthetic-access" not in state_text
browsed = mod.discover_memory({"folder_path": "/plans", "view": "paths"})
assert [w["code"] for w in browsed["warnings"]] == ["auth_required"]
assert browsed["warnings"][0]["indexed_at"]
assert browsed["freshness"]["stale"] is True
assert browsed["freshness"]["graph_complete"] is True
context = mod.get_memory_context({"node_id": "id-alpha", "depth": 1})
codes = [w["code"] for w in context["warnings"] if isinstance(w, dict)]
assert codes == ["auth_required"]
assert context["freshness"]["stale"] is True
assert context["edge_count"] >= 1
status = mod.gateway_status()
assert status["status"] == "auth_required"
assert status["auth_problem"]["reason"] == "auth_required"
# A new login invalidates the recorded failure without clearing files.
mod.TOKEN = "fresh-access"
mod.REFRESH_TOKEN = "fresh-refresh"
assert mod.current_auth_problem() is None
assert mod.gateway_status()["status"] != "auth_required"
def test_hybrid_search_falls_back_to_local_with_server_unavailable_warning(mod):
build_complete_index(mod)
def server_down(*_args, **_kwargs):
raise mod.SafeToolError("http_503", "Napseer GET request failed with HTTP 503.", status=503)
mod.request_json = server_down
result = mod.discover_memory({"q": "gateway reliability", "limit": 25})
assert {item["full_path"] for item in result["items"]} == {"/plans/alpha", "/plans/beta"}
assert [w["code"] for w in result["warnings"]] == ["server_unavailable"]
assert result["freshness"]["stale"] is True
assert result["freshness"]["reason"] == "server_unavailable"
# A non-recoverable server error is not masked by the local answer.
mod.request_json = lambda *a, **k: (_ for _ in ()).throw(mod.SafeToolError("http_403", "forbidden", status=403))
with pytest.raises(mod.SafeToolError):
mod.discover_memory({"q": "gateway reliability", "limit": 25})
def test_whoami_reports_expired_access_without_refresh_ahead_of_gateway_state(mod):
mod.REFRESH_TOKEN = None
mod.TOKEN_EXPIRES_AT = "2020-01-01T00:00:00Z"
status = mod.gateway_status()
assert status["status"] == "auth_required"
assert status["auth_problem"]["reason"] == "access_token_expired_without_refresh"
# -- 3. Atomic claim-next ---------------------------------------------------
def kanban_card(name, priority, rank):
return {
"id": f"id-{name}",
"project_id": "project-1",
"full_path": f"/kanban/todo/{name}",
"folder_path": "/kanban/todo",
"name": name,
"type": "kanban_card",
"tags": ["kanban"],
"links": [],
"metadata": {"status": "todo", "column": "todo", "priority": priority, "rank": rank, "blocked": False},
"content_text": "card",
"updated_at": "2026-05-06T12:00:00Z",
"archived_at": None,
"encryption_state": "plaintext",
}
def kanban_backend(mod, claim_outcomes):
cards = [kanban_card("first", "urgent", "00001024"), kanban_card("second", "high", "00001024"),
kanban_card("third", "normal", "00001024")]
claims = []
def fake_request(method, path, payload=None, **kwargs):
if method == "GET":
return {"items": cards, "next_cursor": None}
assert (method, path) == ("POST", "/v1/projects/project-1/operations")
claims.append(payload)
outcome = claim_outcomes[len(claims) - 1]
if isinstance(outcome, Exception):
raise outcome
return {**outcome, "id": payload["id"], "node_id": payload["node_id"]}
mod.request_json = fake_request
return claims
def held_error(mod, legacy=False):
mod_error = mod.safe_http_error(
http_error(409, {}),
operation="POST request",
body_text=json.dumps(
{"error": {"code": "conflict", "message": "Another execution holds this work until it expires."}}
if legacy else
{"error": {"code": "operation_held", "message": "held", "details": {"active_operation_id": "op-active"}}}
),
)
return mod_error
def test_claim_next_skips_held_cards_and_claims_in_pick_order(mod):
claims = kanban_backend(mod, [held_error(mod), held_error(mod, legacy=True), {"state": "running"}])
mod.CONFIGURED_TOOL_PROFILES = {"workflow"}
result = mod.call_tool_impl("nap_kanban_claim_next", {"holder": "agent-a", "ttl_seconds": 120})
assert [c["node_id"] for c in claims] == ["id-first", "id-second", "id-third"]
assert len({c["id"] for c in claims}) == 3
assert all(c["holder"] == "agent-a" and c["ttl_seconds"] == 120 for c in claims)
assert result["claimed"]["full_path"] == "/kanban/todo/third"
assert result["operation"]["state"] == "running"
assert result["operation_id"] == claims[-1]["id"]
assert [s["reason"] for s in result["skipped"]] == ["operation_held", "operation_held"]
assert result["skipped"][0]["active_operation_id"] == "op-active"
def test_claim_next_returns_null_when_all_held_and_stops_on_ambiguous_error(mod):
kanban_backend(mod, [held_error(mod)] * 3)
none = mod.claim_next_kanban_card({"holder": "agent-a"})
assert none["ok"] is True and none["claimed"] is None
assert none["reason"] == "all_candidates_held"
claims = kanban_backend(mod, [held_error(mod), mod.SafeToolError("request_timeout", "timed out")])
with pytest.raises(mod.SafeToolError) as caught:
mod.claim_next_kanban_card({"holder": "agent-a"})
# Stops at the ambiguous claim; never retries or moves on.
assert len(claims) == 2
payload = mod.safe_exception_payload(caught.value)
assert payload["code"] == "kanban_claim_unresolved"
assert payload["details"]["attempted_operation_id"] == claims[1]["id"]
assert payload["details"]["cause"]["code"] == "request_timeout"
assert claims[1]["id"] in payload["message"]
claims = kanban_backend(mod, [held_error(mod), held_error(mod)])
capped = mod.claim_next_kanban_card({"holder": "agent-a", "max_candidates": 2})
assert len(claims) == 2 and capped["reason"] == "max_candidates_held"
with pytest.raises(mod.SafeToolError):
mod.claim_next_kanban_card({})
def test_claim_next_is_registered_as_workflow_mutation(mod):
assert "nap_kanban_claim_next" in mod.CANONICAL_TOOL_NAMES
assert mod.tool_profile("nap_kanban_claim_next") == "workflow"
assert mod.tool_category("nap_kanban_claim_next") == "kanban"
metadata = mod.tool_contract_metadata("nap_kanban_claim_next")
assert metadata["side_effect"] == "remote-write"
assert metadata["lock_policy"] == "atomic-execution-lease"
assert "nap_kanban_claim_next" not in mod.READ_ONLY_TOOLS
# -- 4. links_count from indexed edges -------------------------------------
def test_local_discovery_rows_report_indexed_link_counts(mod):
build_complete_index(mod)
mod.request_json = lambda *a, **k: (_ for _ in ()).throw(AssertionError("no HTTP expected"))
searched = mod.search_local_index({"q": "gateway reliability", "limit": 10})
by_path = {item["full_path"]: item for item in searched["items"]}
assert by_path["/plans/alpha"]["links_count"] == 1
assert by_path["/plans/beta"]["backlinks_count"] == 1
rows = mod.discover_memory({"q": "gateway reliability", "view": "summary", "mode": "local"})["items"]
rows = {item["full_path"]: item for item in rows}
assert rows["/plans/alpha"]["links_count"] == 1
assert rows["/plans/beta"]["links_count"] == 0
assert rows["/plans/beta"]["backlinks_count"] == 1
# Unknown counts are omitted rather than reported as zero.
summary = mod.summarize_node({"id": "x", "full_path": "/a/x", "name": "x"}, view="summary")
assert "links_count" not in summary
# -- 5. Tool profiles from the committed locator ---------------------------
def write_locator(mod, **extra):
path = mod.project_locator_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({
"schema": "napseer.project.v1",
"base_url": "https://api.example.test",
"project_id": "11111111-1111-4111-8111-111111111111",
"project_slug": "example",
**extra,
}))
def test_locator_tool_profiles_apply_when_env_unset(mod, monkeypatch):
mod.CONFIGURED_TOOL_PROFILES = None
assert mod.configured_tool_profiles() == set()
write_locator(mod, tool_profiles=["workflow", "operations", "bogus"])
assert mod.load_project_locator()["project_slug"] == "example"
assert mod.configured_tool_profiles() == {"workflow", "operations"}
assert "nap_kanban_claim_next" in mod.enabled_tool_names()
apropos = mod.explore_tools({})
assert apropos["profile_source"] == "project_locator"
assert apropos["warnings"][0]["code"] == "unknown_tool_profile"
assert apropos["warnings"][0]["ignored"] == ["bogus"]
mod.BASE_URL = "https://api.example.test"
mod.local_anchor_integrity = lambda: {}
mod.index_status = lambda args: {}
doctor = mod.doctor({})
assert doctor["warnings"][0]["code"] == "unknown_tool_profile"
monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "maintenance")
assert mod.configured_tool_profiles() == {"maintenance"}
assert mod.explore_tools({})["profile_source"] == "environment"
monkeypatch.delenv("NAPSEER_TOOL_PROFILES")
write_locator(mod, tool_profiles="workflow")
assert mod.configured_tool_profiles() == set()
assert mod.explore_tools({})["warnings"][0]["code"] == "invalid_tool_profiles"
# -- 6. Change feed cursor --------------------------------------------------
def changes_backend(mod, responses):
calls = []
def fake_request(method, path, payload=None, **kwargs):
calls.append((method, path))
outcome = responses[len(calls) - 1]
if isinstance(outcome, Exception):
raise outcome
return outcome
mod.request_json = fake_request
return calls
def test_changes_since_starts_at_head_then_advances_only_on_success(mod, tmp_path):
change = {"type": "node_updated", "cursor": "c1.5.9", "event_id": "e1", "project_id": "project-1",
"node_id": "n1", "full_path": "/plans/a", "changed_fields": ["content_text"]}
calls = changes_backend(mod, [
{"items": [], "next_cursor": "c1.5.8", "has_more": False},
{"items": [change], "next_cursor": "c1.5.9", "has_more": True},
mod.SafeToolError("http_503", "unavailable", status=503),
{"items": [], "next_cursor": "c1.9.9", "has_more": False},
])
first = mod.call_tool_impl("nap_changes_since", {})
assert calls[0] == ("GET", "/v1/projects/project-1/changes?after=head&limit=100")
assert first["started_at_head"] is True and first["items"] == []
stored = json.loads((mod.AUTH_DIR / "changes-cursor.json").read_text())
assert stored["project_id"] == "project-1" and stored["cursor"] == "c1.5.8"
assert not (tmp_path / ".napseer" / "changes-cursor.json").exists()
second = mod.changes_since({"limit": 50})
assert calls[1] == ("GET", "/v1/projects/project-1/changes?after=c1.5.8&limit=50")
assert second["items"] == [change] and second["has_more"] is True and second["cursor_advanced"] is True
with pytest.raises(mod.SafeToolError) as caught:
mod.changes_since({})
assert caught.value.details["cursor_advanced"] is False
assert mod.read_changes_cursor("project-1") == "c1.5.9"
peek = mod.changes_since({"advance": False})
assert peek["cursor_advanced"] is False
assert mod.read_changes_cursor("project-1") == "c1.5.9"
def test_changes_since_errors_project_binding_and_reset(mod):
mod.write_changes_cursor("other-project", "c1.1.1")
calls = changes_backend(mod, [
mod.safe_http_error(http_error(404, {}), body_text="{}"),
mod.safe_http_error(http_error(400, {}), body_text=json.dumps({"error": {"code": "validation_error"}})),
{"items": [], "next_cursor": "c1.2.2", "has_more": False},
])
with pytest.raises(mod.SafeToolError) as unsupported:
mod.changes_since({})
# A cursor stored for another project is never reused.
assert "after=head" in calls[0][1]
assert unsupported.value.code == "unsupported_server"
with pytest.raises(mod.SafeToolError) as invalid:
mod.changes_since({"after": "bogus"})
assert invalid.value.code == "invalid_changes_cursor"
reset = mod.changes_since({"reset": True})
assert "after=head" in calls[2][1] and reset["started_at_head"] is True
assert mod.read_changes_cursor("project-1") == "c1.2.2"
mod.REFRESH_TOKEN = None
mod.TOKEN_EXPIRES_AT = "2020-01-01T00:00:00Z"
with pytest.raises(mod.SafeToolError) as auth:
mod.changes_since({})
assert auth.value.code == "auth_required"
assert auth.value.details["auth_problem"]["reason"] == "access_token_expired_without_refresh"
assert len(calls) == 3
assert mod.read_changes_cursor("project-1") == "c1.2.2"
def test_changes_since_is_a_core_read_tool(mod):
assert "nap_changes_since" in mod.CORE_TOOL_NAMES
assert mod.tool_category("nap_changes_since") == "memory"
assert mod.tool_contract_metadata("nap_changes_since")["side_effect"] == "none"
# -- 7. Idempotency keys on remaining node writes --------------------------
def node_write_backend(mod, nodes):
stub_project_lock(mod)
mod.CONFIGURED_TOOL_PROFILES = {"all"}
mod.get_node_by_id = lambda args: nodes[args.get("node_id") or args.get("id")]
mod.index_node = lambda node: None
mod.remove_indexed_node = lambda node_id: None
writes = []
def fake_request(method, path, payload=None, **kwargs):
if method == "GET":
return {"items": list(nodes.values()) if "/nodes?" in path else [], "next_cursor": None}
writes.append((method, path, dict(kwargs.get("extra_headers") or {})))
node_id = path.rsplit("/", 1)[-1]
return {**nodes.get(node_id, next(iter(nodes.values()))), **(payload or {})}
mod.request_json = fake_request
return writes
def test_link_and_kanban_writes_send_scoped_idempotency_keys(mod):
card = kanban_card("first", "high", "00001024")
card["id"] = "card-1"
target = {"id": "target-1", "project_id": "project-1", "full_path": "/plans/t", "links": []}
writes = node_write_backend(mod, {"card-1": card, "target-1": target})
linked = mod.call_tool_impl("nap_ln", {"source_node_id": "card-1", "target_node_id": "target-1"})
assert writes[-1][0:2] == ("PATCH", "/v1/projects/project-1/nodes/card-1")
assert linked["idempotency_key"] == writes[-1][2]["Idempotency-Key"]
uuid.UUID(linked["idempotency_key"])
started = mod.call_tool_impl("nap_kanban_start", {"node_id": "card-1", "idempotency_key": "start-1"})
assert writes[-1][2]["Idempotency-Key"] == "start-1"
assert started["idempotency_key"] == "start-1"
tools = {tool["name"]: tool for tool in mod.all_tools()}
for name in ("nap_ln", "nap_kanban_start", "nap_kanban_update", "nap_agent_tee", "nap_mv_folder"):
assert "idempotency_key" in tools[name]["inputSchema"]["properties"], name
assert mod.tool_contract_metadata(name)["idempotency"] == "same-idempotency-key"
def test_multi_write_tools_derive_keys_and_surface_them_on_failure(mod):
nodes = {
"a": {"id": "a", "project_id": "project-1", "full_path": "/old/a", "folder_path": "/old", "name": "a"},
"b": {"id": "b", "project_id": "project-1", "full_path": "/old/b", "folder_path": "/old", "name": "b"},
}
writes = node_write_backend(mod, nodes)
moved = mod.call_tool_impl("nap_mv_folder", {"from_folder": "/old", "to_folder": "/new", "idempotency_key": "mvf"})
assert [w[2]["Idempotency-Key"] for w in writes] == ["mvf", "mvf:1"]
assert moved["idempotency_key"] == "mvf" and moved["idempotency_keys"] == ["mvf", "mvf:1"]
def ambiguous(method, path, payload=None, **kwargs):
raise mod.SafeToolError("request_timeout", "timed out")
mod.request_json = ambiguous
with pytest.raises(mod.SafeToolError) as caught:
mod.call_tool_impl("nap_create_node", {"folder_path": "/notes", "name": "x", "content_text": "x"})
uuid.UUID(caught.value.details["idempotency_key"])
assert mod.node_write_idempotency_headers("POST", "/v1/projects/p/nodes/n/restore") == {}
assert mod.node_write_idempotency_headers("DELETE", "/v1/projects/p/nodes/n") == {}
assert "Idempotency-Key" in mod.node_write_idempotency_headers("PATCH", "/v1/projects/p/nodes/n/patch")
Reference in New Issue View Git Blame Copy Permalink