490e460c4b
The base_url parameter previously forced wss:// and https:// schemes, breaking air-gapped or private deployments that need ws:// or http://. Extract URL derivation into _derive_deepgram_urls() helper that respects the developers scheme choice while deriving the paired WebSocket and HTTP URLs the Deepgram SDK requires. Closes #4019
52 lines
1.9 KiB
Python
52 lines
1.9 KiB
Python
#
|
|
# Copyright (c) 2024-2026, Daily
|
|
#
|
|
# SPDX-License-Identifier: BSD 2-Clause License
|
|
#
|
|
|
|
import io
|
|
|
|
import pytest
|
|
from loguru import logger
|
|
|
|
from pipecat.services.deepgram.stt import _derive_deepgram_urls
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"base_url, expected_ws, expected_http",
|
|
[
|
|
# Secure schemes
|
|
("wss://mydeepgram.com", "wss://mydeepgram.com", "https://mydeepgram.com"),
|
|
("https://mydeepgram.com", "wss://mydeepgram.com", "https://mydeepgram.com"),
|
|
# Insecure schemes (air-gapped deployments)
|
|
("ws://mydeepgram.com", "ws://mydeepgram.com", "http://mydeepgram.com"),
|
|
("http://mydeepgram.com", "ws://mydeepgram.com", "http://mydeepgram.com"),
|
|
# Bare hostname defaults to secure
|
|
("mydeepgram.com", "wss://mydeepgram.com", "https://mydeepgram.com"),
|
|
# With port
|
|
("ws://localhost:8080", "ws://localhost:8080", "http://localhost:8080"),
|
|
("wss://localhost:443", "wss://localhost:443", "https://localhost:443"),
|
|
("localhost:8080", "wss://localhost:8080", "https://localhost:8080"),
|
|
# With path
|
|
("wss://host/v1/listen", "wss://host/v1/listen", "https://host/v1/listen"),
|
|
("http://host/v1/listen", "ws://host/v1/listen", "http://host/v1/listen"),
|
|
],
|
|
)
|
|
def test_derive_deepgram_urls(base_url, expected_ws, expected_http):
|
|
ws_url, http_url = _derive_deepgram_urls(base_url)
|
|
assert ws_url == expected_ws
|
|
assert http_url == expected_http
|
|
|
|
|
|
def test_derive_deepgram_urls_unknown_scheme_warns():
|
|
sink = io.StringIO()
|
|
handler_id = logger.add(sink, format="{message}")
|
|
try:
|
|
ws_url, http_url = _derive_deepgram_urls("ftp://mydeepgram.com")
|
|
# Falls back to secure
|
|
assert ws_url == "wss://mydeepgram.com"
|
|
assert http_url == "https://mydeepgram.com"
|
|
assert "Unrecognized scheme" in sink.getvalue()
|
|
finally:
|
|
logger.remove(handler_id)
|