mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-03 22:32:27 +08:00
test(n8n): exercise latest DefaultRunner in daily regression (#2830)
* fix(n8n): validate runner timeout values * fix(n8n): validate runner timeout values * test(n8n): add latest DefaultRunner regression * test(n8n): add latest DefaultRunner regression * test(n8n): add latest DefaultRunner regression * test(n8n): fix latest Runner response assertion
This commit is contained in:
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"name": "LangBot n8n latest Runner smoke",
|
||||
"nodes": [
|
||||
{
|
||||
"parameters": {
|
||||
"httpMethod": "POST",
|
||||
"path": "langbot-n8n-smoke",
|
||||
"responseMode": "responseNode",
|
||||
"options": {}
|
||||
},
|
||||
"id": "langbot-n8n-webhook",
|
||||
"name": "Webhook",
|
||||
"type": "n8n-nodes-base.webhook",
|
||||
"typeVersion": 2,
|
||||
"position": [0, 0],
|
||||
"webhookId": "langbot-n8n-smoke"
|
||||
},
|
||||
{
|
||||
"parameters": {
|
||||
"respondWith": "json",
|
||||
"responseBody": "={{ { response: 'LANGBOT_N8N_DAILY_OK|' + $json.body.chatInput + '|' + $json.body.message + '|' + $json.body.user_id + '|' + $json.body.conversation_id + '|' + $json.body.session_id } }}",
|
||||
"options": {}
|
||||
},
|
||||
"id": "langbot-n8n-response",
|
||||
"name": "Respond to Webhook",
|
||||
"type": "n8n-nodes-base.respondToWebhook",
|
||||
"typeVersion": 1.1,
|
||||
"position": [300, 0]
|
||||
}
|
||||
],
|
||||
"connections": {
|
||||
"Webhook": {
|
||||
"main": [[{"node": "Respond to Webhook", "type": "main", "index": 0}]]
|
||||
}
|
||||
},
|
||||
"active": false,
|
||||
"settings": {"executionOrder": "v1"},
|
||||
"versionId": "langbot-n8n-latest-runner-smoke-v1",
|
||||
"meta": {"templateCredsSetupCompleted": true},
|
||||
"tags": []
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Live latest n8n regression for the LangBot n8n Agent Runner."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from langbot_plugin.api.entities.builtin.runner import (
|
||||
ActorContext,
|
||||
AgentEventContext,
|
||||
AgentInput,
|
||||
AgentResources,
|
||||
AgentRunState,
|
||||
AgentRuntimeContext,
|
||||
AgentTrigger,
|
||||
ConversationContext,
|
||||
DeliveryContext,
|
||||
RunnerContext,
|
||||
)
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
PLUGIN = ROOT / "n8n-agent"
|
||||
SENTINEL = "LANGBOT_N8N_DAILY_OK"
|
||||
|
||||
|
||||
def _load_runner():
|
||||
saved = {key: value for key, value in sys.modules.items() if key == "pkg" or key.startswith("pkg.")}
|
||||
for key in saved:
|
||||
del sys.modules[key]
|
||||
sys.path.insert(0, str(PLUGIN))
|
||||
try:
|
||||
spec = importlib.util.spec_from_file_location("live_n8n_agent_runner", PLUGIN / "components/runner/default.py")
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError("cannot load n8n Agent Runner")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module, saved
|
||||
except Exception:
|
||||
sys.path.remove(str(PLUGIN))
|
||||
for key in list(sys.modules):
|
||||
if key == "pkg" or key.startswith("pkg."):
|
||||
del sys.modules[key]
|
||||
sys.modules.update(saved)
|
||||
raise
|
||||
|
||||
|
||||
def _context(state: dict[str, str] | None = None) -> RunnerContext:
|
||||
return RunnerContext(
|
||||
run_id="n8n-latest-daily-regression",
|
||||
trigger=AgentTrigger(type="message.received"),
|
||||
event=AgentEventContext(
|
||||
event_id="n8n-latest-daily-regression",
|
||||
event_type="message.received",
|
||||
source="ci",
|
||||
),
|
||||
conversation=ConversationContext(conversation_id="daily-regression", session_id="daily-regression"),
|
||||
input=AgentInput(text=f"Reply with exactly {SENTINEL}."),
|
||||
delivery=DeliveryContext(surface="ci", supports_streaming=True),
|
||||
resources=AgentResources(),
|
||||
state=AgentRunState(conversation=state or {}),
|
||||
runtime=AgentRuntimeContext(),
|
||||
actor=ActorContext(actor_type="user", actor_id="daily-regression"),
|
||||
config={
|
||||
"webhook-url": os.environ.get("N8N_WEBHOOK_URL", "http://127.0.0.1:5678/webhook/langbot-n8n-smoke"),
|
||||
"auth-type": "none",
|
||||
"output-key": "response",
|
||||
"response-handling": "reply",
|
||||
"timeout": 30,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _kind(result) -> str:
|
||||
return getattr(result.type, "value", str(result.type))
|
||||
|
||||
|
||||
def _summary_path() -> Path:
|
||||
return Path(os.environ.get("N8N_LIVE_OUTPUT", ".qa/n8n-live/summary.json"))
|
||||
|
||||
|
||||
def _write_summary(payload: dict) -> None:
|
||||
path = _summary_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def _assert_run(results, *, expect_state_updates: bool) -> str:
|
||||
types = [_kind(result) for result in results]
|
||||
final_chunks = [
|
||||
result.data["chunk"]
|
||||
for result in results
|
||||
if _kind(result) == "message.delta" and result.data.get("chunk", {}).get("is_final")
|
||||
]
|
||||
if len(final_chunks) != 1:
|
||||
raise AssertionError(f"expected one final chunk, got {len(final_chunks)}: {types}")
|
||||
content = str(final_chunks[0].get("content") or "")
|
||||
if not content.startswith(f"{SENTINEL}|Reply with exactly {SENTINEL}."):
|
||||
raise AssertionError(f"unexpected n8n response: {content}")
|
||||
if "|user_daily-regression|n8n_conversation_" not in content or "|n8n_session_" not in content:
|
||||
raise AssertionError(f"Runner payload/state fields missing: {content}")
|
||||
if types.count("state.updated") != (2 if expect_state_updates else 0):
|
||||
raise AssertionError(f"unexpected state updates: {types}")
|
||||
if types[-1] != "run.completed":
|
||||
raise AssertionError(f"Runner did not complete: {types}")
|
||||
return content
|
||||
|
||||
|
||||
async def _run() -> None:
|
||||
module, saved_modules = _load_runner()
|
||||
try:
|
||||
runner = object.__new__(module.DefaultRunner)
|
||||
first = [result async for result in runner.run(_context())]
|
||||
first_content = _assert_run(first, expect_state_updates=True)
|
||||
state = {result.data["key"]: result.data["value"] for result in first if _kind(result) == "state.updated"}
|
||||
second = [result async for result in runner.run(_context(state))]
|
||||
second_content = _assert_run(second, expect_state_updates=False)
|
||||
if second_content != first_content:
|
||||
raise AssertionError("persisted n8n conversation/session state changed the response")
|
||||
_write_summary(
|
||||
{
|
||||
"passed": True,
|
||||
"runner": "DefaultRunner",
|
||||
"turns": 2,
|
||||
"state_keys": sorted(state),
|
||||
"final_content": first_content,
|
||||
}
|
||||
)
|
||||
print(f"n8n latest Agent Runner regression passed: state_keys={sorted(state)}")
|
||||
finally:
|
||||
sys.path.remove(str(PLUGIN))
|
||||
for key in list(sys.modules):
|
||||
if key == "pkg" or key.startswith("pkg."):
|
||||
del sys.modules[key]
|
||||
sys.modules.update(saved_modules)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
asyncio.run(_run())
|
||||
except Exception as exc:
|
||||
_write_summary({"passed": False, "error": str(exc)})
|
||||
print(f"n8n latest Agent Runner regression failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -6,6 +6,7 @@ Real n8n webhook integration supporting streaming and non-streaming responses.
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import math
|
||||
import typing
|
||||
import uuid
|
||||
|
||||
@@ -129,6 +130,16 @@ class DefaultRunner(Runner):
|
||||
if auth_type == "basic" and basic_encoding not in ("utf-8", "latin1"):
|
||||
raise N8nConfigError("basic-encoding must be utf-8 or latin1")
|
||||
|
||||
timeout_value = config.get("timeout", 120)
|
||||
if timeout_value is None or isinstance(timeout_value, bool):
|
||||
raise N8nConfigError("timeout must be a positive finite number")
|
||||
try:
|
||||
timeout = float(timeout_value)
|
||||
except (TypeError, ValueError):
|
||||
raise N8nConfigError("timeout must be a positive finite number") from None
|
||||
if not math.isfinite(timeout) or timeout <= 0:
|
||||
raise N8nConfigError("timeout must be a positive finite number")
|
||||
|
||||
return {
|
||||
"webhook_url": webhook_url,
|
||||
"auth_type": auth_type,
|
||||
@@ -141,7 +152,7 @@ class DefaultRunner(Runner):
|
||||
"header_name": config.get("header-name", ""),
|
||||
"header_value": config.get("header-value", ""),
|
||||
},
|
||||
"timeout": float(config.get("timeout", 120)),
|
||||
"timeout": timeout,
|
||||
"output_key": config.get("output-key", "response"),
|
||||
"response_handling": response_handling,
|
||||
"langbot_assets_enabled": _to_bool(config.get("langbot-assets-enabled"), False),
|
||||
|
||||
@@ -303,6 +303,16 @@ async def test_timeout_includes_waiting_for_response_headers(runner_module, monk
|
||||
assert results[0]["data"]["code"] == "n8n.timeout"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("value", [None, True, 0, -1, "invalid", float("inf"), float("nan")])
|
||||
async def test_invalid_timeout_fails_before_http(runner_module, monkeypatch, value):
|
||||
requests = wire(monkeypatch, Body([b'{"response":"ok"}']))
|
||||
results = await collect(runner_module, context({"timeout": value}))
|
||||
assert [result["type"] for result in results] == ["run.failed"]
|
||||
assert results[0]["data"]["code"] == "n8n.config_invalid"
|
||||
assert requests == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reply_timeout_is_total_not_only_per_chunk(runner_module, monkeypatch):
|
||||
class SlowBody(Body):
|
||||
|
||||
Reference in New Issue
Block a user