4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
208 lines
6.3 KiB
Python
208 lines
6.3 KiB
Python
"""Tests for tools/TwilioNotifyTool — SMS notification surface."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from integrations.twilio.tools.twilio_notify_tool import TwilioNotifyTool, twilio_notify
|
|
|
|
|
|
@pytest.fixture
|
|
def twilio_source() -> dict[str, Any]:
|
|
"""The flat/runtime ``sources`` shape passed to is_available()."""
|
|
return {
|
|
"twilio": {
|
|
"account_sid": "AC1",
|
|
"auth_token": "tok",
|
|
"sms": {
|
|
"enabled": True,
|
|
"from_number": "+14155551111",
|
|
"messaging_service_sid": "",
|
|
"default_to": "+14155550000",
|
|
},
|
|
}
|
|
}
|
|
|
|
|
|
def _patch_resolve(monkeypatch: pytest.MonkeyPatch, twilio_config: dict[str, Any] | None) -> None:
|
|
"""Patch resolve_effective_integrations to return a wrapped twilio entry."""
|
|
effective: dict[str, Any] = {}
|
|
if twilio_config is not None:
|
|
effective["twilio"] = {"config": twilio_config, "source": "local env"}
|
|
monkeypatch.setattr(
|
|
"integrations.catalog.resolve_effective_integrations",
|
|
lambda: effective,
|
|
)
|
|
|
|
|
|
def test_metadata_declares_twilio_source() -> None:
|
|
metadata = TwilioNotifyTool.metadata()
|
|
assert metadata.name == "twilio_notify"
|
|
assert metadata.source == "twilio"
|
|
|
|
|
|
# ---- is_available -------------------------------------------------------------
|
|
|
|
|
|
def test_is_available_true_when_sms_configured(twilio_source: dict[str, Any]) -> None:
|
|
assert twilio_notify.is_available(twilio_source) is True
|
|
|
|
|
|
def test_is_available_false_when_no_twilio() -> None:
|
|
assert twilio_notify.is_available({}) is False
|
|
|
|
|
|
def test_is_available_false_when_sms_disabled(twilio_source: dict[str, Any]) -> None:
|
|
twilio_source["twilio"]["sms"]["enabled"] = False
|
|
assert twilio_notify.is_available(twilio_source) is False
|
|
|
|
|
|
def test_is_available_false_when_no_sender(twilio_source: dict[str, Any]) -> None:
|
|
twilio_source["twilio"]["sms"]["from_number"] = ""
|
|
twilio_source["twilio"]["sms"]["messaging_service_sid"] = ""
|
|
assert twilio_notify.is_available(twilio_source) is False
|
|
|
|
|
|
def test_is_available_true_with_only_messaging_service(twilio_source: dict[str, Any]) -> None:
|
|
twilio_source["twilio"]["sms"]["from_number"] = ""
|
|
twilio_source["twilio"]["sms"]["messaging_service_sid"] = "MG1"
|
|
assert twilio_notify.is_available(twilio_source) is True
|
|
|
|
|
|
# ---- credentials never travel through traced kwargs --------------------------
|
|
|
|
|
|
def test_extract_params_returns_no_credentials(twilio_source: dict[str, Any]) -> None:
|
|
"""extract_params output is serialized into traces — it must hold no secrets."""
|
|
params = twilio_notify.extract_params(twilio_source)
|
|
assert params == {}
|
|
|
|
|
|
# ---- run ----------------------------------------------------------------------
|
|
|
|
|
|
def test_run_resolves_credentials_internally_and_dispatches(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_resolve(
|
|
monkeypatch,
|
|
{
|
|
"account_sid": "AC1",
|
|
"auth_token": "tok",
|
|
"sms": {
|
|
"enabled": True,
|
|
"from_number": "+14155551111",
|
|
"messaging_service_sid": "",
|
|
"default_to": "+14155550000",
|
|
},
|
|
},
|
|
)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def _fake_send(report: str, ctx: dict[str, Any]) -> tuple[bool, str, str]:
|
|
captured["report"] = report
|
|
captured["ctx"] = ctx
|
|
return True, "", "SM-SENT"
|
|
|
|
monkeypatch.setattr(
|
|
"integrations.twilio.tools.twilio_notify_tool.send_twilio_sms_report", _fake_send
|
|
)
|
|
|
|
result = twilio_notify.run(body="page on-call", to="+14155559999")
|
|
|
|
assert result["status"] == "sent"
|
|
assert result["sid"] == "SM-SENT"
|
|
assert captured["report"] == "page on-call"
|
|
assert captured["ctx"]["to"] == "+14155559999"
|
|
assert captured["ctx"]["account_sid"] == "AC1"
|
|
assert captured["ctx"]["auth_token"] == "tok"
|
|
|
|
|
|
def test_run_falls_back_to_default_recipient(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_resolve(
|
|
monkeypatch,
|
|
{
|
|
"account_sid": "AC1",
|
|
"auth_token": "tok",
|
|
"sms": {
|
|
"enabled": True,
|
|
"from_number": "+14155551111",
|
|
"messaging_service_sid": "",
|
|
"default_to": "+14155550000",
|
|
},
|
|
},
|
|
)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def _fake_send(report: str, ctx: dict[str, Any]) -> tuple[bool, str, str]:
|
|
captured["ctx"] = ctx
|
|
return True, "", "SM-DEF"
|
|
|
|
monkeypatch.setattr(
|
|
"integrations.twilio.tools.twilio_notify_tool.send_twilio_sms_report", _fake_send
|
|
)
|
|
|
|
result = twilio_notify.run(body="hi")
|
|
|
|
assert result["status"] == "sent"
|
|
assert captured["ctx"]["to"] == "+14155550000"
|
|
|
|
|
|
def test_run_failed_when_twilio_not_configured(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_resolve(monkeypatch, None)
|
|
|
|
result = twilio_notify.run(body="hi", to="+14155550000")
|
|
|
|
assert result["status"] == "failed"
|
|
assert "not configured" in result["error"].lower()
|
|
assert result["sid"] == ""
|
|
|
|
|
|
def test_run_failed_when_no_recipient(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_resolve(
|
|
monkeypatch,
|
|
{
|
|
"account_sid": "AC1",
|
|
"auth_token": "tok",
|
|
"sms": {
|
|
"enabled": True,
|
|
"from_number": "+14155551111",
|
|
"messaging_service_sid": "",
|
|
"default_to": "",
|
|
},
|
|
},
|
|
)
|
|
|
|
result = twilio_notify.run(body="hi")
|
|
|
|
assert result["status"] == "failed"
|
|
assert "recipient" in result["error"].lower()
|
|
|
|
|
|
def test_run_propagates_send_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_patch_resolve(
|
|
monkeypatch,
|
|
{
|
|
"account_sid": "AC1",
|
|
"auth_token": "tok",
|
|
"sms": {
|
|
"enabled": True,
|
|
"from_number": "+14155551111",
|
|
"messaging_service_sid": "",
|
|
"default_to": "+14155550000",
|
|
},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
"integrations.twilio.tools.twilio_notify_tool.send_twilio_sms_report",
|
|
lambda _r, _c: (False, "twilio rejected", ""),
|
|
)
|
|
|
|
result = twilio_notify.run(body="hi", to="+14155550000")
|
|
|
|
assert result["status"] == "failed"
|
|
assert result["error"] == "twilio rejected"
|
|
assert result["sid"] == ""
|