diff --git a/README.rst b/README.rst index 4fcf34be..394db23c 100644 --- a/README.rst +++ b/README.rst @@ -215,6 +215,23 @@ You can install it with ``python3 -m pip install SpeechRecognition[groq]``. Please set the environment variable ``GROQ_API_KEY`` before calling ``recognizer_instance.recognize_groq``. +Proxy Support +~~~~~~~~~~~~~ + +All cloud-based recognizers support proxying via the ``recognizer_instance.proxy_url`` attribute: + +.. code:: python + + import speech_recognition as sr + r = sr.Recognizer() + r.proxy_url = "http://proxy.example.com:8080" # HTTP proxy + # r.proxy_url = "socks5://proxy.example.com:1080" # SOCKS5 proxy (requires PySocks) + # r.proxy_url = "" # explicitly disable proxies + +By default ``proxy_url`` is ``None``, which preserves existing behaviour (system/environment proxy settings are used). + +SOCKS proxy support requires the ``PySocks`` package: ``pip install PySocks``. + Troubleshooting --------------- diff --git a/reference/library-reference.rst b/reference/library-reference.rst index e96fde87..27008397 100644 --- a/reference/library-reference.rst +++ b/reference/library-reference.rst @@ -153,6 +153,24 @@ Represents the timeout (in seconds) for internal operations, such as API request Setting this to a reasonable value ensures that these operations will never block indefinitely, though good values depend on your network speed and the expected length of the audio to recognize. +``recognizer_instance.proxy_url = None # type: Union[str, None]`` +------------------------------------------------------------------ + +Configures an HTTP or SOCKS proxy for all cloud-based API requests. Can be changed. + +* ``None`` (default) -- use system/environment proxy settings (backward compatible). +* ``""`` (empty string) -- explicitly disable all proxies. +* ``"http://host:port"`` -- use an HTTP proxy. +* ``"socks5://host:port"`` -- use a SOCKS5 proxy (requires the ``PySocks`` package: ``pip install PySocks``). + +Example: + +.. code:: python + + import speech_recognition as sr + r = sr.Recognizer() + r.proxy_url = "http://proxy.example.com:8080" + ``recognizer_instance.record(source: AudioSource, duration: Union[float, None] = None, offset: Union[float, None] = None) -> AudioData`` ---------------------------------------------------------------------------------------------------------------------------------------- diff --git a/speech_recognition/__init__.py b/speech_recognition/__init__.py index 61b769fd..154ba644 100644 --- a/speech_recognition/__init__.py +++ b/speech_recognition/__init__.py @@ -24,7 +24,7 @@ from collections.abc import Iterable from urllib.error import HTTPError, URLError from urllib.parse import urlencode -from urllib.request import Request, urlopen +from urllib.request import Request from .audio import AudioData, get_flac_converter from .exceptions import ( @@ -34,6 +34,11 @@ UnknownValueError, WaitTimeoutError, ) +from .proxy import ( + build_boto3_proxy_config, + build_requests_proxies, + urlopen_with_proxy, +) __author__ = "Anthony Zhang (Uberi)" __version__ = "3.14.5" @@ -327,6 +332,7 @@ def __init__(self): self.dynamic_energy_ratio = 1.5 self.pause_threshold = 0.8 # seconds of non-speaking audio before a phrase is considered complete self.operation_timeout = None # seconds after an internal operation (e.g., an API request) starts before it times out, or ``None`` for no timeout + self.proxy_url = None # proxy URL for API requests: ``None`` uses system/env defaults, ``""`` disables proxies, or a URL like ``"http://host:port"`` or ``"socks5://host:port"`` self.phrase_threshold = 0.3 # minimum seconds of speaking audio before we consider the speaking audio a phrase - values below this are ignored (for filtering out clicks and pops) self.non_speaking_duration = 0.5 # seconds of non-speaking audio to keep on both sides of the recording @@ -625,7 +631,7 @@ def recognize_wit(self, audio_data, key, show_all=False): url = "https://api.wit.ai/speech?v=20170307" request = Request(url, data=wav_data, headers={"Authorization": "Bearer {}".format(key), "Content-Type": "audio/wav"}) try: - response = urlopen(request, timeout=self.operation_timeout) + response = urlopen_with_proxy(request, timeout=self.operation_timeout, proxy_url=self.proxy_url) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: @@ -680,7 +686,7 @@ def recognize_azure(self, audio_data, key, language="en-US", profanity="masked", start_time = monotonic() try: - credential_response = urlopen(credential_request, timeout=60) # credential response can take longer, use longer timeout instead of default one + credential_response = urlopen_with_proxy(credential_request, timeout=60, proxy_url=self.proxy_url) # credential response can take longer, use longer timeout instead of default one except HTTPError as e: raise RequestError("credential request failed: {}".format(e.reason)) except URLError as e: @@ -719,7 +725,7 @@ def recognize_azure(self, audio_data, key, language="en-US", profanity="masked", }) try: - response = urlopen(request, timeout=self.operation_timeout) + response = urlopen_with_proxy(request, timeout=self.operation_timeout, proxy_url=self.proxy_url) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: @@ -774,7 +780,7 @@ def recognize_bing(self, audio_data, key, language="en-US", show_all=False): start_time = monotonic() try: - credential_response = urlopen(credential_request, timeout=60) # credential response can take longer, use longer timeout instead of default one + credential_response = urlopen_with_proxy(credential_request, timeout=60, proxy_url=self.proxy_url) # credential response can take longer, use longer timeout instead of default one except HTTPError as e: raise RequestError("credential request failed: {}".format(e.reason)) except URLError as e: @@ -813,7 +819,7 @@ def recognize_bing(self, audio_data, key, language="en-US", show_all=False): }) try: - response = urlopen(request, timeout=self.operation_timeout) + response = urlopen_with_proxy(request, timeout=self.operation_timeout, proxy_url=self.proxy_url) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: @@ -849,7 +855,8 @@ def recognize_lex(self, audio_data, bot_name, bot_alias, user_id, content_type=" client = boto3.client('lex-runtime', aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, - region_name=region) + region_name=region, + config=build_boto3_proxy_config(self.proxy_url)) raw_data = audio_data.get_raw_data( convert_rate=16000, convert_width=2 @@ -899,7 +906,7 @@ def recognize_houndify(self, audio_data, client_id, client_key, show_all=False): "Hound-Client-Authentication": "{};{};{}".format(client_id, request_time, request_signature) }) try: - response = urlopen(request, timeout=self.operation_timeout) + response = urlopen_with_proxy(request, timeout=self.operation_timeout, proxy_url=self.proxy_url) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: @@ -944,13 +951,15 @@ def recognize_amazon(self, audio_data, bucket_name=None, access_key_id=None, sec 'transcribe', aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, - region_name=region) + region_name=region, + config=build_boto3_proxy_config(self.proxy_url)) s3 = boto3.client( 's3', aws_access_key_id=access_key_id, aws_secret_access_key=secret_access_key, - region_name=region) + region_name=region, + config=build_boto3_proxy_config(self.proxy_url)) session = boto3.Session( aws_access_key_id=access_key_id, @@ -1004,7 +1013,8 @@ def recognize_amazon(self, audio_data, bucket_name=None, access_key_id=None, sec transcript_uri = job['Transcript']['TranscriptFileUri'] import json import urllib.request - with urllib.request.urlopen(transcript_uri) as json_data: + transcript_request = urllib.request.Request(transcript_uri) + with urlopen_with_proxy(transcript_request, timeout=self.operation_timeout, proxy_url=self.proxy_url) as json_data: d = json.load(json_data) confidences = [] for item in d['results']['items']: @@ -1093,6 +1103,8 @@ def read_file(filename, chunk_size=5242880): import requests + proxies = build_requests_proxies(self.proxy_url) + check_existing = audio_data is None and job_name if check_existing: # Query status. @@ -1101,7 +1113,7 @@ def read_file(filename, chunk_size=5242880): headers = { "authorization": api_token, } - response = requests.get(endpoint, headers=headers) + response = requests.get(endpoint, headers=headers, proxies=proxies) data = response.json() status = data['status'] @@ -1128,7 +1140,8 @@ def read_file(filename, chunk_size=5242880): headers = {'authorization': api_token} response = requests.post('https://api.assemblyai.com/v2/upload', headers=headers, - data=read_file(audio_data)) + data=read_file(audio_data), + proxies=proxies) upload_url = response.json()['upload_url'] # Queue file for transcription. @@ -1138,7 +1151,7 @@ def read_file(filename, chunk_size=5242880): "authorization": api_token, "content-type": "application/json" } - response = requests.post(endpoint, json=json, headers=headers) + response = requests.post(endpoint, json=json, headers=headers, proxies=proxies) data = response.json() transciption_id = data['id'] exc = TranscriptionNotReady() @@ -1175,7 +1188,7 @@ def recognize_ibm(self, audio_data, key, language="en-US", show_all=False): authorization_value = base64.standard_b64encode("{}:{}".format(username, password).encode("utf-8")).decode("utf-8") request.add_header("Authorization", "Basic {}".format(authorization_value)) try: - response = urlopen(request, timeout=self.operation_timeout) + response = urlopen_with_proxy(request, timeout=self.operation_timeout, proxy_url=self.proxy_url) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: @@ -1312,7 +1325,7 @@ def recognize_api(self, audio_data, client_access_token, language="en", session_ if session_id is None: session_id = uuid.uuid4().hex data = b"--" + boundary.encode("utf-8") + b"\r\n" + b"Content-Disposition: form-data; name=\"request\"\r\n" + b"Content-Type: application/json\r\n" + b"\r\n" + b"{\"v\": \"20150910\", \"sessionId\": \"" + session_id.encode("utf-8") + b"\", \"lang\": \"" + language.encode("utf-8") + b"\"}\r\n" + b"--" + boundary.encode("utf-8") + b"\r\n" + b"Content-Disposition: form-data; name=\"voiceData\"; filename=\"audio.wav\"\r\n" + b"Content-Type: audio/wav\r\n" + b"\r\n" + wav_data + b"\r\n" + b"--" + boundary.encode("utf-8") + b"--\r\n" request = Request(url, data=data, headers={"Authorization": "Bearer {}".format(client_access_token), "Content-Length": str(len(data)), "Expect": "100-continue", "Content-Type": "multipart/form-data; boundary={}".format(boundary)}) - try: response = urlopen(request, timeout=10) + try: response = urlopen_with_proxy(request, timeout=10, proxy_url=getattr(self, 'proxy_url', None)) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: raise RequestError("recognition connection failed: {}".format(e.reason)) response_text = response.read().decode("utf-8") diff --git a/speech_recognition/proxy.py b/speech_recognition/proxy.py new file mode 100644 index 00000000..71cbd211 --- /dev/null +++ b/speech_recognition/proxy.py @@ -0,0 +1,173 @@ +"""Centralized proxy utilities for SpeechRecognition. + +Provides helper functions that build proxy-aware clients for each HTTP +library used in the project (urllib, httpx, requests, boto3, gRPC). + +Proxy URL semantics: + None - use system/env proxy settings (default, backward compatible) + "" - explicitly disable proxies + "http://host:port" - use that HTTP proxy + "socks5://host:port" - SOCKS proxy (requires PySocks) +""" + +from __future__ import annotations + +import contextlib +import os +from urllib.request import ( + OpenerDirector, + ProxyHandler, + Request, + build_opener, + urlopen, +) + +from speech_recognition.exceptions import SetupError + + +def build_urllib_opener(proxy_url: str | None) -> OpenerDirector | None: + """Return an ``OpenerDirector`` configured with *proxy_url*. + + Returns ``None`` when *proxy_url* is ``None`` (use default behaviour). + An empty string disables proxying; a ``socks5://`` URL requires + ``PySocks`` + ``sockshandler``. + """ + if proxy_url is None: + return None + + if proxy_url == "": + return build_opener(ProxyHandler({})) + + if proxy_url.startswith("socks"): + try: + from sockshandler import SocksiPyHandler + except ImportError: + raise SetupError( + "SOCKS proxy support requires the PySocks package. " + "Install it with: pip install PySocks" + ) + + from urllib.parse import urlparse + + parsed = urlparse(proxy_url) + import socks + + socks_type_map = { + "socks4": socks.SOCKS4, + "socks5": socks.SOCKS5, + "socks5h": socks.SOCKS5, + } + socks_type = socks_type_map.get(parsed.scheme) + if socks_type is None: + raise SetupError( + f"Unsupported SOCKS scheme: {parsed.scheme!r}. " + "Use socks4, socks5, or socks5h." + ) + + return build_opener( + SocksiPyHandler( + socks_type, + parsed.hostname, + parsed.port or 1080, + username=parsed.username, + password=parsed.password, + ) + ) + + # HTTP/HTTPS proxy + return build_opener( + ProxyHandler({"http": proxy_url, "https": proxy_url}) + ) + + +def urlopen_with_proxy(request: Request, timeout: int | None, proxy_url: str | None): + """Drop-in replacement for ``urlopen()`` that respects *proxy_url*.""" + opener = build_urllib_opener(proxy_url) + kwargs = {} + if timeout is not None: + kwargs["timeout"] = timeout + + if opener is not None: + return opener.open(request, **kwargs) + return urlopen(request, **kwargs) + + +def build_httpx_client(proxy_url: str | None): + """Return an ``httpx.Client`` configured with *proxy_url*. + + Returns ``None`` when *proxy_url* is ``None`` (caller should use + the SDK default client). + """ + if proxy_url is None: + return None + + import httpx + + if proxy_url == "": + return httpx.Client(proxy=None) + + return httpx.Client(proxy=proxy_url) + + +def build_requests_proxies(proxy_url: str | None) -> dict | None: + """Return a *proxies* dict suitable for ``requests.get/post(proxies=...)``. + + Returns ``None`` when *proxy_url* is ``None`` (use default behaviour). + """ + if proxy_url is None: + return None + + if proxy_url == "": + return {"http": None, "https": None} + + return {"http": proxy_url, "https": proxy_url} + + +def build_boto3_proxy_config(proxy_url: str | None): + """Return a ``botocore.config.Config`` with proxy settings. + + Returns ``None`` when *proxy_url* is ``None`` (use default behaviour). + """ + if proxy_url is None: + return None + + from botocore.config import Config + + if proxy_url == "": + return Config(proxies={}) + + return Config(proxies={"http": proxy_url, "https": proxy_url}) + + +@contextlib.contextmanager +def grpc_proxy_env(proxy_url: str | None): + """Context manager that temporarily sets gRPC-compatible env vars. + + gRPC reads ``http_proxy`` / ``https_proxy`` from the environment. + This sets them for the duration of the ``with`` block, then restores + the previous values. + + .. warning:: This is NOT thread-safe. + """ + if proxy_url is None: + yield + return + + env_keys = ("http_proxy", "https_proxy", + "HTTP_PROXY", "HTTPS_PROXY") + saved = {k: os.environ.get(k) for k in env_keys} + + try: + if proxy_url == "": + for k in env_keys: + os.environ.pop(k, None) + else: + for k in env_keys: + os.environ[k] = proxy_url + yield + finally: + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v diff --git a/speech_recognition/recognizers/google.py b/speech_recognition/recognizers/google.py index 21216f5c..6418f111 100644 --- a/speech_recognition/recognizers/google.py +++ b/speech_recognition/recognizers/google.py @@ -4,12 +4,13 @@ from typing import Dict, Literal, Optional, TypedDict from urllib.error import HTTPError, URLError from urllib.parse import urlencode -from urllib.request import Request, urlopen +from urllib.request import Request from typing_extensions import NotRequired from speech_recognition.audio import AudioData from speech_recognition.exceptions import RequestError, UnknownValueError +from speech_recognition.proxy import urlopen_with_proxy class Alternative(TypedDict): @@ -211,9 +212,9 @@ def find_best_hypothesis(alternatives: list[Alternative]) -> Alternative: return best_hypothesis -def obtain_transcription(request: Request, timeout: int) -> str: +def obtain_transcription(request: Request, timeout: int, proxy_url: str | None = None) -> str: try: - response = urlopen(request, timeout=timeout) + response = urlopen_with_proxy(request, timeout=timeout, proxy_url=proxy_url) except HTTPError as e: raise RequestError("recognition request failed: {}".format(e.reason)) except URLError as e: @@ -254,7 +255,8 @@ def recognize_legacy( request = request_builder.build(audio_data) response_text = obtain_transcription( - request, timeout=recognizer.operation_timeout + request, timeout=recognizer.operation_timeout, + proxy_url=getattr(recognizer, "proxy_url", None), ) output_parser = OutputParser( diff --git a/speech_recognition/recognizers/google_cloud.py b/speech_recognition/recognizers/google_cloud.py index adf4186f..37be56dd 100644 --- a/speech_recognition/recognizers/google_cloud.py +++ b/speech_recognition/recognizers/google_cloud.py @@ -102,33 +102,38 @@ def recognize( "missing google-cloud-speech module: ensure that google-cloud-speech is set up correctly." ) - client = ( - speech.SpeechClient.from_service_account_json(credentials_json_path) - if credentials_json_path - else speech.SpeechClient() - ) + proxy_url = getattr(recognizer, "proxy_url", None) - flac_data = audio_data.get_flac_data( - # audio sample rate must be between 8 kHz and 48 kHz inclusive - clamp sample rate into this range - convert_rate=( - None - if 8000 <= audio_data.sample_rate <= 48000 - else max(8000, min(audio_data.sample_rate, 48000)) - ), - convert_width=2, # audio samples must be 16-bit - ) - audio = speech.RecognitionAudio(content=flac_data) + from speech_recognition.proxy import grpc_proxy_env - config = _build_config(audio_data, kwargs.copy()) + with grpc_proxy_env(proxy_url): + client = ( + speech.SpeechClient.from_service_account_json(credentials_json_path) + if credentials_json_path + else speech.SpeechClient() + ) - try: - response = client.recognize(config=config, audio=audio) - except GoogleAPICallError as e: - raise RequestError(e) - except URLError as e: - raise RequestError( - "recognition connection failed: {0}".format(e.reason) + flac_data = audio_data.get_flac_data( + # audio sample rate must be between 8 kHz and 48 kHz inclusive - clamp sample rate into this range + convert_rate=( + None + if 8000 <= audio_data.sample_rate <= 48000 + else max(8000, min(audio_data.sample_rate, 48000)) + ), + convert_width=2, # audio samples must be 16-bit ) + audio = speech.RecognitionAudio(content=flac_data) + + config = _build_config(audio_data, kwargs.copy()) + + try: + response = client.recognize(config=config, audio=audio) + except GoogleAPICallError as e: + raise RequestError(e) + except URLError as e: + raise RequestError( + "recognition connection failed: {0}".format(e.reason) + ) if kwargs.get("show_all"): return response diff --git a/speech_recognition/recognizers/whisper_api/groq.py b/speech_recognition/recognizers/whisper_api/groq.py index 6bc9a7f7..730332d1 100644 --- a/speech_recognition/recognizers/whisper_api/groq.py +++ b/speech_recognition/recognizers/whisper_api/groq.py @@ -50,5 +50,14 @@ def recognize( "missing groq module: ensure that groq is set up correctly." ) - groq_recognizer = OpenAICompatibleRecognizer(groq.Groq()) + proxy_url = getattr(recognizer, "proxy_url", None) + client_kwargs = {} + if proxy_url is not None: + from speech_recognition.proxy import build_httpx_client + + http_client = build_httpx_client(proxy_url) + if http_client is not None: + client_kwargs["http_client"] = http_client + + groq_recognizer = OpenAICompatibleRecognizer(groq.Groq(**client_kwargs)) return groq_recognizer.recognize(audio_data, model, **kwargs) diff --git a/speech_recognition/recognizers/whisper_api/openai.py b/speech_recognition/recognizers/whisper_api/openai.py index 884c3136..0109fa7c 100644 --- a/speech_recognition/recognizers/whisper_api/openai.py +++ b/speech_recognition/recognizers/whisper_api/openai.py @@ -54,7 +54,16 @@ def recognize( "missing openai module: ensure that openai is set up correctly." ) - openai_recognizer = OpenAICompatibleRecognizer(openai.OpenAI()) + proxy_url = getattr(recognizer, "proxy_url", None) + client_kwargs = {} + if proxy_url is not None: + from speech_recognition.proxy import build_httpx_client + + http_client = build_httpx_client(proxy_url) + if http_client is not None: + client_kwargs["http_client"] = http_client + + openai_recognizer = OpenAICompatibleRecognizer(openai.OpenAI(**client_kwargs)) return openai_recognizer.recognize(audio_data, model, **kwargs) diff --git a/tests/test_proxy.py b/tests/test_proxy.py new file mode 100644 index 00000000..1e284fe1 --- /dev/null +++ b/tests/test_proxy.py @@ -0,0 +1,218 @@ +"""Unit tests for speech_recognition.proxy utilities.""" + +from __future__ import annotations + +import os +import unittest +from unittest.mock import MagicMock, patch +from urllib.request import OpenerDirector, Request + +from speech_recognition.exceptions import SetupError +from speech_recognition.proxy import ( + build_boto3_proxy_config, + build_httpx_client, + build_requests_proxies, + build_urllib_opener, + grpc_proxy_env, + urlopen_with_proxy, +) + + +class TestBuildUrllibOpener(unittest.TestCase): + def test_none_returns_none(self): + self.assertIsNone(build_urllib_opener(None)) + + def test_empty_string_returns_opener(self): + opener = build_urllib_opener("") + self.assertIsInstance(opener, OpenerDirector) + + def test_http_proxy_returns_opener(self): + opener = build_urllib_opener("http://proxy.example.com:8080") + self.assertIsInstance(opener, OpenerDirector) + + def test_socks_proxy_without_pysocks_raises_setup_error(self): + with patch.dict("sys.modules", {"sockshandler": None}): + with self.assertRaises((SetupError, ImportError)): + build_urllib_opener("socks5://proxy.example.com:1080") + + def test_socks_proxy_with_pysocks(self): + mock_handler = MagicMock() + mock_socks = MagicMock() + mock_socks.SOCKS5 = 2 + mock_socks.SOCKS4 = 1 + mock_sockshandler = MagicMock() + mock_sockshandler.SocksiPyHandler = mock_handler + + with patch.dict( + "sys.modules", + {"socks": mock_socks, "sockshandler": mock_sockshandler}, + ): + opener = build_urllib_opener("socks5://proxy.example.com:1080") + self.assertIsNotNone(opener) + mock_handler.assert_called_once_with( + 2, "proxy.example.com", 1080, username=None, password=None + ) + + +class TestUrlopenWithProxy(unittest.TestCase): + @patch("speech_recognition.proxy.urlopen") + def test_none_proxy_uses_default_urlopen(self, mock_urlopen): + req = Request("http://example.com") + urlopen_with_proxy(req, timeout=10, proxy_url=None) + mock_urlopen.assert_called_once_with(req, timeout=10) + + @patch("speech_recognition.proxy.urlopen") + def test_none_timeout(self, mock_urlopen): + req = Request("http://example.com") + urlopen_with_proxy(req, timeout=None, proxy_url=None) + mock_urlopen.assert_called_once_with(req) + + def test_http_proxy_uses_opener(self): + req = Request("http://example.com") + with patch("speech_recognition.proxy.build_urllib_opener") as mock_build: + mock_opener = MagicMock() + mock_build.return_value = mock_opener + urlopen_with_proxy( + req, timeout=5, proxy_url="http://proxy:8080" + ) + mock_opener.open.assert_called_once_with(req, timeout=5) + + +class TestBuildHttpxClient(unittest.TestCase): + def test_none_returns_none(self): + self.assertIsNone(build_httpx_client(None)) + + def test_empty_string_creates_client_no_proxy(self): + mock_httpx = MagicMock() + with patch.dict("sys.modules", {"httpx": mock_httpx}): + result = build_httpx_client("") + mock_httpx.Client.assert_called_once_with(proxy=None) + self.assertEqual(result, mock_httpx.Client.return_value) + + def test_proxy_url_creates_client(self): + mock_httpx = MagicMock() + with patch.dict("sys.modules", {"httpx": mock_httpx}): + result = build_httpx_client("http://proxy:8080") + mock_httpx.Client.assert_called_once_with( + proxy="http://proxy:8080" + ) + self.assertEqual(result, mock_httpx.Client.return_value) + + +class TestBuildRequestsProxies(unittest.TestCase): + def test_none_returns_none(self): + self.assertIsNone(build_requests_proxies(None)) + + def test_empty_string_disables_proxies(self): + result = build_requests_proxies("") + self.assertEqual(result, {"http": None, "https": None}) + + def test_proxy_url(self): + result = build_requests_proxies("http://proxy:8080") + self.assertEqual( + result, + {"http": "http://proxy:8080", "https": "http://proxy:8080"}, + ) + + +class TestBuildBoto3ProxyConfig(unittest.TestCase): + def test_none_returns_none(self): + self.assertIsNone(build_boto3_proxy_config(None)) + + def test_empty_string(self): + mock_config_cls = MagicMock() + mock_botocore_config = MagicMock() + mock_botocore_config.Config = mock_config_cls + + with patch.dict( + "sys.modules", + { + "botocore": MagicMock(), + "botocore.config": mock_botocore_config, + }, + ): + result = build_boto3_proxy_config("") + mock_config_cls.assert_called_once_with(proxies={}) + + def test_proxy_url(self): + mock_config_cls = MagicMock() + mock_botocore_config = MagicMock() + mock_botocore_config.Config = mock_config_cls + + with patch.dict( + "sys.modules", + { + "botocore": MagicMock(), + "botocore.config": mock_botocore_config, + }, + ): + result = build_boto3_proxy_config("http://proxy:8080") + mock_config_cls.assert_called_once_with( + proxies={ + "http": "http://proxy:8080", + "https": "http://proxy:8080", + } + ) + + +class TestGrpcProxyEnv(unittest.TestCase): + def test_none_is_noop(self): + original = os.environ.get("http_proxy") + with grpc_proxy_env(None): + self.assertEqual(os.environ.get("http_proxy"), original) + self.assertEqual(os.environ.get("http_proxy"), original) + + def test_sets_and_restores_env(self): + original_http = os.environ.get("http_proxy") + original_https = os.environ.get("https_proxy") + + with grpc_proxy_env("http://proxy:8080"): + self.assertEqual(os.environ["http_proxy"], "http://proxy:8080") + self.assertEqual(os.environ["https_proxy"], "http://proxy:8080") + self.assertEqual(os.environ["HTTP_PROXY"], "http://proxy:8080") + self.assertEqual(os.environ["HTTPS_PROXY"], "http://proxy:8080") + + # Should be restored + self.assertEqual(os.environ.get("http_proxy"), original_http) + self.assertEqual(os.environ.get("https_proxy"), original_https) + + def test_empty_string_clears_env(self): + os.environ["http_proxy"] = "old_value" + try: + with grpc_proxy_env(""): + self.assertNotIn("http_proxy", os.environ) + # Restored after context + self.assertEqual(os.environ["http_proxy"], "old_value") + finally: + os.environ.pop("http_proxy", None) + + def test_restores_on_exception(self): + original = os.environ.get("http_proxy") + + class TestError(Exception): + pass + + with self.assertRaises(TestError): + with grpc_proxy_env("http://proxy:8080"): + raise TestError("boom") + + self.assertEqual(os.environ.get("http_proxy"), original) + + +class TestRecognizerProxyUrl(unittest.TestCase): + def test_recognizer_has_proxy_url_attribute(self): + import speech_recognition as sr + + r = sr.Recognizer() + self.assertIsNone(r.proxy_url) + + def test_proxy_url_is_settable(self): + import speech_recognition as sr + + r = sr.Recognizer() + r.proxy_url = "http://proxy:8080" + self.assertEqual(r.proxy_url, "http://proxy:8080") + + +if __name__ == "__main__": + unittest.main()