test(sms): use clear=True in test_missing_phone_number_is_non_retryable

Prevents pre-existing TWILIO_PHONE_NUMBER or SMS_WEBHOOK_URL values in
the outer test environment from leaking into the assertion context.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
briandevans 2026-04-26 22:34:10 -07:00 committed by Teknium
parent 83080772f2
commit ce22301dc6
2 changed files with 53 additions and 7 deletions

View file

@ -10,7 +10,7 @@ Shares credentials with the optional telephony skill — same env vars:
Gateway-specific env vars: Gateway-specific env vars:
- SMS_WEBHOOK_PORT (default 8080) - SMS_WEBHOOK_PORT (default 8080)
- SMS_WEBHOOK_HOST (default 0.0.0.0) - SMS_WEBHOOK_HOST (default 127.0.0.1)
- SMS_WEBHOOK_URL (public URL for Twilio signature validation required) - SMS_WEBHOOK_URL (public URL for Twilio signature validation required)
- SMS_INSECURE_NO_SIGNATURE (true to disable signature validation dev only) - SMS_INSECURE_NO_SIGNATURE (true to disable signature validation dev only)
- SMS_ALLOWED_USERS (comma-separated E.164 phone numbers) - SMS_ALLOWED_USERS (comma-separated E.164 phone numbers)
@ -41,7 +41,7 @@ logger = logging.getLogger(__name__)
TWILIO_API_BASE = "https://api.twilio.com/2010-04-01/Accounts" TWILIO_API_BASE = "https://api.twilio.com/2010-04-01/Accounts"
MAX_SMS_LENGTH = 1600 # ~10 SMS segments MAX_SMS_LENGTH = 1600 # ~10 SMS segments
DEFAULT_WEBHOOK_PORT = 8080 DEFAULT_WEBHOOK_PORT = 8080
DEFAULT_WEBHOOK_HOST = "0.0.0.0" DEFAULT_WEBHOOK_HOST = "127.0.0.1"
def check_sms_requirements() -> bool: def check_sms_requirements() -> bool:
@ -91,19 +91,23 @@ class SmsAdapter(BasePlatformAdapter):
from aiohttp import web from aiohttp import web
if not self._from_number: if not self._from_number:
logger.error("[sms] TWILIO_PHONE_NUMBER not set — cannot send replies") msg = "[sms] TWILIO_PHONE_NUMBER not set — cannot send replies"
logger.error(msg)
self._set_fatal_error("sms_missing_phone_number", msg, retryable=False)
return False return False
insecure_no_sig = os.getenv("SMS_INSECURE_NO_SIGNATURE", "").lower() == "true" insecure_no_sig = os.getenv("SMS_INSECURE_NO_SIGNATURE", "").lower() == "true"
if not self._webhook_url and not insecure_no_sig: if not self._webhook_url and not insecure_no_sig:
logger.error( msg = (
"[sms] Refusing to start: SMS_WEBHOOK_URL is required for Twilio " "[sms] Refusing to start: SMS_WEBHOOK_URL is required for Twilio "
"signature validation. Set it to the public URL configured in your " "signature validation. Set it to the public URL configured in your "
"Twilio console (e.g. https://example.com/webhooks/twilio). " "Twilio console (e.g. https://example.com/webhooks/twilio). "
"For local development without validation, set " "For local development without validation, set "
"SMS_INSECURE_NO_SIGNATURE=true (NOT recommended for production).", "SMS_INSECURE_NO_SIGNATURE=true (NOT recommended for production)."
) )
logger.error(msg)
self._set_fatal_error("sms_missing_webhook_url", msg, retryable=False)
return False return False
if insecure_no_sig and not self._webhook_url: if insecure_no_sig and not self._webhook_url:

View file

@ -169,9 +169,9 @@ class TestSmsRequirements:
class TestWebhookHostConfig: class TestWebhookHostConfig:
"""Verify SMS_WEBHOOK_HOST env var and default.""" """Verify SMS_WEBHOOK_HOST env var and default."""
def test_default_host_is_all_interfaces(self): def test_default_host_is_localhost(self):
from gateway.platforms.sms import DEFAULT_WEBHOOK_HOST from gateway.platforms.sms import DEFAULT_WEBHOOK_HOST
assert DEFAULT_WEBHOOK_HOST == "0.0.0.0" assert DEFAULT_WEBHOOK_HOST == "127.0.0.1"
def test_host_from_env(self): def test_host_from_env(self):
from gateway.platforms.sms import SmsAdapter from gateway.platforms.sms import SmsAdapter
@ -242,6 +242,48 @@ class TestStartupGuard:
result = await adapter.connect() result = await adapter.connect()
assert result is False assert result is False
@pytest.mark.asyncio
async def test_missing_webhook_url_is_non_retryable(self):
adapter = self._make_adapter()
await adapter.connect()
assert adapter.has_fatal_error is True
assert adapter.fatal_error_retryable is False
assert "sms_missing_webhook_url" == adapter.fatal_error_code
@pytest.mark.asyncio
async def test_missing_phone_number_is_non_retryable(self):
from gateway.platforms.sms import SmsAdapter
env = {
"TWILIO_ACCOUNT_SID": "ACtest",
"TWILIO_AUTH_TOKEN": "tok",
"TWILIO_PHONE_NUMBER": "",
"SMS_WEBHOOK_URL": "",
}
with patch.dict(os.environ, env, clear=True):
pc = PlatformConfig(enabled=True, api_key="tok")
adapter = SmsAdapter(pc)
await adapter.connect()
assert adapter.has_fatal_error is True
assert adapter.fatal_error_retryable is False
assert adapter.fatal_error_code == "sms_missing_phone_number"
@pytest.mark.asyncio
async def test_insecure_flag_does_not_set_fatal_error(self):
mock_session = AsyncMock()
with patch.dict(os.environ, {"SMS_INSECURE_NO_SIGNATURE": "true"}), \
patch("aiohttp.web.AppRunner") as mock_runner_cls, \
patch("aiohttp.web.TCPSite") as mock_site_cls, \
patch("aiohttp.ClientSession", return_value=mock_session):
mock_runner_cls.return_value.setup = AsyncMock()
mock_runner_cls.return_value.cleanup = AsyncMock()
mock_site_cls.return_value.start = AsyncMock()
adapter = self._make_adapter()
result = await adapter.connect()
assert result is True
assert adapter.has_fatal_error is False
await adapter.disconnect()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_insecure_flag_allows_start_without_url(self): async def test_insecure_flag_allows_start_without_url(self):
mock_session = AsyncMock() mock_session = AsyncMock()