Stéphane Tailland
105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
"""What the adapter must do regardless of any workflow.
|
|
|
|
The adapter owns the protocol and nothing else, so what is worth testing is the
|
|
protocol edges: a token that is absent or oddly cased, a workflow that answers
|
|
off-contract, and the escalation flag that must not be silently swallowed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from a2a_n8n_adapter.config import Settings
|
|
from a2a_n8n_adapter.server import Outcome, _read, build_agent_card
|
|
|
|
|
|
def settings(**over) -> Settings:
|
|
base = dict(
|
|
agent_name="supply",
|
|
agent_description="Supply decisions.",
|
|
skill_id="check_availability",
|
|
skill_name="Check availability",
|
|
skill_description="Can this be promised?",
|
|
skill_examples="Can we promise S10_1949 by 15 March?\nIs S18_3782 running out?",
|
|
webhook_url="http://n8n/webhook/x",
|
|
)
|
|
return Settings(**{**base, **over})
|
|
|
|
|
|
def response(payload, status=200) -> httpx.Response:
|
|
if isinstance(payload, str):
|
|
return httpx.Response(status, text=payload)
|
|
return httpx.Response(status, json=payload)
|
|
|
|
|
|
class TestReadingTheWorkflow:
|
|
def test_the_contracted_shape(self) -> None:
|
|
assert _read(response({"answer": "yes", "escalate": False})) == ("yes", Outcome.ANSWERED)
|
|
|
|
def test_n8n_wraps_its_last_node_in_an_array(self) -> None:
|
|
assert _read(response([{"answer": "yes"}])) == ("yes", Outcome.ANSWERED)
|
|
|
|
def test_escalation_is_carried_not_swallowed(self) -> None:
|
|
# An autonomous agent meeting a rule that needs approval must escalate.
|
|
# Losing this flag would turn "a human must look" into "done".
|
|
assert _read(response({"answer": "needs approval", "escalate": True})) == (
|
|
"needs approval",
|
|
Outcome.ESCALATED,
|
|
)
|
|
|
|
def test_an_off_contract_reply_is_returned_raw(self) -> None:
|
|
# Visible rather than reinterpreted: a workflow that has drifted should
|
|
# look wrong, not be quietly guessed at.
|
|
# Off contract is not agreement. Returned raw so the drift is visible,
|
|
# but never as a completed answer.
|
|
text, outcome = _read(response("plain text"))
|
|
assert text == "plain text" and outcome is Outcome.OFF_CONTRACT
|
|
|
|
def test_an_empty_reply_says_so(self) -> None:
|
|
# HTTP 200 with an empty body is what n8n returns when a node throws
|
|
# before the reply is built — a refused token exchange, usually. Reading
|
|
# it as a completed answer told the caller the task succeeded, which was
|
|
# the one thing certainly false.
|
|
text, outcome = _read(response(""))
|
|
assert outcome is Outcome.SILENT
|
|
assert "no answer" in text
|
|
|
|
|
|
class TestAgentCard:
|
|
def test_it_advertises_exactly_one_skill(self) -> None:
|
|
# One skill per adapter instance. A second one is a card change, which
|
|
# is the moment to ask whether the agent has become a data proxy.
|
|
card = build_agent_card(settings())
|
|
assert len(card.skills) == 1
|
|
assert card.skills[0].id == "check_availability"
|
|
|
|
def test_examples_reach_the_card(self) -> None:
|
|
# A calling agent's model reads these to decide whether to ask at all.
|
|
card = build_agent_card(settings())
|
|
assert len(card.skills[0].examples) == 2
|
|
|
|
def test_the_card_advertises_the_public_url_not_the_bind_address(self) -> None:
|
|
card = build_agent_card(settings(public_url="http://supply-agent.agents:8080/"))
|
|
assert card.supported_interfaces[0].url == "http://supply-agent.agents:8080/"
|
|
|
|
|
|
class TestSubjectToken:
|
|
@pytest.mark.parametrize("raw", ["Bearer abc", "bearer abc", "BEARER abc", "abc"])
|
|
def test_the_prefix_is_optional_and_case_insensitive(self, raw: str) -> None:
|
|
from a2a_n8n_adapter.server import _subject_token
|
|
|
|
class Ctx:
|
|
class call_context: # noqa: N801
|
|
state = {"headers": {"Authorization": raw}}
|
|
|
|
assert _subject_token(Ctx()) == "abc"
|
|
|
|
def test_an_absent_token_is_none_rather_than_an_empty_string(self) -> None:
|
|
from a2a_n8n_adapter.server import _subject_token
|
|
|
|
class Ctx:
|
|
call_context = None
|
|
|
|
assert _subject_token(Ctx()) is None
|