diff --git a/.github/workflows/unit_test.yml b/.github/workflows/unit_test.yml index 9beeb4e9..d6c14d06 100644 --- a/.github/workflows/unit_test.yml +++ b/.github/workflows/unit_test.yml @@ -15,8 +15,19 @@ jobs: runs-on: ${{ matrix.os }} strategy: matrix: - os: [ ubuntu-latest ] - python-version: [ '3.9', '3.10', '3.11', '3.12', '3.13' ] + # ubuntu-latest (26.04) ships no CPython 3.9 build, so pin the + # still-supported 3.9 leg to 24.04 and keep the rest on latest. + include: + - os: ubuntu-24.04 + python-version: '3.9' + - os: ubuntu-latest + python-version: '3.10' + - os: ubuntu-latest + python-version: '3.11' + - os: ubuntu-latest + python-version: '3.12' + - os: ubuntu-latest + python-version: '3.13' steps: - name: Checkout repository @@ -68,8 +79,19 @@ jobs: runs-on: ${{ matrix.os }} strategy: matrix: - os: [ ubuntu-latest ] - python-version: [ '3.9', '3.10', '3.11', '3.12', '3.13' ] + # ubuntu-latest (26.04) ships no CPython 3.9 build, so pin the + # still-supported 3.9 leg to 24.04 and keep the rest on latest. + include: + - os: ubuntu-24.04 + python-version: '3.9' + - os: ubuntu-latest + python-version: '3.10' + - os: ubuntu-latest + python-version: '3.11' + - os: ubuntu-latest + python-version: '3.12' + - os: ubuntu-latest + python-version: '3.13' steps: - name: Checkout repository diff --git a/dashscope/_logging.py b/dashscope/_logging.py index 1ecfc2e8..f91514df 100644 --- a/dashscope/_logging.py +++ b/dashscope/_logging.py @@ -41,6 +41,29 @@ REDACTED = "[REDACTED]" +def request_id_from_headers( + headers: Optional[Mapping[str, Any]], +) -> Optional[str]: + """Find a gateway request ID without assuming header casing.""" + names = {"x-request-id", "x-dashscope-requestid", "x-dashscope-request-id"} + for name, value in (headers or {}).items(): + if not isinstance(name, str) or name.lower() not in names: + continue + values = value if isinstance(value, (list, tuple)) else (value,) + for item in values: + if isinstance(item, str) and item.strip(): + return item + return None + + +def request_id_for_log(*request_ids: Any) -> str: + """Use real ID strings only; never stringify arbitrary response objects.""" + for request_id in request_ids: + if isinstance(request_id, str) and request_id.strip(): + return request_id + return "unknown" + + def redact_headers( headers: Optional[Mapping[str, Any]], ) -> Optional[dict]: diff --git a/dashscope/errors.py b/dashscope/errors.py index 005f53b8..ea974471 100644 --- a/dashscope/errors.py +++ b/dashscope/errors.py @@ -40,8 +40,15 @@ from dashscope import _error_registry as _registry from dashscope._error_registry import PublicError, SdkErrorDef +from dashscope._logging import ( + get_logger, + request_id_for_log, + request_id_from_headers, +) from dashscope._shared._errors import DashScopeException +logger = get_logger("response") + class ErrorCode(str, Enum): """Server-side error codes returned in ``response.code``.""" @@ -369,7 +376,7 @@ def _request_id( body: Dict[str, Any], headers: Mapping[str, str], ) -> Optional[str]: - return body.get("request_id") or headers.get("x-request-id") + return body.get("request_id") or request_id_from_headers(headers) def _catalog_message( @@ -540,13 +547,20 @@ def internal_response_error( """Build the catalog ``internal_error`` for an unusable response body.""" entry = _entry("internal_error") data = body if isinstance(body, dict) else {} + request_id = _request_id(data, response.headers) + logger.error( + "Failed to parse HTTP response, status=%s, request_id=%s", + response.status_code, + request_id_for_log( + request_id, + request_id_from_headers(response.headers), + ), + ) return InternalServerError( entry.error_msg, code=entry.error_code, status_code=entry.status_code, - request_id=( - data.get("request_id") or response.headers.get("x-request-id") - ), + request_id=request_id, body=body, headers=response.headers, ) @@ -609,5 +623,15 @@ def check_response( raise_on_error: bool = True, ) -> None: error = response_error(response) - if error is not None and raise_on_error: - raise error + if error is not None: + logger.error( + "HTTP service error: status=%s, code=%s, request_id=%s", + response.status_code, + error.code, + request_id_for_log( + error.request_id, + request_id_from_headers(response.headers), + ), + ) + if raise_on_error: + raise error diff --git a/dashscope/resources/_oss_upload.py b/dashscope/resources/_oss_upload.py index e1e92d88..831d416f 100644 --- a/dashscope/resources/_oss_upload.py +++ b/dashscope/resources/_oss_upload.py @@ -25,7 +25,7 @@ import httpx -from dashscope._logging import get_logger +from dashscope._logging import get_logger, request_id_from_headers from dashscope.errors import ( check_response, error_from_status, @@ -182,9 +182,12 @@ def upload_local_file( ) if not response.is_success: logger.error( - "uploading file %s to oss failed, status %s", + "uploading file %s to oss failed, status %s, request_id=%s", file_path, response.status_code, + request_id_from_headers(response.headers) + or response.headers.get("x-oss-request-id") + or "unknown", ) # OSS answers with XML, so there is no business code to read and the # error is mapped from the status code alone. diff --git a/dashscope/resources/_sse.py b/dashscope/resources/_sse.py index 2d4f1de3..ecce9025 100644 --- a/dashscope/resources/_sse.py +++ b/dashscope/resources/_sse.py @@ -11,8 +11,13 @@ from __future__ import annotations import json -from typing import Any, AsyncIterator, Iterator, Optional +from typing import Any, AsyncIterator, Iterator, Mapping, Optional +from dashscope._logging import ( + get_logger, + request_id_for_log, + request_id_from_headers, +) from dashscope.errors import ( DashScopeError, error_from_business_code, @@ -29,7 +34,30 @@ def _try_json(data: str) -> Any: return data -def _stream_error(status_code: Optional[int], data: str) -> DashScopeError: +def _log_stream_error( + data: str, + headers: Optional[Mapping[str, str]], +) -> None: + if headers is None: + return + body = _try_json(data) + body = body if isinstance(body, dict) else {} + code = body.get("code") + get_logger("transport.sse").error( + "SSE business error: code=%s, request_id=%s", + code if isinstance(code, str) else "unknown", + request_id_for_log( + body.get("request_id"), + request_id_from_headers(headers), + ), + ) + + +def _stream_error( + status_code: Optional[int], + data: str, + headers: Optional[Mapping[str, str]] = None, +) -> DashScopeError: """Build the error for one ``event:error`` frame. ``status:`` is optional on the wire. Absent, the frame is a business @@ -41,38 +69,56 @@ def _stream_error(status_code: Optional[int], data: str) -> DashScopeError: if isinstance(body, dict): message = body.get("message") or data if status_code is None: - return error_from_business_code(message=message, body=body) - return error_from_status(status_code, message=message, body=body) + error = error_from_business_code(message=message, body=body) + else: + error = error_from_status(status_code, message=message, body=body) + # Preserve the historical body-only ID contract on job-event exceptions. + _log_stream_error(data, headers) + return error def _iter_sse(response) -> Iterator[Any]: is_error = False status_code: Optional[int] = None - for line in response.iter_lines(): - if not line: - continue - if line.startswith("event:error"): - is_error = True - elif line.startswith("status:"): - status_code = int(line[len("status:") :].strip()) - elif line.startswith("data:"): - yield is_error, status_code, line[len("data:") :].strip() - if is_error: - break + try: + for line in response.iter_lines(): + if not line: + continue + if line.startswith("event:error"): + is_error = True + elif line.startswith("status:"): + status_code = int(line[len("status:") :].strip()) + elif line.startswith("data:"): + yield is_error, status_code, line[len("data:") :].strip() + if is_error: + break + except Exception: + get_logger("transport.sse").error( + "Failed to read or parse job SSE stream, request_id=%s", + request_id_for_log(request_id_from_headers(response.headers)), + ) + raise async def _aiter_sse(response) -> AsyncIterator[Any]: # Async twin of ``_iter_sse``; the framing rules are identical. is_error = False status_code: Optional[int] = None - async for line in response.aiter_lines(): - if not line: - continue - if line.startswith("event:error"): - is_error = True - elif line.startswith("status:"): - status_code = int(line[len("status:") :].strip()) - elif line.startswith("data:"): - yield is_error, status_code, line[len("data:") :].strip() - if is_error: - break + try: + async for line in response.aiter_lines(): + if not line: + continue + if line.startswith("event:error"): + is_error = True + elif line.startswith("status:"): + status_code = int(line[len("status:") :].strip()) + elif line.startswith("data:"): + yield is_error, status_code, line[len("data:") :].strip() + if is_error: + break + except Exception: + get_logger("transport.sse").error( + "Failed to read or parse job SSE stream, request_id=%s", + request_id_for_log(request_id_from_headers(response.headers)), + ) + raise diff --git a/dashscope/resources/audio/http_tts.py b/dashscope/resources/audio/http_tts.py index 002439fe..16003d49 100644 --- a/dashscope/resources/audio/http_tts.py +++ b/dashscope/resources/audio/http_tts.py @@ -30,6 +30,7 @@ Tuple, ) +from dashscope._logging import get_logger, request_id_from_headers from dashscope.errors import ( APIStatusError, DashScopeError, @@ -47,12 +48,47 @@ ] MODULE = "audio" +_logger = get_logger("audio.http_tts") _SSE_HEADER = {"X-DashScope-SSE": "enable"} _DEFAULT_PATH = "/services/audio/tts/SpeechSynthesizer" _QWEN_PATH = "/services/aigc/multimodal-generation/generation" +def _audio_body_request_id(body: Any) -> Optional[str]: + if not isinstance(body, dict): + return None + output = body.get("output") + for source in (body, output): + if isinstance(source, dict): + value = source.get("request_id") + if isinstance(value, str) and value.strip(): + return value + return None + + +def _log_audio_error( + error: Exception, + request_id: Optional[str] = None, + headers: Optional[Mapping[str, str]] = None, +) -> None: + # Exception text may contain raw SSE/audio or credentials. Log metadata only. + request_id = ( + _audio_body_request_id(getattr(error, "body", None)) + or request_id + or request_id_from_headers(headers) + or request_id_from_headers(getattr(error, "headers", None)) + or getattr(error, "request_id", None) + ) + if not isinstance(request_id, str) or not request_id.strip(): + request_id = None + _logger.error( + "HTTP TTS request failed: %s, request_id=%s", + type(error).__name__, + request_id or "unknown", + ) + + _Frame = Tuple[bool, Optional[int], str] @@ -123,7 +159,7 @@ def _with_audio_request_id(error: DashScopeError) -> DashScopeError: error.request_id = ( body.get("request_id") or output.get("request_id") - or error.headers.get("x-request-id") + or request_id_from_headers(error.headers) or error.request_id ) return error @@ -281,9 +317,13 @@ def _chunk_from_data( except ValueError as error: raise DashScopeError( f"invalid synthesis stream data: {data!r}", + request_id=header_request_id or None, ) from error if not isinstance(body, dict): - raise DashScopeError(f"invalid synthesis stream data: {data!r}") + raise DashScopeError( + f"invalid synthesis stream data: {data!r}", + request_id=header_request_id or None, + ) output = body.get("output") output = output if isinstance(output, dict) else {} payload = _audio_dict(output).get("data") @@ -291,9 +331,9 @@ def _chunk_from_data( **{ **body, "output": output, - "audio": base64.b64decode(payload, validate=True) - if payload - else None, + "audio": ( + base64.b64decode(payload, validate=True) if payload else None + ), "usage": _usage_dict(body), "request_id": body.get("request_id") or header_request_id, }, @@ -303,6 +343,47 @@ def _chunk_from_data( return chunk +def _parse_stream_chunk( + data: str, + header_request_id: str, + request_id: Optional[str], +) -> HttpSynthesisChunk: + chunk = _chunk_from_data(data, header_request_id) + if chunk.is_error: + _logger.error( + "HTTP TTS business error, request_id=%s", + request_id or "unknown", + ) + return chunk + + +def _synthesis_from_response( + response, + *, + raise_on_error: bool, +) -> HttpSynthesis: + request_id = request_id_from_headers(response.headers) + try: + body = response_body(response, raise_on_error=raise_on_error) + request_id = _audio_body_request_id(body) or request_id + result = with_headers( + _synthesis_from_body( + body, + response.headers.get("x-request-id", ""), + ), + response, + ) + if result.is_error: + _logger.error( + "HTTP TTS business error, request_id=%s", + request_id or "unknown", + ) + return result + except Exception as error: + _log_audio_error(error, request_id, response.headers) + raise + + class HttpSynthesisStream: """Typed iterator over streaming synthesis parts, mirroring AppStream.""" @@ -387,15 +468,9 @@ async def synthesize(self, text: str, **kwargs) -> HttpSynthesis: api_key=kwargs.pop("api_key", None), workspace=kwargs.pop("workspace", None), ) - return with_headers( - _synthesis_from_body( - response_body( - response, - raise_on_error=self._client.raise_on_error, - ), - response.headers.get("x-request-id", ""), - ), + return _synthesis_from_response( response, + raise_on_error=self._client.raise_on_error, ) async def stream(self, text: str, **kwargs) -> AsyncHttpSynthesisStream: @@ -403,34 +478,40 @@ async def stream(self, text: str, **kwargs) -> AsyncHttpSynthesisStream: voice = kwargs.pop("voice") headers = dict(_SSE_HEADER) headers.update(kwargs.pop("headers", None) or {}) - response = await self._client.transport.open_stream( - "POST", - _path_for(model), - json=_build_body( - text, - model=model, - voice=voice, - **{ - key: kwargs.pop(key, None) - for key in ( - "format", - "sample_rate", - "language_type", - "parameters", - "extra_input", - ) - }, - ), - headers=headers, - timeout=kwargs.pop("timeout", None), - sdk_module=MODULE, - api_key=kwargs.pop("api_key", None), - workspace=kwargs.pop("workspace", None), - ) + try: + response = await self._client.transport.open_stream( + "POST", + _path_for(model), + json=_build_body( + text, + model=model, + voice=voice, + **{ + key: kwargs.pop(key, None) + for key in ( + "format", + "sample_rate", + "language_type", + "parameters", + "extra_input", + ) + }, + ), + headers=headers, + timeout=kwargs.pop("timeout", None), + sdk_module=MODULE, + api_key=kwargs.pop("api_key", None), + workspace=kwargs.pop("workspace", None), + ) + except APIStatusError as error: + _with_audio_request_id(error) + _log_audio_error(error) + raise async def events(): decoder = _SSEDecoder() header_id = response.headers.get("x-request-id", "") + request_id = request_id_from_headers(response.headers) try: media_type = response.headers.get("content-type", "") media_type = media_type.split(";", 1)[0].strip().lower() @@ -448,23 +529,34 @@ async def events(): if frame is None: continue is_error, status, data = frame + request_id = ( + _audio_body_request_id(_try_json(data)) or request_id + ) if is_error: raise _audio_stream_error( status, data, response.headers, ) - yield _chunk_from_data(data, header_id) + yield _parse_stream_chunk(data, header_id, request_id) frame = decoder.feed("") if frame is not None: is_error, status, data = frame + request_id = ( + _audio_body_request_id(_try_json(data)) or request_id + ) if is_error: raise _audio_stream_error( status, data, response.headers, ) - yield _chunk_from_data(data, header_id) + yield _parse_stream_chunk(data, header_id, request_id) + except Exception as error: + if isinstance(error, DashScopeError): + _with_audio_request_id(error) + _log_audio_error(error, request_id, response.headers) + raise finally: await response.aclose() @@ -513,15 +605,9 @@ def synthesize( api_key=api_key, workspace=workspace, ) - return with_headers( - _synthesis_from_body( - response_body( - response, - raise_on_error=self._client.raise_on_error, - ), - response.headers.get("x-request-id", ""), - ), + return _synthesis_from_response( response, + raise_on_error=self._client.raise_on_error, ) def stream( @@ -573,15 +659,29 @@ def stream( except APIStatusError as error: # >=400 is raised (and closed) before open_stream returns a response. _with_audio_request_id(error) + _log_audio_error(error) raise return HttpSynthesisStream(self._events(response), response) def _events(self, response) -> Generator[HttpSynthesisChunk, None, None]: header_request_id = response.headers.get("x-request-id", "") + request_id = request_id_from_headers(response.headers) try: for is_error, status, data in _iter_audio_sse(response): + request_id = ( + _audio_body_request_id(_try_json(data)) or request_id + ) if is_error: raise _audio_stream_error(status, data, response.headers) - yield _chunk_from_data(data, header_request_id) + yield _parse_stream_chunk( + data, + header_request_id, + request_id, + ) + except Exception as error: + if isinstance(error, DashScopeError): + _with_audio_request_id(error) + _log_audio_error(error, request_id, response.headers) + raise finally: response.close() diff --git a/dashscope/resources/audio/realtime.py b/dashscope/resources/audio/realtime.py index 576a3987..cdb5695e 100644 --- a/dashscope/resources/audio/realtime.py +++ b/dashscope/resources/audio/realtime.py @@ -13,6 +13,7 @@ from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from dashscope._config import WS_REALTIME_PATH, workspace_ws_url +from dashscope._logging import get_logger, request_id_from_headers from dashscope.errors import ( APIConnectionError, APITimeoutError, @@ -30,6 +31,27 @@ MODULE = "audio" _UNSET = object() +_logger = get_logger("audio.realtime") + + +def _event_request_id(event: Any) -> Optional[str]: + """Read real request IDs only; event_id is a protocol correlation ID.""" + if not isinstance(event, dict): + return None + error = event.get("error") + response = event.get("response") + details = ( + response.get("status_details") if isinstance(response, dict) else None + ) + response_error = ( + details.get("error") if isinstance(details, dict) else None + ) + for source in (event, error, response_error): + if isinstance(source, dict): + value = source.get("request_id") + if isinstance(value, str) and value.strip(): + return value + return None def _default_connector( @@ -147,6 +169,10 @@ def __init__( self._reset_state() def _reset_state(self) -> None: + # Logging context is separate from the public latency/response metrics. + self._request_id: Optional[str] = None + self._protocol_request_id: Optional[str] = None + self._header_request_id: Optional[str] = None self._closed = False self._finish_requested = False self._finished = False @@ -173,6 +199,7 @@ def connect(self) -> "RealtimeSession": if self._ws is not None: raise RuntimeError("realtime session is already open") self._generation += 1 + self._reset_state() try: if self._connector is None: self._ws = _default_connector( @@ -182,9 +209,30 @@ def connect(self) -> "RealtimeSession": ) else: self._ws = self._connector(self._url, self._headers) + self._header_request_id = request_id_from_headers( + getattr(self._ws, "headers", None), + ) if self._timeout is not None: self._ws.settimeout(self._timeout) except BaseException as exc: + body = getattr(exc, "resp_body", None) + if isinstance(body, (str, bytes, bytearray)): + try: + body = json.loads(body) + except (ValueError, UnicodeError, RecursionError): + body = None + request_id = ( + _event_request_id(body) + or request_id_from_headers( + getattr(exc, "resp_headers", None), + ) + or self._header_request_id + ) + _logger.error( + "Realtime handshake failed: %s, request_id=%s", + type(exc).__name__, + request_id or "unknown", + ) self.close() self._reset_state() from websocket import WebSocketTimeoutException @@ -192,9 +240,16 @@ def connect(self) -> "RealtimeSession": if isinstance(exc, WebSocketTimeoutException): raise APITimeoutError(detail=str(exc)) from exc raise - self._reset_state() return self + def _request_id_for_log(self) -> str: + return ( + self._request_id + or self._header_request_id + or self._protocol_request_id + or "unknown" + ) + def __enter__(self) -> "RealtimeSession": return self.connect() @@ -219,7 +274,12 @@ def send_event(self, event: Dict[str, Any]) -> None: self._ws.send(raw) if payload.get("type") == "response.create": self._pending_response_ids.append(payload["event_id"]) - except BaseException: + except BaseException as exc: + _logger.error( + "Realtime send failed: %s, request_id=%s", + type(exc).__name__, + self._request_id_for_log(), + ) self.close() raise @@ -680,6 +740,7 @@ def recv_event(self) -> Optional[Dict[str, Any]]: return None raise RuntimeError("realtime session is not open") failure = None + received_on_current_connection = False try: try: raw = ws.recv() @@ -693,9 +754,14 @@ def recv_event(self) -> Optional[Dict[str, Any]]: if ws is not self._ws: self._check_unfinished() return None + received_on_current_connection = True if raw is None or raw == "" or raw == b"": self.close() self._check_unfinished() + _logger.debug( + "Realtime connection closed, request_id=%s", + self._request_id_for_log(), + ) return None if isinstance(raw, (bytes, bytearray)): raw = raw.decode("utf-8") @@ -722,6 +788,15 @@ def recv_event(self) -> Optional[Dict[str, Any]]: code=code, body=event, ) + # Only real IDs belong on exceptions; protocol fallbacks + # and "unknown" are strictly logging context. + explicit_request_id = _event_request_id(event) + if explicit_request_id is not None: + failure.request_id = explicit_request_id + _logger.error( + "Realtime business error, request_id=%s", + self._request_id_for_log(), + ) recoverable = ( error.get("type") == "invalid_request_error" ) @@ -735,8 +810,16 @@ def recv_event(self) -> Optional[Dict[str, Any]]: self._pending_response_ids.remove(request_id) if not recoverable or self._finish_requested: self.close() - except BaseException: + except BaseException as exc: with self._lock: + if generation == self._generation and ( + ws is self._ws or received_on_current_connection + ): + _logger.error( + "Realtime receive failed: %s, request_id=%s", + type(exc).__name__, + self._request_id_for_log(), + ) if ws is self._ws and generation == self._generation: self.close() raise @@ -763,8 +846,37 @@ def _check_unfinished(self) -> None: "realtime connection closed before response.done", ) + def _track_request_context(self, event: Dict[str, Any]) -> None: + """Keep request logging separate from response latency tracking.""" + kind = event.get("type") + self._request_id = _event_request_id(event) or self._request_id + protocol_id = event.get("response_id") or event.get("session_id") + response = event.get("response") + if kind in {"session.created", "session.updated"}: + session = event.get("session") + protocol_id = ( + session.get("id") if isinstance(session, dict) else None + ) or protocol_id + elif isinstance(kind, str) and kind.startswith("response."): + protocol_id = ( + response.get("id") if isinstance(response, dict) else None + ) or protocol_id + if isinstance(protocol_id, str) and protocol_id.strip(): + self._protocol_request_id = protocol_id + if kind == "conversation.item.input_audio_transcription.failed" or ( + kind == "response.done" + and isinstance(response, dict) + and response.get("status") == "failed" + ): + # These are raw terminal events, not exceptions in this API. + _logger.error( + "Realtime ASR/TTS response failed, request_id=%s", + self._request_id_for_log(), + ) + def _track_event(self, event: Dict[str, Any]) -> None: """Track first-output delays in monotonic milliseconds per response.""" + self._track_request_context(event) kind = event.get("type") if kind in {"session.created", "session.updated"}: self.session_id = (event.get("session") or {}).get( diff --git a/dashscope/resources/deployments.py b/dashscope/resources/deployments.py index 34c29241..0c45bafd 100644 --- a/dashscope/resources/deployments.py +++ b/dashscope/resources/deployments.py @@ -26,6 +26,7 @@ from dashscope.resources._sse import ( _aiter_sse, _iter_sse, + _log_stream_error, _stream_error, _try_json, ) @@ -289,7 +290,12 @@ def stream_events( for is_error, status_code, data in _iter_sse(response): if is_error: if self._client.raise_on_error: - raise _stream_error(status_code, data) + raise _stream_error( + status_code, + data, + response.headers, + ) + _log_stream_error(data, response.headers) yield DeploymentEvent( output=_try_json(data), is_error=True, @@ -444,7 +450,12 @@ async def stream_events( async for is_error, status_code, data in _aiter_sse(response): if is_error: if self._client.raise_on_error: - raise _stream_error(status_code, data) + raise _stream_error( + status_code, + data, + response.headers, + ) + _log_stream_error(data, response.headers) yield DeploymentEvent( output=_try_json(data), is_error=True, diff --git a/dashscope/resources/fine_tunes.py b/dashscope/resources/fine_tunes.py index 6946ff55..47ed066f 100644 --- a/dashscope/resources/fine_tunes.py +++ b/dashscope/resources/fine_tunes.py @@ -52,6 +52,7 @@ from dashscope.resources._sse import ( _aiter_sse, _iter_sse, + _log_stream_error, _stream_error, _try_json, ) @@ -398,7 +399,12 @@ def stream_events( for is_error, status_code, data in _iter_sse(response): if is_error: if self._client.raise_on_error: - raise _stream_error(status_code, data) + raise _stream_error( + status_code, + data, + response.headers, + ) + _log_stream_error(data, response.headers) yield FineTuneEvent(output=_try_json(data), is_error=True) return yield FineTuneEvent(output=_try_json(data)) @@ -626,7 +632,12 @@ async def stream_events( async for is_error, status_code, data in _aiter_sse(response): if is_error: if self._client.raise_on_error: - raise _stream_error(status_code, data) + raise _stream_error( + status_code, + data, + response.headers, + ) + _log_stream_error(data, response.headers) yield FineTuneEvent( output=_try_json(data), is_error=True, diff --git a/dashscope/transport/http.py b/dashscope/transport/http.py index 2c9164d2..799e264e 100644 --- a/dashscope/transport/http.py +++ b/dashscope/transport/http.py @@ -11,7 +11,7 @@ import httpx from dashscope._config import DEFAULT_TIMEOUT -from dashscope._logging import get_logger +from dashscope._logging import get_logger, request_id_from_headers from dashscope.errors import ( APIConnectionError, APITimeoutError, @@ -27,7 +27,10 @@ _WORKSPACE_HEADER = "X-DashScope-WorkSpace" -def _map_transport_error(exc: httpx.TransportError) -> DashScopeError: +def _map_transport_error( + exc: httpx.TransportError, + request_id: Optional[str] = None, +) -> DashScopeError: """Map raw httpx transport failures into the SDK error hierarchy. The message text comes from the catalog entry behind each class @@ -35,8 +38,8 @@ def _map_transport_error(exc: httpx.TransportError) -> DashScopeError: the ``{detail}``. """ if isinstance(exc, httpx.TimeoutException): - return APITimeoutError(detail=str(exc)) - return APIConnectionError(detail=str(exc)) + return APITimeoutError(detail=str(exc), request_id=request_id) + return APIConnectionError(detail=str(exc), request_id=request_id) def _resolve_timeout( @@ -204,39 +207,62 @@ def request( ) -> httpx.Response: url = _abs_url(base_url or self._base_url, path) for attempt in range(self._max_retries + 1): + response = None try: - response = self._client.request( - method, - url, - json=json, - content=content, - data=data, - files=files, - params=params, - headers=_request_headers( - self._default_headers, - sdk_headers(sdk_module), - headers, - api_key, - workspace, - ), - **_timeout_kwarg( - self._request_timeout if timeout is None else timeout, + response = self._client.send( + self._client.build_request( + method, + url, + json=json, + content=content, + data=data, + files=files, + params=params, + headers=_request_headers( + self._default_headers, + sdk_headers(sdk_module), + headers, + api_key, + workspace, + ), + **_timeout_kwarg( + ( + self._request_timeout + if timeout is None + else timeout + ), + ), ), + stream=True, ) + try: + response.read() + finally: + response.close() except httpx.TransportError as exc: + request_id = ( + request_id_from_headers(response.headers) + if response is not None + else None + ) + context = ( + f" (request_id={request_id or 'unknown'})" + if response is not None + else "" + ) if attempt >= self._max_retries: logger.warning( - "request %s %s failed after %d attempt(s): %s", + "request %s %s failed after %d attempt(s): %s%s", method, path, attempt + 1, exc, + context, ) - raise _map_transport_error(exc) from exc + raise _map_transport_error(exc, request_id) from exc delay = backoff_delay(attempt, None) logger.info( - "retrying %s %s in %.2fs (attempt %d/%d): %s: %s", + "retrying %s %s in %.2fs (attempt %d/%d): %s: %s%s", method, path, delay, @@ -244,10 +270,11 @@ def request( self._max_retries, type(exc).__name__, exc, + context, ) _sleep(delay) continue - request_id = response.headers.get("x-request-id") + request_id = request_id_from_headers(response.headers) or "unknown" logger.debug( "request %s %s -> %d (request_id=%s)", method, @@ -312,9 +339,11 @@ def open_stream( workspace, ), **_timeout_kwarg( - self._request_timeout - if timeout is None - else timeout, + ( + self._request_timeout + if timeout is None + else timeout + ), ), ), stream=True, @@ -340,7 +369,9 @@ def open_stream( attempt < self._max_retries ): delay = backoff_delay(attempt, response.headers) - request_id = response.headers.get("x-request-id") + request_id = ( + request_id_from_headers(response.headers) or "unknown" + ) try: response.read() except httpx.TransportError: @@ -368,7 +399,12 @@ def open_stream( response.read() check_response(response) except httpx.TransportError as exc: - raise _map_transport_error(exc) from exc + request_id = request_id_from_headers(response.headers) + logger.error( + "Failed to read HTTP response, request_id=%s", + request_id or "unknown", + ) + raise _map_transport_error(exc, request_id) from exc finally: response.close() logger.debug( @@ -376,7 +412,7 @@ def open_stream( method, path, response.status_code, - response.headers.get("x-request-id"), + request_id_from_headers(response.headers) or "unknown", ) return response raise AssertionError("unreachable") @@ -441,39 +477,62 @@ async def request( ) -> httpx.Response: url = _abs_url(base_url or self._base_url, path) for attempt in range(self._max_retries + 1): + response = None try: - response = await self._client.request( - method, - url, - json=json, - content=content, - data=data, - files=files, - params=params, - headers=_request_headers( - self._default_headers, - sdk_headers(sdk_module), - headers, - api_key, - workspace, - ), - **_timeout_kwarg( - self._request_timeout if timeout is None else timeout, + response = await self._client.send( + self._client.build_request( + method, + url, + json=json, + content=content, + data=data, + files=files, + params=params, + headers=_request_headers( + self._default_headers, + sdk_headers(sdk_module), + headers, + api_key, + workspace, + ), + **_timeout_kwarg( + ( + self._request_timeout + if timeout is None + else timeout + ), + ), ), + stream=True, ) + try: + await response.aread() + finally: + await response.aclose() except httpx.TransportError as exc: + request_id = ( + request_id_from_headers(response.headers) + if response is not None + else None + ) + context = ( + f" (request_id={request_id or 'unknown'})" + if response is not None + else "" + ) if attempt >= self._max_retries: logger.warning( - "request %s %s failed after %d attempt(s): %s", + "request %s %s failed after %d attempt(s): %s%s", method, path, attempt + 1, exc, + context, ) - raise _map_transport_error(exc) from exc + raise _map_transport_error(exc, request_id) from exc delay = backoff_delay(attempt, None) logger.info( - "retrying %s %s in %.2fs (attempt %d/%d): %s: %s", + "retrying %s %s in %.2fs (attempt %d/%d): %s: %s%s", method, path, delay, @@ -481,10 +540,11 @@ async def request( self._max_retries, type(exc).__name__, exc, + context, ) await _asleep(delay) continue - request_id = response.headers.get("x-request-id") + request_id = request_id_from_headers(response.headers) or "unknown" logger.debug( "request %s %s -> %d (request_id=%s)", method, @@ -567,7 +627,9 @@ async def open_stream( attempt < self._max_retries ): delay = backoff_delay(attempt, response.headers) - request_id = response.headers.get("x-request-id") + request_id = ( + request_id_from_headers(response.headers) or "unknown" + ) try: await response.aread() except httpx.TransportError: @@ -593,7 +655,12 @@ async def open_stream( await response.aread() check_response(response) except httpx.TransportError as exc: - raise _map_transport_error(exc) from exc + request_id = request_id_from_headers(response.headers) + logger.error( + "Failed to read HTTP response, request_id=%s", + request_id or "unknown", + ) + raise _map_transport_error(exc, request_id) from exc finally: await response.aclose() logger.debug( @@ -601,7 +668,7 @@ async def open_stream( method, path, response.status_code, - response.headers.get("x-request-id"), + request_id_from_headers(response.headers) or "unknown", ) return response raise AssertionError("unreachable") diff --git a/dashscope/transport/sse.py b/dashscope/transport/sse.py index 1dfb6245..ea0ec12c 100644 --- a/dashscope/transport/sse.py +++ b/dashscope/transport/sse.py @@ -30,10 +30,17 @@ ) import httpx -from httpx_sse import EventSource +from httpx_sse import EventSource, SSEError +from dashscope._logging import ( + get_logger, + request_id_for_log, + request_id_from_headers, +) from dashscope.errors import DashScopeError, error_from_business_code +logger = get_logger("transport.sse") + T = TypeVar("T") # OpenAI-compatible endpoints terminate streams with this sentinel. @@ -80,17 +87,63 @@ def stream_error( def _next_event( data: str, headers: Optional[Mapping[str, str]] = None, + request_id: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """Decode one event, raising on in-stream errors; None to skip.""" if not data or data == DONE_MARKER: return None - body = parse_event_data(data) + try: + body = parse_event_data(data) + except DashScopeError as exc: + exc.request_id = request_id or request_id_from_headers(headers) + exc.headers = headers or {} + logger.error( + "Failed to parse SSE event, request_id=%s", + request_id_for_log( + exc.request_id, + request_id_from_headers(headers), + ), + ) + raise error = stream_error(body, headers) if error is not None: + error.request_id = ( + body.get("request_id") or request_id or error.request_id + ) + logger.error( + "SSE business error: code=%s, request_id=%s", + error.code, + request_id_for_log( + body.get("request_id"), + request_id, + error.request_id, + request_id_from_headers(headers), + ), + ) raise error return body +def _parse_chunk( + parse: Callable[[Dict[str, Any]], Optional[T]], + body: Dict[str, Any], + headers: Mapping[str, str], + request_id: Optional[str] = None, +) -> Optional[T]: + try: + return parse(body) + except Exception: + logger.error( + "Failed to parse SSE result, request_id=%s", + request_id_for_log( + body.get("request_id"), + request_id, + request_id_from_headers(headers), + ), + ) + raise + + class Stream(Generic[T]): """Typed iterator over SSE chunks; in-stream errors raise.""" @@ -101,16 +154,41 @@ def __init__( ) -> None: self._response = response self._parse = parse + self._request_id = request_id_from_headers(response.headers) def __iter__(self) -> Iterator[T]: try: for sse in EventSource(self._response).iter_sse(): - body = _next_event(sse.data, self._response.headers) + body = _next_event( + sse.data, + self._response.headers, + self._request_id, + ) if body is None: continue - chunk = self._parse(body) + body_request_id = body.get("request_id") + if ( + isinstance(body_request_id, str) + and body_request_id.strip() + ): + self._request_id = body_request_id + chunk = _parse_chunk( + self._parse, + body, + self._response.headers, + self._request_id, + ) if chunk is not None: yield chunk + except (httpx.TransportError, httpx.DecodingError, SSEError): + logger.error( + "Failed to read SSE stream, request_id=%s", + request_id_for_log( + self._request_id, + request_id_from_headers(self._response.headers), + ), + ) + raise finally: self._response.close() @@ -131,6 +209,7 @@ def __init__( ) -> None: self._response = response self._parse = parse + self._request_id = request_id_from_headers(response.headers) def __aiter__(self): return self._iterate() @@ -138,12 +217,36 @@ def __aiter__(self): async def _iterate(self): try: async for sse in EventSource(self._response).aiter_sse(): - body = _next_event(sse.data, self._response.headers) + body = _next_event( + sse.data, + self._response.headers, + self._request_id, + ) if body is None: continue - chunk = self._parse(body) + body_request_id = body.get("request_id") + if ( + isinstance(body_request_id, str) + and body_request_id.strip() + ): + self._request_id = body_request_id + chunk = _parse_chunk( + self._parse, + body, + self._response.headers, + self._request_id, + ) if chunk is not None: yield chunk + except (httpx.TransportError, httpx.DecodingError, SSEError): + logger.error( + "Failed to read SSE stream, request_id=%s", + request_id_for_log( + self._request_id, + request_id_from_headers(self._response.headers), + ), + ) + raise finally: await self._response.aclose() diff --git a/dashscope/transport/ws.py b/dashscope/transport/ws.py index 3a2201a7..e6221446 100644 --- a/dashscope/transport/ws.py +++ b/dashscope/transport/ws.py @@ -16,6 +16,7 @@ settings, workspace_ws_url, ) +from dashscope._logging import get_logger, request_id_from_headers from dashscope.errors import ( APIConnectionError, WsEventError, @@ -42,6 +43,42 @@ Connector = Callable[[str, Dict[str, str]], Any] +_logger = get_logger("transport.ws") + + +def _log_error( + operation: str, + exc: BaseException, + request_id: Optional[str] = None, +) -> None: + # Exception text/tracebacks can contain credentials or entire frames. + if isinstance(exc, Exception): + _logger.error( + "websocket %s failed request_id=%s error_type=%s", + operation, + request_id or "unknown", + type(exc).__name__, + ) + + +def _handshake_request_id(exc: BaseException) -> Optional[str]: + headers = getattr(exc, "resp_headers", None) + request_id = request_id_from_headers( + headers if isinstance(headers, Mapping) else None, + ) + body = getattr(exc, "resp_body", None) + if isinstance(body, (str, bytes, bytearray)): + try: + body = json.loads(body) + except Exception: + # Diagnostic parsing must not replace the handshake exception. + return request_id + if isinstance(body, dict): + body_id = body.get("request_id") + if isinstance(body_id, str) and body_id.strip(): + return body_id + return request_id + def _default_connector( url: str, @@ -108,10 +145,15 @@ def _connect( except BaseException as exc: from websocket import WebSocketTimeoutException + request_id = _handshake_request_id(exc) + _log_error("handshake", exc, request_id) if isinstance(exc, WebSocketTimeoutException): from dashscope.errors import APITimeoutError - raise APITimeoutError(detail=str(exc)) from exc + raise APITimeoutError( + detail=str(exc), + request_id=request_id, + ) from exc raise try: if timeout is not None: @@ -157,7 +199,10 @@ def _process_event(self, event: Dict[str, Any]) -> Dict[str, Any]: event.get("header"), dict, ): - raise WsEventError(detail="expected a websocket event header") + raise WsEventError( + detail="expected a websocket event header", + request_id=self._request_id, + ) header = event["header"] if header.get("task_id") not in (None, self.task_id): raise WsEventError( @@ -165,19 +210,20 @@ def _process_event(self, event: Dict[str, Any]) -> Dict[str, Any]: request_id=self._request_id, body=event, ) - self._request_id = ( - header.get("request_id") - or header.get("task_id") - or self._request_id - ) + request_id = header.get("request_id") + if isinstance(request_id, str) and request_id.strip(): + self._request_id = request_id if header.get("event") == EVENT_FAILED: self._finished = True - raise error_from_business_code( + error = error_from_business_code( message=header.get("error_message") or "task failed", code=header.get("error_code"), body=event, headers={"x-request-id": self._request_id}, ) + error.request_id = self._request_id + _log_error(EVENT_FAILED, error, self._request_id) + raise error if header.get("event") == EVENT_FINISHED: self._finished = True return event @@ -244,12 +290,17 @@ def _frame(self, action: str, payload: Dict[str, Any]) -> Dict[str, Any]: } def _send(self, data: Any, *, opcode: int = 1) -> None: + sending = opcode == 2 try: if opcode == 2: self._ws.send(data, opcode=2) else: - self._ws.send(json.dumps(data)) - except BaseException: + encoded = json.dumps(data) + sending = True + self._ws.send(encoded) + except BaseException as exc: + if sending and not self._closed: + _log_error("send", exc, self.request_id) self.close() raise @@ -359,7 +410,11 @@ def recv(self) -> Union[Dict[str, Any], bytes, None]: if isinstance(raw, (bytes, bytearray)): return bytes(raw) return self._process_event(json.loads(raw)) - except BaseException: + except BaseException as exc: + # task-failed is already logged by _process_event; a concurrent + # explicit close should still wake receivers without an error log. + if not self._finished and not self._closed: + _log_error("receive", exc, self.request_id) self.close() raise diff --git a/dashscope/transport/ws_connection.py b/dashscope/transport/ws_connection.py index 43165757..44fd372f 100644 --- a/dashscope/transport/ws_connection.py +++ b/dashscope/transport/ws_connection.py @@ -57,6 +57,7 @@ _close_websocket, _connect, _EventProcessor, + _log_error, _recv_frame, ) @@ -99,6 +100,8 @@ def __init__( # Reader-side routing uses the connection lock, not the send lock. self._routing_active = False self._routing_finished = False + # Retain IDs even when events are still waiting for the consumer. + self._routing_request_id = task_id self._error: Optional[BaseException] = None @property @@ -127,9 +130,12 @@ def _enqueue(self, item: _QueueItem) -> None: # fail only this task and let its reader surface the error. item = ( _ERROR, - WsQueueCapacityExceededError(), + WsQueueCapacityExceededError( + request_id=self._routing_request_id, + ), ) self._error = item[1] + _log_error("receive queue", item[1], self._routing_request_id) # Preserve terminal visibility without allowing an unbounded # backlog to retain audio forever: evict the oldest frame. try: @@ -178,7 +184,7 @@ def run_task( if self._closed or self._run_sent: raise WsStateError(detail="task already started or closed") self._run_sent = True - self._conn._send(frame) + self._conn._send(frame, task=self) try: event = self.recv_event() if event["header"].get("event") != EVENT_STARTED: @@ -214,12 +220,15 @@ def send_binary(self, data: bytes) -> None: with self._lock: self._check_input() self._conn._claim_binary_uplink(self) - self._conn._send(bytes(data), opcode=2) + self._conn._send(bytes(data), opcode=2, task=self) def continue_task(self, input: Dict[str, Any]) -> None: with self._lock: self._check_input() - self._conn._send(self._frame("continue-task", {"input": input})) + self._conn._send( + self._frame("continue-task", {"input": input}), + task=self, + ) def finish_task(self, input: Optional[Dict[str, Any]] = None) -> None: """End input, or cancel synthesis while output is still arriving.""" @@ -233,6 +242,7 @@ def finish_task(self, input: Optional[Dict[str, Any]] = None) -> None: return self._conn._send( self._frame("finish-task", {"input": input or {}}), + task=self, ) self._input_finished = True self._cancel_sent = cancel @@ -253,9 +263,12 @@ def recv(self) -> Union[Dict[str, Any], bytes, None]: try: kind, payload = self._queue.get(timeout=self._timeout) except queue.Empty: - raise APITimeoutError( + error = APITimeoutError( detail="timed out waiting for a websocket event", - ) from None + request_id=self._routing_request_id, + ) + _log_error("receive", error, self._routing_request_id) + raise error from None if kind == _ERROR: # Requeue the terminal marker so subsequent readers also fail. self._queue.put((kind, payload)) @@ -363,16 +376,31 @@ def open_task(self, *, timeout: Optional[float] = None) -> WsTask: self._tasks[task.task_id] = task return task - def _send(self, data: Any, *, opcode: int = 1) -> None: + def _send( + self, + data: Any, + *, + opcode: int = 1, + task: Optional[WsTask] = None, + ) -> None: with self._send_lock: if self._closed: raise WsStateError(detail="connection is closed") + sending = opcode == 2 try: if opcode == 2: self._ws.send(data, opcode=2) else: - self._ws.send(json.dumps(data)) - except BaseException: + encoded = json.dumps(data) + sending = True + self._ws.send(encoded) + except BaseException as exc: + if sending and not self._closed: + _log_error( + "send", + exc, + task._routing_request_id if task is not None else None, + ) self.close() raise @@ -435,7 +463,7 @@ def _heartbeat_main(self) -> None: return self._ws.ping() except BaseException as exc: - self._fail(APIConnectionError(str(exc))) + self._fail(APIConnectionError(str(exc)), operation="heartbeat") return def _reader_main(self) -> None: @@ -465,6 +493,9 @@ def _route_event(self, event: Any) -> None: if self._closed or task is None or task._routing_finished: # Late events for detached or terminal tasks are harmless. return + request_id = header.get("request_id") + if isinstance(request_id, str) and request_id.strip(): + task._routing_request_id = request_id name = header.get("event") if name == EVENT_STARTED: task._routing_active = True @@ -497,7 +528,12 @@ def _route_binary(self, data: bytes) -> None: self._binary_downlink_owner = owner owner._enqueue((_FRAME, data)) - def _fail(self, exc: BaseException) -> None: + def _fail( + self, + exc: BaseException, + *, + operation: Optional[str] = "receive", + ) -> None: with self._lock: if self._closed: # close() already broadcast; the reader's socket error is @@ -507,6 +543,16 @@ def _fail(self, exc: BaseException) -> None: self._heartbeat_stop.set() self._binary_uplink_owner = None self._binary_downlink_owner = None + request_ids: List[Optional[str]] = [ + task._routing_request_id + for task in self._tasks.values() + if not task._closed + and not task._routing_finished + and task._error is None + ] + if operation is not None: + for request_id in request_ids or [None]: + _log_error(operation, exc, request_id) _close_websocket(self._ws) self._broadcast(exc) @@ -514,10 +560,14 @@ def _broadcast(self, exc: BaseException) -> None: with self._lock: for task in self._tasks.values(): task._routing_active = False + # This exception is shared: never stamp one task's ID on it. task._enqueue((_ERROR, exc)) def close(self) -> None: - self._fail(APIConnectionError("websocket connection closed")) + self._fail( + APIConnectionError("websocket connection closed"), + operation=None, + ) # Initialization may fail before the reader is created or started. reader = getattr(self, "_reader", None) if ( diff --git a/dashscope/types/common.py b/dashscope/types/common.py index 0c08f563..d79f60ff 100644 --- a/dashscope/types/common.py +++ b/dashscope/types/common.py @@ -9,6 +9,11 @@ import httpx from pydantic import BaseModel, ConfigDict, Field, model_validator +from dashscope._logging import ( + get_logger, + request_id_for_log, + request_id_from_headers, +) from dashscope.errors import ( error_from_business_code, error_from_status, @@ -121,6 +126,10 @@ def with_headers(model: _M, response: Any) -> _M: ``ChatStream.get_final_message()``. """ model.headers = dict(response.headers) + if hasattr(model, "request_id") and not model.request_id: + request_id = request_id_from_headers(response.headers) + if request_id is not None: + model.request_id = request_id return model @@ -166,7 +175,7 @@ def response_body( request_id = ( body.get("request_id") or output.get("request_id") - or response.headers.get("x-request-id") + or request_id_from_headers(response.headers) ) error = None if not success: @@ -182,9 +191,20 @@ def response_body( body=raw, headers=response.headers, ) - if error is not None and raise_on_error: + if error is not None: error.request_id = request_id - raise error + get_logger("response").error( + "HTTP service error: status=%s, code=%s, request_id=%s", + response.status_code, + error.code, + request_id_for_log( + body.get("request_id"), + output.get("request_id"), + request_id_from_headers(response.headers), + ), + ) + if raise_on_error: + raise error body = {**body, "status_code": response.status_code} if request_id is not None: body["request_id"] = request_id diff --git a/tests/test_audio.py b/tests/test_audio.py index fecfa1a9..bd0b7176 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -911,7 +911,8 @@ def test_recognition_results_preserve_optional_request_id_and_extra_fields(): assert result.get_usage() == {"duration": 0} assert result.future_field == {"ok": True} assert result.request_id is None - assert rec.request_id == rec.session.task_id + # An event without request_id must not erase task-started's ID. + assert rec.request_id == "r" def test_recognition_same_object_restarts_with_fresh_state_after_close(): @@ -1020,9 +1021,11 @@ def send(self, payload, opcode=1): { "header": { "event": "task-started", - "task_id": "foreign-task" - if phase == "start" - else task_id, + "task_id": ( + "foreign-task" + if phase == "start" + else task_id + ), }, "payload": {}, }, diff --git a/tests/test_audio_tts_http.py b/tests/test_audio_tts_http.py index 670a7e2d..6a4b45bf 100644 --- a/tests/test_audio_tts_http.py +++ b/tests/test_audio_tts_http.py @@ -9,6 +9,7 @@ import base64 import json +import logging import httpx import pytest @@ -1137,3 +1138,421 @@ def test_audio_chunk_keeps_wire_status_flag_local(raise_on_error, status): assert response.is_closed and wire.closed finally: client.transport.http_client.close() + + +_HTTP_TTS_LOGGER = "dashscope._logging.audio.http_tts" + + +def _http_tts_errors(caplog): + return [ + record + for record in caplog.records + if record.name == _HTTP_TTS_LOGGER and record.levelno >= logging.ERROR + ] + + +def _assert_http_tts_log(caplog, request_id): + (record,) = _http_tts_errors(caplog) + assert f"request_id={request_id}" in record.getMessage() + assert record.exc_info is None + assert "private-payload" not in record.getMessage() + assert "private-key" not in record.getMessage() + + +class _AsyncTrackedStream(httpx.AsyncByteStream): + def __init__(self, *parts, error=None): + self.parts = parts + self.error = error + self.closed = False + + async def __aiter__(self): + for part in self.parts: + yield part + if self.error is not None: + raise self.error + + async def aclose(self): + self.closed = True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + "fault", + [ + "json", + "shape", + "status", + "base64", + "output", + "business", + "error-frame", + "error-status", + "json-envelope", + "http-status", + ], +) +@pytest.mark.parametrize( + "request_ids", + [ + ("body-id", "output-id", "body-id"), + (None, "output-id", "output-id"), + (None, None, "header-id"), + ("", "", "header-id"), + ], +) +async def test_http_tts_stream_failure_logs_and_preserves_contract( + caplog, + async_mode, + fault, + request_ids, +): + outer_id, inner_id, expected_id = request_ids + caplog.set_level(logging.ERROR, logger=_HTTP_TTS_LOGGER) + body = { + "request_id": outer_id, + "output": {"request_id": inner_id, "audio": {}}, + } + if fault == "json": + data = b"data: private-payload not JSON\n\n" + elif fault == "shape": + data = b'data: ["private-payload"]\n\n' + elif fault == "status": + data = b"status: private-payload\n" + _frame(body) + elif fault == "base64": + body["output"]["audio"]["data"] = "private-payload%%%" + data = _frame(body) + elif fault == "output": + body["status_code"] = "private-payload" + data = _frame(body) + else: + body.update( + code="InvalidParameter", + message="private-payload Bearer private-key", + ) + prefixes = { + "error-frame": b"event: error\n", + "error-status": b"event: error\nstatus: 400\n", + } + data = ( + json.dumps(body).encode() + if fault in {"json-envelope", "http-status"} + else prefixes.get(fault, b"") + _frame(body) + ) + media_type = ( + "application/json" + if fault in {"json-envelope", "http-status"} + else "text/event-stream" + ) + wire = (_AsyncTrackedStream if async_mode else _TrackedStream)(data) + response = httpx.Response( + 400 if fault == "http-status" else 200, + stream=wire, + headers={"content-type": media_type, "X-Request-ID": "header-id"}, + ) + client = (_make_async_client if async_mode else _make_client)( + lambda request: response, + max_retries=0, + ) + if fault in { + "business", + "error-frame", + "error-status", + "json-envelope", + "http-status", + }: + error_type = BadRequestError + elif fault in {"json", "shape"}: + error_type = DashScopeError + else: + error_type = ValueError + try: + with pytest.raises(error_type) as caught: + if async_mode: + stream = await client.audio.tts.http.stream( + "private-payload", + **_TTS_ARGS, + ) + async for _ in stream: + pass + else: + list( + client.audio.tts.http.stream( + "private-payload", + **_TTS_ARGS, + ), + ) + if fault in {"json", "shape", "status"}: + expected_id = "header-id" + _assert_http_tts_log(caplog, expected_id) + if fault in {"json", "shape"}: + assert caught.value.request_id == "header-id" + elif error_type is BadRequestError: + assert caught.value.request_id == expected_id + assert caught.value.body == body + assert caught.value.status_code == ( + 400 if fault in {"error-status", "http-status"} else 200 + ) + assert response.is_closed and wire.closed + finally: + if async_mode: + await client.aclose() + else: + client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + "headers, expected_id", + [ + ({"X-Request-ID": "mixed-id"}, "mixed-id"), + ({"X-DashScope-RequestId": "alias-id"}, "alias-id"), + ( + {"X-Request-ID": "", "X-DashScope-Request-ID": "fallback-id"}, + "fallback-id", + ), + ({}, "unknown"), + ], +) +async def test_http_tts_malformed_stream_logs_header_alias_or_unknown( + caplog, + async_mode, + headers, + expected_id, +): + caplog.set_level(logging.ERROR, logger=_HTTP_TTS_LOGGER) + wire = (_AsyncTrackedStream if async_mode else _TrackedStream)( + b"data: private-payload\n\n", + ) + response = httpx.Response(200, stream=wire, headers=headers) + client = (_make_async_client if async_mode else _make_client)( + lambda request: response, + ) + try: + with pytest.raises(DashScopeError) as caught: + if async_mode: + stream = await client.audio.tts.http.stream( + "text", + **_TTS_ARGS, + ) + async for _ in stream: + pass + else: + list(client.audio.tts.http.stream("text", **_TTS_ARGS)) + # Header aliases enrich logs without changing per-chunk/public ID rules. + assert caught.value.request_id == (headers.get("X-Request-ID") or None) + assert response.is_closed and wire.closed + _assert_http_tts_log(caplog, expected_id) + finally: + if async_mode: + await client.aclose() + else: + client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("fault", ["json", "read"]) +async def test_http_tts_stream_retains_body_id_only_in_log_context( + caplog, + async_mode, + fault, +): + caplog.set_level(logging.ERROR, logger=_HTTP_TTS_LOGGER) + parts = [ + _frame({"request_id": "body-id", "output": {"audio": {}}}), + _frame({"output": {"request_id": "output-id", "audio": {}}}), + ] + if fault == "json": + parts.append(b"data: private-payload\n\n") + wire = (_AsyncTrackedStream if async_mode else _TrackedStream)( + *parts, + error=httpx.ReadError("private-payload") if fault == "read" else None, + ) + response = httpx.Response( + 200, + stream=wire, + headers={"x-request-id": "header-id"}, + ) + client = (_make_async_client if async_mode else _make_client)( + lambda request: response, + ) + chunks = [] + try: + with pytest.raises( + DashScopeError if fault == "json" else httpx.ReadError, + ): + if async_mode: + stream = await client.audio.tts.http.stream( + "text", + **_TTS_ARGS, + ) + async for chunk in stream: + chunks.append(chunk) + else: + stream = client.audio.tts.http.stream("text", **_TTS_ARGS) + chunks.append(next(stream)) + chunks.append(next(stream)) + next(stream) + assert [chunk.request_id for chunk in chunks] == [ + "body-id", + "header-id", + ] + _assert_http_tts_log(caplog, "output-id") + assert response.is_closed and wire.closed + finally: + if async_mode: + await client.aclose() + else: + client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("raise_on_error", [False, True]) +@pytest.mark.parametrize("terminated", [False, True]) +@pytest.mark.parametrize( + "request_ids", + [ + ("body-id", "output-id", "body-id"), + (None, "output-id", "output-id"), + (None, None, "header-id"), + ], +) +async def test_http_tts_error_chunk_logs_without_raising( + caplog, + async_mode, + raise_on_error, + terminated, + request_ids, +): + outer_id, inner_id, expected_id = request_ids + caplog.set_level(logging.ERROR, logger=_HTTP_TTS_LOGGER) + body = { + "status_code": 503, + "request_id": outer_id, + "output": {"request_id": inner_id, "audio": {}}, + } + data = _frame(body) + if not terminated: + data = data.rstrip(b"\n") + wire = (_AsyncTrackedStream if async_mode else _TrackedStream)(data) + response = httpx.Response( + 200, + stream=wire, + headers={"x-request-id": "header-id"}, + ) + client = (_make_async_client if async_mode else _make_client)( + lambda request: response, + raise_on_error=raise_on_error, + ) + try: + if async_mode: + stream = await client.audio.tts.http.stream("text", **_TTS_ARGS) + chunks = [chunk async for chunk in stream] + else: + chunks = list(client.audio.tts.http.stream("text", **_TTS_ARGS)) + (chunk,) = chunks + assert chunk.is_error and chunk.status_code == 503 + assert chunk.request_id == (outer_id or "header-id") + assert chunk.output == body["output"] + _assert_http_tts_log(caplog, expected_id) + assert response.is_closed and wire.closed + finally: + if async_mode: + await client.aclose() + else: + client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("raise_on_error", [False, True]) +@pytest.mark.parametrize("malformed", [False, True]) +async def test_http_tts_one_shot_failure_logs( + caplog, + async_mode, + raise_on_error, + malformed, +): + caplog.set_level(logging.ERROR, logger=_HTTP_TTS_LOGGER) + body = { + "code": "InvalidParameter", + "message": "private-payload Bearer private-key", + "request_id": "body-id", + "output": {"request_id": "output-id"}, + } + response = httpx.Response( + 200, + content=b"private-payload" if malformed else json.dumps(body).encode(), + headers={"X-DashScope-RequestId": "header-id"}, + ) + client = (_make_async_client if async_mode else _make_client)( + lambda request: response, + raise_on_error=raise_on_error, + ) + try: + if malformed or raise_on_error: + with pytest.raises( + InternalServerError if malformed else BadRequestError, + ): + if async_mode: + await client.audio.tts.http.synthesize("text", **_TTS_ARGS) + else: + client.audio.tts.http.synthesize("text", **_TTS_ARGS) + else: + if async_mode: + result = await client.audio.tts.http.synthesize( + "text", + **_TTS_ARGS, + ) + else: + result = client.audio.tts.http.synthesize("text", **_TTS_ARGS) + assert result.is_error + assert result.request_id == "body-id" + _assert_http_tts_log(caplog, "header-id" if malformed else "body-id") + finally: + if async_mode: + await client.aclose() + else: + client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_http_tts_success_close_and_validation_do_not_log_errors( + caplog, + async_mode, +): + caplog.set_level(logging.DEBUG, logger=_HTTP_TTS_LOGGER) + wire = (_AsyncTrackedStream if async_mode else _TrackedStream)( + _frame({"output": {}}), + ) + response = httpx.Response(200, stream=wire) + client = (_make_async_client if async_mode else _make_client)( + lambda request: response, + ) + try: + with pytest.raises(KeyError if async_mode else TypeError): + if async_mode: + await client.audio.tts.http.stream("text") + else: + client.audio.tts.http.stream("text") + if async_mode: + stream = await client.audio.tts.http.stream("text", **_TTS_ARGS) + async for chunk in stream: + assert chunk.request_id == "" + await stream.aclose() + else: + stream = client.audio.tts.http.stream("text", **_TTS_ARGS) + (chunk,) = list(stream) + assert chunk.request_id == "" + stream.close() + assert response.is_closed and wire.closed + assert not _http_tts_errors(caplog) + finally: + if async_mode: + await client.aclose() + else: + client.close() diff --git a/tests/test_audio_tts_realtime.py b/tests/test_audio_tts_realtime.py index f1a5fe76..6ee5723f 100644 --- a/tests/test_audio_tts_realtime.py +++ b/tests/test_audio_tts_realtime.py @@ -8,15 +8,18 @@ from __future__ import annotations import json +import logging import pytest from websocket import ( + WebSocketBadStatusException, WebSocketConnectionClosedException, WebSocketTimeoutException, ) from dashscope import Client from dashscope.errors import ( + APIConnectionError, APITimeoutError, BadRequestError, DashScopeError, @@ -507,3 +510,426 @@ def test_realtime_recv_raises_on_transport_error(): ) as rt: with pytest.raises(APITimeoutError): rt.recv_event() + + +_REALTIME_LOGGER = "dashscope._logging.audio.realtime" + + +def _realtime_errors(caplog): + return [ + record + for record in caplog.records + if record.name == _REALTIME_LOGGER and record.levelno >= logging.ERROR + ] + + +def _assert_realtime_log(caplog, request_id): + (record,) = _realtime_errors(caplog) + assert f"request_id={request_id}" in record.getMessage() + assert record.exc_info is None + assert "private-payload" not in record.getMessage() + assert "private-key" not in record.getMessage() + return record + + +@pytest.mark.parametrize( + "body, headers, expected_id", + [ + ({"request_id": "body-id"}, {"X-Request-ID": "header-id"}, "body-id"), + ( + {"request_id": "", "error": {"request_id": "nested-id"}}, + {"X-DashScope-RequestId": "header-id"}, + "nested-id", + ), + ( + b'{"request_id":"bytes-id"}', + {"X-DashScope-Request-ID": "header-id"}, + "bytes-id", + ), + ( + "private-payload not JSON", + {"X-Request-ID": "", "X-DashScope-Request-ID": "alias-id"}, + "alias-id", + ), + ("[" * 2000, {"X-Request-ID": "header-id"}, "header-id"), + ({"event_id": "not-a-request"}, {}, "unknown"), + ], +) +def test_realtime_handshake_request_id_logs( + caplog, + body, + headers, + expected_id, +): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + failure = WebSocketBadStatusException( + "private-payload Bearer private-key", + status_code=401, + resp_headers=headers, + resp_body=json.dumps(body) if isinstance(body, dict) else body, + ) + + def connector(url, request_headers): + raise failure + + with _make_client() as client: + session = client.audio.realtime.connect( + model="omni", + connector=connector, + ) + with pytest.raises(WebSocketBadStatusException) as caught: + session.connect() + assert caught.value is failure + assert session.session_id is None + assert session.last_response_id is None + _assert_realtime_log(caplog, expected_id) + + +@pytest.mark.parametrize( + "outer_id, nested_id, expected_id", + [ + ("outer-id", "nested-id", "outer-id"), + (None, "nested-id", "nested-id"), + (None, None, "handshake-id"), + ], +) +@pytest.mark.parametrize("recoverable", [False, True]) +def test_realtime_business_request_id_context( + caplog, + outer_id, + nested_id, + expected_id, + recoverable, +): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + event = { + "type": "error", + "event_id": "server-event-not-request", + "request_id": outer_id, + "error": { + "type": "invalid_request_error" if recoverable else "server_error", + "code": "InvalidParameter", + "message": "private-payload Bearer private-key", + "request_id": nested_id, + "event_id": "client-event-not-request", + }, + } + ws = FakeWebSocket( + [ + json.dumps( + {"type": "session.created", "session": {"id": "session-id"}}, + ), + json.dumps(event), + ], + ) + ws.headers = {"X-DashScope-RequestId": "handshake-id"} + with _make_client() as client: + with client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: ws, + ) as session: + session.recv_event() + with pytest.raises(BadRequestError) as caught: + session.recv_event() + assert caught.value.request_id == (outer_id or nested_id) + assert caught.value.body == event + assert caught.value.status_code == 200 + assert session.session_id == "session-id" + assert session.last_response_id is None + assert ws.closed == (not recoverable) + if recoverable: + session.commit() + assert not ws.closed + _assert_realtime_log(caplog, expected_id) + + +@pytest.mark.parametrize("stage", ["send", "recv", "disconnect", "parse"]) +@pytest.mark.parametrize( + "context, expected_id", + [ + ({}, "unknown"), + ({"session": {"id": "session-id"}}, "session-id"), + ( + {"request_id": "body-id", "session": {"id": "session-id"}}, + "body-id", + ), + ], +) +def test_realtime_runtime_errors_log_request_context( + caplog, + monkeypatch, + stage, + context, + expected_id, +): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + ws = FakeWebSocket([json.dumps({"type": "session.created", **context})]) + with _make_client() as client: + with client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: ws, + ) as session: + session.recv_event() + if stage == "disconnect": + session.create_response() + with pytest.raises(APIConnectionError) as caught: + session.recv_event() + assert caught.value.request_id is None + elif stage == "parse": + ws.inbox.append("private-payload not JSON") + with pytest.raises(json.JSONDecodeError): + session.recv_event() + else: + failure = WebSocketTimeoutException( + "private-payload Bearer private-key", + ) + + def fail(*args): + raise failure + + monkeypatch.setattr(ws, stage, fail) + expected_type = ( + WebSocketTimeoutException + if stage == "send" + else APITimeoutError + ) + with pytest.raises(expected_type) as caught: + if stage == "send": + session.commit() + else: + session.recv_event() + if stage == "send": + assert caught.value is failure + else: + assert caught.value.__cause__ is failure + assert caught.value.request_id is None + assert ws.closed + _assert_realtime_log(caplog, expected_id) + + +@pytest.mark.parametrize("kind", ["asr", "tts"]) +@pytest.mark.parametrize("explicit_id", [None, "failure-id"]) +def test_realtime_raw_asr_tts_failures_log_without_changing_events( + caplog, + kind, + explicit_id, +): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + error = { + "code": "InternalError", + "message": "private-payload", + "request_id": explicit_id, + } + event = ( + { + "type": "conversation.item.input_audio_transcription.failed", + "error": error, + } + if kind == "asr" + else { + "type": "response.done", + "response": { + "id": "response-id", + "status": "failed", + "status_details": {"error": error}, + }, + } + ) + ws = FakeWebSocket( + [ + json.dumps( + {"type": "session.created", "session": {"id": "session-id"}}, + ), + json.dumps( + { + "type": "response.created", + "response": {"id": "response-id"}, + }, + ), + json.dumps(event), + ], + ) + with _make_client() as client: + with client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: ws, + ) as session: + session.recv_event() + session.recv_event() + started = session.last_response_create_time + assert session.recv_event() == event + assert session.last_message == event + assert session.last_response_id == "response-id" + assert session.last_response_create_time == started + assert not ws.closed + _assert_realtime_log(caplog, explicit_id or "response-id") + + +def test_realtime_reconnect_drops_old_request_context(caplog): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + old = FakeWebSocket( + [json.dumps({"type": "session.created", "request_id": "old-id"})], + ) + fresh = FakeWebSocket( + [ + json.dumps( + { + "type": "error", + "event_id": "server-event", + "error": {"code": "InvalidParameter"}, + }, + ), + ], + ) + handshake = WebSocketBadStatusException( + "private-payload", + status_code=401, + resp_headers={}, + resp_body="bad", + ) + attempts = iter([old, handshake, fresh]) + + def connector(url, headers): + result = next(attempts) + if isinstance(result, Exception): + raise result + return result + + with _make_client() as client: + session = client.audio.realtime.connect( + model="omni", + connector=connector, + ) + with session: + session.recv_event() + with pytest.raises(WebSocketBadStatusException): + session.connect() + _assert_realtime_log(caplog, "unknown") + caplog.clear() + with session: + assert session.session_id is None + assert session.last_response_id is None + with pytest.raises(BadRequestError) as caught: + session.recv_event() + assert caught.value.request_id is None + _assert_realtime_log(caplog, "unknown") + + +def test_realtime_validation_and_proactive_close_do_not_log_errors(caplog): + caplog.set_level(logging.DEBUG, logger=_REALTIME_LOGGER) + ws = FakeWebSocket() + with _make_client() as client: + with pytest.raises(ValueError): + client.audio.realtime.connect(model="omni", timeout=0) + session = client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: ws, + ) + with pytest.raises(RuntimeError): + session.commit() + with session: + with pytest.raises(RuntimeError): + session.connect() + with pytest.raises(TypeError): + session.send_event({"type": "custom", "payload": object()}) + with pytest.raises(TypeError): + session.append_audio(123) + assert session.recv_event() is None + assert not _realtime_errors(caplog) + + +@pytest.mark.parametrize("field", ["session_id", "response_id"]) +def test_realtime_error_protocol_id_is_logging_only(caplog, field): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + event = { + "type": "error", + field: "protocol-id", + "event_id": "server-event", + "error": {"code": "InvalidParameter", "event_id": "client-event"}, + } + ws = FakeWebSocket([json.dumps(event)]) + with _make_client() as client: + with client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: ws, + ) as session: + with pytest.raises(BadRequestError) as caught: + session.recv_event() + assert caught.value.request_id is None + assert caught.value.body == event + assert session.session_id is None + assert session.last_response_id is None + _assert_realtime_log(caplog, "protocol-id") + + +@pytest.mark.parametrize("stage", ["handshake", "receive"]) +def test_realtime_successful_handshake_header_context( + caplog, + monkeypatch, + stage, +): + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + ws = FakeWebSocket(exc=WebSocketTimeoutException("private-payload")) + ws.headers = {"X-Request-ID": "", "X-DashScope-RequestId": "header-id"} + if stage == "handshake": + + def fail_timeout(seconds): + raise WebSocketTimeoutException("private-payload") + + monkeypatch.setattr(ws, "settimeout", fail_timeout) + with _make_client() as client: + session = client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: ws, + ) + with pytest.raises(APITimeoutError) as caught: + session.connect() + session.recv_event() + assert caught.value.request_id is None + assert isinstance(caught.value.__cause__, WebSocketTimeoutException) + assert ws.closed + _assert_realtime_log(caplog, "header-id") + + +@pytest.mark.parametrize("reconnect", [False, True]) +def test_realtime_retired_reader_does_not_log_current_request_errors( + caplog, + monkeypatch, + reconnect, +): + import threading + from concurrent.futures import ThreadPoolExecutor + + caplog.set_level(logging.ERROR, logger=_REALTIME_LOGGER) + entered, release = threading.Event(), threading.Event() + old = FakeWebSocket() + fresh = FakeWebSocket() + sockets = iter([old, fresh]) + + def delayed_recv(): + entered.set() + assert release.wait(2) + raise ConnectionResetError("private-payload") + + monkeypatch.setattr(old, "recv", delayed_recv) + with _make_client() as client: + session = client.audio.realtime.connect( + model="omni", + connector=lambda url, headers: next(sockets), + ) + session.connect() + with ThreadPoolExecutor(max_workers=1) as executor: + pending = executor.submit(session.recv_event) + try: + assert entered.wait(2) + session.close() + if reconnect: + session.connect() + release.set() + with pytest.raises(ConnectionResetError): + pending.result(timeout=2) + assert not fresh.closed + finally: + release.set() + session.close() + assert not _realtime_errors(caplog) diff --git a/tests/test_logging.py b/tests/test_logging.py index f3b88ae8..803d7072 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -8,11 +8,13 @@ from __future__ import annotations import io +import json import logging import httpx import pytest +from dashscope import AsyncClient, Client from dashscope import _logging as logging_mod from dashscope._logging import ( REDACTED, @@ -20,8 +22,18 @@ get_logger, redact_headers, redact_text, + request_id_from_headers, ) -from dashscope.transport import SyncTransport +from dashscope.errors import ( + APIConnectionError, + DashScopeError, + InternalServerError, + check_response, + response_json, +) +from dashscope.transport import AsyncTransport, SyncTransport +from dashscope.transport.sse import AsyncStream, Stream +from dashscope.types.common import ResponseBase, response_body, with_headers @pytest.fixture @@ -197,3 +209,501 @@ def test_child_logger_output_redacted_via_user_handler( messages = [r.getMessage() for r in caplog.records] assert any("Bearer " + REDACTED in m for m in messages) assert not any("sk-abcdef1234567890" in m for m in messages) + + +@pytest.mark.parametrize( + "name", + ["X-Request-ID", "X-DashScope-RequestId", "X-DashScope-Request-ID"], +) +def test_request_id_header_aliases(name): + assert request_id_from_headers({name: "req-alias"}) == "req-alias" + assert request_id_from_headers({name: [None, "", "req-list"]}) == ( + "req-list" + ) + + +def test_request_id_header_skips_empty_values(): + assert ( + request_id_from_headers( + {"x-request-id": " ", "X-DashScope-RequestId": "req-fallback"}, + ) + == "req-fallback" + ) + assert request_id_from_headers(None) is None + assert request_id_from_headers({"x-request-id": ""}) is None + + +@pytest.mark.parametrize("payload", ["not-json", "[]"]) +@pytest.mark.parametrize("request_id", [None, "req-http-parse"]) +def test_http_parse_failure_logs_request_id( + reset_logging, + caplog, + payload, + request_id, +): + response = httpx.Response( + 200, + text=payload, + headers={"X-DashScope-RequestId": request_id} if request_id else {}, + ) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(InternalServerError) as caught: + response_json(response) + assert caught.value.request_id == request_id + assert any( + "Failed to parse HTTP response" in record.message + and f"request_id={request_id or 'unknown'}" in record.message + for record in caplog.records + ) + + +@pytest.mark.parametrize("status", [200, 400]) +@pytest.mark.parametrize("raise_on_error", [True, False]) +@pytest.mark.parametrize("parse", [check_response, response_body]) +def test_http_service_error_logs_body_request_id( + reset_logging, + caplog, + status, + raise_on_error, + parse, +): + response = httpx.Response( + status, + json={"code": "InvalidParameter", "request_id": "req-body"}, + headers={"X-Request-ID": "req-header"}, + ) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + if raise_on_error: + with pytest.raises(DashScopeError) as caught: + parse(response, raise_on_error=True) + assert caught.value.request_id == "req-body" + else: + parse(response, raise_on_error=False) + assert len(caplog.records) == 1 + assert "request_id=req-body" in caplog.records[0].message + assert "req-header" not in caplog.records[0].message + + +@pytest.mark.parametrize("body_id", [None, "req-body"]) +def test_response_header_request_id_backfill(body_id): + response = httpx.Response( + 200, + json={}, + headers={"X-DashScope-Request-ID": "req-header"}, + ) + model = with_headers(ResponseBase(request_id=body_id), response) + assert model.request_id == (body_id or "req-header") + assert response_body(response)["request_id"] == "req-header" + assert with_headers(ResponseBase(), httpx.Response(200)).request_id is None + + +def _logging_sse_response(data): + return httpx.Response( + 200, + text=f"data: {data}\n\n", + headers={ + "Content-Type": "text/event-stream", + "X-DashScope-RequestId": "req-sse-header", + }, + ) + + +@pytest.mark.parametrize("data", ["not-json", "[]"]) +def test_sse_parse_failure_logs_request_id(reset_logging, caplog, data): + response = _logging_sse_response(data) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(DashScopeError) as caught: + list(Stream(response, lambda body: body)) + assert caught.value.request_id == "req-sse-header" + assert response.is_closed + assert len(caplog.records) == 1 + assert "request_id=req-sse-header" in caplog.records[0].message + + +@pytest.mark.asyncio +async def test_async_sse_parse_failure_logs_request_id(reset_logging, caplog): + response = _logging_sse_response("not-json") + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(DashScopeError) as caught: + async for _ in AsyncStream(response, lambda body: body): + pass + assert caught.value.request_id == "req-sse-header" + assert response.is_closed + assert len(caplog.records) == 1 + assert "request_id=req-sse-header" in caplog.records[0].message + + +def test_sse_service_error_logs_body_request_id(reset_logging, caplog): + response = _logging_sse_response( + json.dumps({"code": "Throttling", "request_id": "req-sse-body"}), + ) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(DashScopeError) as caught: + list(Stream(response, lambda body: body)) + assert caught.value.request_id == "req-sse-body" + assert len(caplog.records) == 1 + assert "request_id=req-sse-body" in caplog.records[0].message + assert response.is_closed + + +def test_sse_result_parse_failure_logs_body_request_id(reset_logging, caplog): + response = _logging_sse_response( + json.dumps({"request_id": "req-result-body"}), + ) + + def reject_result(body): + raise ValueError("result has an invalid shape") + + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(ValueError, match="invalid shape"): + list(Stream(response, reject_result)) + assert len(caplog.records) == 1 + assert "request_id=req-result-body" in caplog.records[0].message + assert response.is_closed + + +def test_pre_response_failure_log_has_no_request_id(reset_logging, caplog): + def fail(request): + raise httpx.ConnectError("connection refused", request=request) + + with httpx.Client(transport=httpx.MockTransport(fail)) as http_client: + transport = SyncTransport( + base_url="https://api.test.local", + http_client=http_client, + max_retries=0, + ) + with caplog.at_level(logging.DEBUG, logger=logging_mod.ROOT_NAME): + with pytest.raises(APIConnectionError) as caught: + transport.request("POST", "/chat") + assert caught.value.request_id is None + assert len(caplog.records) == 1 + assert "request_id" not in caplog.records[0].message + + +def test_static_sse_parser_does_not_log_request_id(reset_logging, caplog): + from dashscope.transport.sse import parse_event_data + + with caplog.at_level(logging.DEBUG, logger=logging_mod.ROOT_NAME): + with pytest.raises(DashScopeError): + parse_event_data("not-json") + assert caplog.records == [] + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_sse_later_failure_retains_stream_request_id( + reset_logging, + caplog, + async_mode, +): + response = _logging_sse_response('{"request_id": "req-first"}') + response = httpx.Response( + 200, + text=( + 'data: {"request_id": "req-first"}\n\n' + 'data: {"output": {}}\n\n' + "data: not-json\n\n" + ), + headers=response.headers, + ) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(DashScopeError) as caught: + if async_mode: + async for _ in AsyncStream(response, lambda body: body): + pass + else: + list(Stream(response, lambda body: body)) + assert caught.value.request_id == "req-first" + assert len(caplog.records) == 1 + assert "request_id=req-first" in caplog.records[0].message + assert response.is_closed + + +def test_job_event_log_uses_header_without_changing_error_contract( + reset_logging, + caplog, +): + from dashscope.resources._sse import _stream_error + + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + error = _stream_error( + None, + '{"code": "Throttling", "output": {"request_id": "inner-id"}}', + {"X-DashScope-Request-ID": "req-job-header"}, + ) + assert error.request_id is None + assert len(caplog.records) == 1 + assert "request_id=req-job-header" in caplog.records[0].message + + +class _BrokenSyncResponse(httpx.SyncByteStream): + def __iter__(self): + yield b"" + raise httpx.ReadError("read failed") + + +class _BrokenAsyncResponse(httpx.AsyncByteStream): + async def __aiter__(self): + yield b"" + raise httpx.ReadError("read failed") + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("streaming", [False, True]) +@pytest.mark.asyncio +async def test_response_read_failure_logs_request_id( + reset_logging, + caplog, + async_mode, + streaming, +): + response = httpx.Response( + 400, + stream=_BrokenAsyncResponse() if async_mode else _BrokenSyncResponse(), + headers={"X-DashScope-Request-ID": "req-read-error"}, + ) + mock = httpx.MockTransport(lambda request: response) + with caplog.at_level(logging.WARNING, logger=logging_mod.ROOT_NAME): + with pytest.raises(APIConnectionError) as caught: + if async_mode: + async with httpx.AsyncClient(transport=mock) as http_client: + transport = AsyncTransport( + base_url="https://api.test.local", + http_client=http_client, + max_retries=0, + ) + request = ( + transport.open_stream + if streaming + else transport.request + ) + await request("POST", "/chat") + else: + with httpx.Client(transport=mock) as http_client: + transport = SyncTransport( + base_url="https://api.test.local", + http_client=http_client, + max_retries=0, + ) + request = ( + transport.open_stream + if streaming + else transport.request + ) + request("POST", "/chat") + assert caught.value.request_id == "req-read-error" + assert response.is_closed + assert len(caplog.records) == 1 + assert "request_id=req-read-error" in caplog.records[0].message + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_sse_protocol_failure_logs_request_id( + reset_logging, + caplog, + async_mode, +): + from httpx_sse import SSEError + + response = httpx.Response( + 200, + text="not an event stream", + headers={ + "Content-Type": "application/json", + "X-DashScope-Request-ID": "req-sse-protocol", + }, + ) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(SSEError): + if async_mode: + async for _ in AsyncStream(response, lambda body: body): + pass + else: + list(Stream(response, lambda body: body)) + assert response.is_closed + assert len(caplog.records) == 1 + assert "request_id=req-sse-protocol" in caplog.records[0].message + + +@pytest.mark.parametrize("resource", ["fine_tunes", "deployments"]) +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_nonraising_job_event_keeps_malformed_code( + reset_logging, + caplog, + resource, + async_mode, +): + body = {"code": {"unexpected": "object"}} + response = httpx.Response( + 200, + text="event:error\ndata: " + json.dumps(body) + "\n\n", + headers={ + "Content-Type": "text/event-stream", + "X-DashScope-RequestId": "req-nonraising-job", + }, + ) + mock = httpx.MockTransport(lambda request: response) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + if async_mode: + async with httpx.AsyncClient(transport=mock) as http_client: + async with AsyncClient( + base_url="https://api.test.local", + http_client=http_client, + raise_on_error=False, + ) as client: + events = getattr(client, resource).stream_events("job-1") + result = [event async for event in events] + else: + with httpx.Client(transport=mock) as http_client: + with Client( + base_url="https://api.test.local", + http_client=http_client, + raise_on_error=False, + ) as client: + result = list( + getattr(client, resource).stream_events("job-1"), + ) + assert len(result) == 1 + assert result[0].is_error + assert result[0].output == body + assert response.is_closed + assert len(caplog.records) == 1 + assert "request_id=req-nonraising-job" in caplog.records[0].message + + +@pytest.mark.parametrize("parse", [check_response, response_body]) +@pytest.mark.parametrize("raise_on_error", [False, True]) +@pytest.mark.parametrize("header_id", [None, "safe-header-id"]) +def test_http_error_log_never_stringifies_object_request_id( + reset_logging, + caplog, + parse, + raise_on_error, + header_id, +): + invalid_id = {"credential": "private-payload"} + response = httpx.Response( + 400, + json={"code": "InvalidParameter", "request_id": invalid_id}, + headers={"X-DashScope-RequestId": header_id} if header_id else {}, + ) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + if raise_on_error: + with pytest.raises(DashScopeError) as caught: + parse(response) + assert caught.value.request_id == invalid_id + else: + parse(response, raise_on_error=False) + (record,) = caplog.records + assert f"request_id={header_id or 'unknown'}" in record.message + assert "private-payload" not in record.message + assert "credential" not in record.message + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize("fault", ["business", "parse", "result"]) +@pytest.mark.asyncio +async def test_sse_invalid_id_preserves_safe_log_context( + reset_logging, + caplog, + async_mode, + fault, +): + invalid_id = {"credential": "private-payload"} + body = {"request_id": invalid_id} + if fault == "business": + body["code"] = "InvalidParameter" + data = ( + 'data: {"request_id": "first-id"}\n\n' + + "data: " + + json.dumps(body) + + "\n\n" + ) + if fault == "parse": + data += "data: not-json\n\n" + response = httpx.Response( + 200, + text=data, + headers={ + "Content-Type": "text/event-stream", + "X-Request-ID": "header-id", + }, + ) + + def parse(body): + if fault == "result" and body.get("request_id") == invalid_id: + raise ValueError("rejected") + return body + + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises( + ValueError if fault == "result" else DashScopeError, + ): + if async_mode: + async for _ in AsyncStream(response, parse): + pass + else: + list(Stream(response, parse)) + (record,) = caplog.records + assert "request_id=first-id" in record.message + assert "private-payload" not in record.message + assert "credential" not in record.message + assert response.is_closed + + +@pytest.mark.parametrize("resource", ["fine_tunes", "deployments"]) +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_job_event_status_parse_failure_logs_and_closes( + reset_logging, + caplog, + resource, + async_mode, +): + response = httpx.Response( + 200, + text="status: private-payload\n", + headers={ + "Content-Type": "text/event-stream", + "X-DashScope-RequestId": "job-header-id", + }, + ) + mock = httpx.MockTransport(lambda request: response) + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + with pytest.raises(ValueError): + if async_mode: + async with httpx.AsyncClient(transport=mock) as http_client: + async with AsyncClient( + base_url="https://api.test.local", + http_client=http_client, + ) as client: + events = getattr(client, resource).stream_events("job") + async for _ in events: + pass + else: + with httpx.Client(transport=mock) as http_client: + with Client( + base_url="https://api.test.local", + http_client=http_client, + ) as client: + list(getattr(client, resource).stream_events("job")) + (record,) = caplog.records + assert "request_id=job-header-id" in record.message + assert "private-payload" not in record.message + assert response.is_closed + + +def test_job_business_log_never_stringifies_object_request_id( + reset_logging, + caplog, +): + from dashscope.resources._sse import _log_stream_error + + body = {"code": "Throttling", "request_id": {"secret": "private-payload"}} + with caplog.at_level(logging.ERROR, logger=logging_mod.ROOT_NAME): + _log_stream_error(json.dumps(body), {"X-Request-ID": "header-id"}) + (record,) = caplog.records + assert "request_id=header-id" in record.message + assert "private-payload" not in record.message diff --git a/tests/test_ws_connection.py b/tests/test_ws_connection.py index f018dd08..2026933c 100644 --- a/tests/test_ws_connection.py +++ b/tests/test_ws_connection.py @@ -5,6 +5,7 @@ # pylint: disable=protected-access import json +import logging import queue import threading @@ -78,10 +79,12 @@ def _event(name, task_id, payload=None, request_id="r"): ) -def _started(conn, task, **run_kwargs): +def _started(conn, task, request_id="r", **run_kwargs): """Feed task-started, then run the task (reader routes async).""" task_id = task.task_id - conn._ws.inbox.put(_event("task-started", task_id)) + conn._ws.inbox.put( + _event("task-started", task_id, request_id=request_id), + ) kwargs = { "task_group": "audio", "task": "asr", @@ -227,7 +230,7 @@ def test_socket_error_broadcasts_to_all_tasks(): assert conn.closed -def test_slow_consumer_queue_full_fails_only_that_task(): +def test_slow_consumer_queue_full_fails_only_that_task(ws_error_logs): """One stalled task must not tear down the shared connection. Overflowing t1's bounded receive queue fails t1 alone; t2 keeps @@ -259,6 +262,10 @@ def test_slow_consumer_queue_full_fails_only_that_task(): t1.recv() assert not conn.closed conn.close() + assert _ws_errors(ws_error_logs) == [ + "websocket receive queue failed request_id=r " + "error_type=WsQueueCapacityExceededError", + ] def test_peer_close_normalizes_to_api_connection_error(): @@ -348,7 +355,7 @@ def test_recv_timeout_raises(): conn.close() -def test_close_broadcasts_and_closes_socket(): +def test_close_broadcasts_and_closes_socket(ws_error_logs): conn, connects = _make_connection(timeout=5) task = conn.open_task() _started(conn, task) @@ -358,6 +365,7 @@ def test_close_broadcasts_and_closes_socket(): with pytest.raises(APIConnectionError, match="connection closed"): task.recv() conn.close() # idempotent + assert _ws_errors(ws_error_logs) == [] @pytest.fixture @@ -916,7 +924,7 @@ def fail_start(self): reader.join(timeout=1) -def test_independent_session_close_wakes_blocked_receiver(): +def test_independent_session_close_wakes_blocked_receiver(ws_error_logs): entered = threading.Event() class ObservedSocket(QueueWebSocket): @@ -956,3 +964,528 @@ def receive(): ws.close() receiver.join(timeout=1) assert not receiver.is_alive() + assert _ws_errors(ws_error_logs) == [] + + +@pytest.fixture +def ws_error_logs(caplog): + caplog.set_level(logging.ERROR, logger="dashscope._logging.transport.ws") + return caplog + + +def _ws_errors(caplog): + return [ + record.getMessage() + for record in caplog.records + if record.name == "dashscope._logging.transport.ws" + and record.levelno >= logging.ERROR + ] + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +@pytest.mark.parametrize( + "headers, body, expected_id", + [ + ({"X-ReQuEsT-Id": "header-id"}, None, "header-id"), + ({"X-DASHSCOPE-REQUESTID": "header-id"}, "{", "header-id"), + ({"x-DashScope-Request-Id": "header-id"}, "[]", "header-id"), + ({"x-request-id": "header-id"}, b"\xff", "header-id"), + ({"x-request-id": "header-id"}, "null", "header-id"), + ({"x-request-id": "header-id"}, {"request_id": "body-id"}, "body-id"), + ( + {"x-request-id": "header-id"}, + b'{"request_id": "body-id", "message": "private-body"}', + "body-id", + ), + ({"x-request-id": "header-id"}, '{"request_id": " "}', "header-id"), + ({"x-request-id": "header-id"}, '{"request_id": 42}', "header-id"), + ({"x-request-id": "header-id"}, "[" * 2000, "header-id"), + ( + {"X-Request-Id": "", "X-DashScope-RequestId": "alias-id"}, + None, + "alias-id", + ), + ({"x-request-id": ["", "list-id"]}, None, "list-id"), + ({"x-request-id": " "}, "private-non-json-body", "unknown"), + (None, '{"request_id": "body-only"}', "body-only"), + (None, None, "unknown"), + ], +) +def test_handshake_failure_logs_response_id_without_payload( + transport, + headers, + body, + expected_id, + ws_error_logs, +): + from websocket import WebSocketBadStatusException + + error = WebSocketBadStatusException( + "Bearer private-token Authorization Cookie private-body", + status_code=403, + resp_headers=headers, + resp_body=body, + ) + + def connector(url, actual_headers): + assert actual_headers["Authorization"] == "Bearer private-key" + raise error + + with pytest.raises(WebSocketBadStatusException) as caught: + transport(url=URL, connector=connector, api_key="private-key") + assert caught.value is error + messages = _ws_errors(ws_error_logs) + assert messages == [ + "websocket handshake failed request_id=" + + expected_id + + " error_type=WebSocketBadStatusException", + ] + assert not any( + secret in " ".join(messages) + for secret in ( + "private-token", + "private-key", + "private-body", + "Cookie", + ) + ) + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +def test_handshake_timeout_keeps_conversion_and_logs_id( + transport, + ws_error_logs, +): + from websocket import WebSocketTimeoutException + + error = WebSocketTimeoutException("timed out") + error.resp_headers = {"X-DashScope-Request-ID": "timeout-id"} + + def connector(url, headers): + raise error + + with pytest.raises(APITimeoutError) as caught: + transport(url=URL, connector=connector) + assert caught.value.__cause__ is error + assert caught.value.request_id == "timeout-id" + assert _ws_errors(ws_error_logs) == [ + "websocket handshake failed request_id=timeout-id " + "error_type=WebSocketTimeoutException", + ] + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +@pytest.mark.parametrize( + "missing_id", + [None, "", " ", {"audio": "private-audio"}], +) +def test_task_request_id_survives_events_without_server_id( + transport, + missing_id, + ws_error_logs, +): + ws = QueueWebSocket() + conn = transport(url=URL, connector=lambda url, headers: ws, timeout=0.5) + task = conn.open_task() if transport is WsConnection else conn + try: + _started(conn, task, request_id="server-id") + for name in ("result-generated", "task-finished"): + event = json.loads( + _event(name, task.task_id, request_id=missing_id), + ) + if missing_id is None: + event["header"].pop("request_id") + ws.inbox.put(json.dumps(event)) + task.recv_event() + assert task.request_id == "server-id" + assert task.finished + finally: + conn.close() + assert _ws_errors(ws_error_logs) == [] + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +@pytest.mark.parametrize( + "server_id", + ["failed-id", None, " ", {"audio": "private-audio"}], +) +@pytest.mark.parametrize("started", [False, True]) +def test_task_failed_logs_once_with_task_scoped_precedence( + transport, + server_id, + started, + ws_error_logs, +): + ws = QueueWebSocket() + conn = transport(url=URL, connector=lambda url, headers: ws, timeout=0.5) + task = conn.open_task() if transport is WsConnection else conn + try: + other = None + if transport is WsConnection: + other = conn.open_task() + _started(conn, other, request_id="other-id") + if started: + _started(conn, task, request_id="known-id") + expected_id = "known-id" if started else task.task_id + if isinstance(server_id, str) and server_id.strip(): + expected_id = server_id + ws.inbox.put( + json.dumps( + { + "header": { + "event": "task-failed", + "task_id": task.task_id, + "request_id": server_id, + "error_code": "InvalidParameter", + "error_message": ( + "private-error-message Bearer private-token" + ), + }, + "request_id": "wrong-body-id", + "payload": { + "text": "private-full-text", + "audio": "private-audio", + }, + }, + ), + ) + with pytest.raises(DashScopeError) as caught: + task.recv() + assert caught.value.code == "InvalidParameter" + assert caught.value.request_id == expected_id + assert task.request_id == expected_id + assert task.finished and task.recv() is None + assert _ws_errors(ws_error_logs) == [ + "websocket task-failed failed request_id=" + + expected_id + + " error_type=BadRequestError", + ] + if other is not None: + assert not conn.closed + other.continue_task({}) + ws.inbox.put( + _event("result-generated", other.task_id, request_id=None), + ) + other.recv_event() + assert other.request_id == "other-id" + finally: + conn.close() + assert len(_ws_errors(ws_error_logs)) == 1 + + +@pytest.mark.parametrize( + "raw, error_type", + [ + (None, APIConnectionError), + ("", APIConnectionError), + ('{"private-full-text":', json.JSONDecodeError), + ('{"payload": "private-full-text"}', WsEventError), + (OSError("private-error-message Bearer private-token"), OSError), + ], +) +def test_shared_receive_failure_logs_each_id_once_without_stamping_exception( + bounded_connection, + raw, + error_type, + ws_error_logs, +): + conn = bounded_connection() + first, second = conn.open_task(), conn.open_task() + _started(conn, first, request_id="first-id") + _started(conn, second, request_id="second-id") + # The latest ID is still queued when the connection fails; failure + # logging must use reader-side bookkeeping, not consumer progress. + conn._ws.inbox.put( + _event("result-generated", first.task_id, request_id="latest-id"), + ) + conn._ws.inbox.put( + _event("result-generated", first.task_id, request_id=None), + ) + conn._ws.inbox.put(raw) + conn._reader.join(timeout=1) + assert not conn._reader.is_alive() + assert first.recv_event()["header"]["request_id"] == "latest-id" + first.recv_event() + assert first.request_id == "latest-id" + caught_errors = [] + for task in (first, second): + for _ in range(2): + with pytest.raises(error_type) as caught: + task.recv() + caught_errors.append(caught.value) + assert all(error is caught_errors[0] for error in caught_errors) + assert getattr(caught_errors[0], "request_id", None) is None + if isinstance(raw, BaseException): + assert caught_errors[0] is raw + assert _ws_errors(ws_error_logs) == [ + "websocket receive failed request_id=latest-id error_type=" + + error_type.__name__, + "websocket receive failed request_id=second-id error_type=" + + error_type.__name__, + ] + conn.close() + assert len(_ws_errors(ws_error_logs)) == 2 + + +@pytest.mark.parametrize( + "raw, error_type", + [ + (None, APIConnectionError), + ("", APIConnectionError), + ('{"private-full-text":', json.JSONDecodeError), + ('{"payload": "private-full-text"}', WsEventError), + (OSError("private-error-message Bearer private-token"), OSError), + ], +) +def test_session_receive_failure_logs_known_id_and_preserves_error( + raw, + error_type, + ws_error_logs, +): + ws = QueueWebSocket() + session = WsSession( + url=URL, + connector=lambda url, headers: ws, + timeout=0.5, + ) + try: + _started(session, session, request_id="session-id") + ws.inbox.put(raw) + with pytest.raises(error_type) as caught: + session.recv() + if isinstance(raw, BaseException): + assert caught.value is raw + assert ws.closed + assert _ws_errors(ws_error_logs) == [ + "websocket receive failed request_id=session-id error_type=" + + error_type.__name__, + ] + with pytest.raises(WsStateError): + session.recv() + finally: + session.close() + assert len(_ws_errors(ws_error_logs)) == 1 + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +@pytest.mark.parametrize("action", ["run", "binary", "continue", "finish"]) +def test_socket_send_failure_logs_only_initiating_task( + transport, + action, + ws_error_logs, +): + error = OSError("private-error-message Bearer private-token") + + class BrokenSendSocket(QueueWebSocket): + fail_send = False + + def send(self, payload, opcode=1): + if self.fail_send: + raise error + super().send(payload, opcode=opcode) + + ws = BrokenSendSocket() + conn = transport(url=URL, connector=lambda url, headers: ws, timeout=0.5) + task = conn.open_task() if transport is WsConnection else conn + try: + other = None + if transport is WsConnection: + other = conn.open_task() + _started(conn, other, request_id="other-id") + if action != "run": + _started(conn, task, request_id="sender-id") + expected_id = task.task_id if action == "run" else "sender-id" + ws.fail_send = True + with pytest.raises(OSError) as caught: + if action == "run": + task.run_task( + task_group="audio", + task="asr", + function="recognition", + model="test", + ) + elif action == "binary": + task.send_binary(b"private-audio") + elif action == "continue": + task.continue_task({"text": "private-full-text"}) + else: + task.finish_task() + assert caught.value is error + assert ws.closed + if other is not None: + with pytest.raises(APIConnectionError): + other.recv() + assert _ws_errors(ws_error_logs) == [ + "websocket send failed request_id=" + + expected_id + + " error_type=OSError", + ] + finally: + conn.close() + assert len(_ws_errors(ws_error_logs)) == 1 + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +def test_receive_timeout_logs_task_id_without_changing_lifecycle( + transport, + ws_error_logs, +): + from websocket import WebSocketTimeoutException + + ws = QueueWebSocket() + conn = transport(url=URL, connector=lambda url, headers: ws, timeout=0.02) + task = conn.open_task() if transport is WsConnection else conn + try: + _started(conn, task, request_id="timeout-id") + if transport is WsSession: + ws.inbox.put(WebSocketTimeoutException("private-error-message")) + with pytest.raises(APITimeoutError): + task.recv() + assert _ws_errors(ws_error_logs) == [ + "websocket receive failed request_id=timeout-id " + "error_type=APITimeoutError", + ] + assert ws.closed is (transport is WsSession) + finally: + conn.close() + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +def test_invalid_arguments_states_and_explicit_close_do_not_log_errors( + transport, + ws_error_logs, +): + ws = QueueWebSocket() + conn = transport(url=URL, connector=lambda url, headers: ws, timeout=0.5) + task = conn.open_task() if transport is WsConnection else conn + try: + with pytest.raises(TypeError): + task.send_binary("not-bytes") + with pytest.raises(WsStateError): + task.continue_task({}) + _started(conn, task, request_id="normal-id") + with pytest.raises(TypeError): + task.continue_task({"not-json": object()}) + assert ws.closed # Retain preexisting serialization-error cleanup. + conn.close() + with pytest.raises(WsStateError): + task.send_binary(b"late") + finally: + conn.close() + assert _ws_errors(ws_error_logs) == [] + + +def test_shared_receive_failure_without_tasks_logs_unknown( + bounded_connection, + ws_error_logs, +): + conn = bounded_connection() + conn._ws.inbox.put(OSError("private-error-message")) + conn._reader.join(timeout=1) + assert not conn._reader.is_alive() + assert _ws_errors(ws_error_logs) == [ + "websocket receive failed request_id=unknown error_type=OSError", + ] + + +@pytest.mark.parametrize("transport", [WsSession, WsConnection]) +@pytest.mark.parametrize("failure", ["peer-close", "socket-timeout"]) +def test_library_receive_errors_keep_normalization_and_request_id_logs( + transport, + failure, + ws_error_logs, +): + from websocket import ( + WebSocketConnectionClosedException, + WebSocketTimeoutException, + ) + + ws = QueueWebSocket() + conn = transport(url=URL, connector=lambda url, headers: ws, timeout=0.5) + task = conn.open_task() if transport is WsConnection else conn + try: + _started(conn, task, request_id="normalized-id") + error = ( + WebSocketConnectionClosedException("private-error-message") + if failure == "peer-close" + else WebSocketTimeoutException("private-error-message") + ) + ws.inbox.put(error) + expected_type = ( + APIConnectionError if failure == "peer-close" else APITimeoutError + ) + with pytest.raises(expected_type): + task.recv() + assert ws.closed + assert _ws_errors(ws_error_logs) == [ + "websocket receive failed request_id=normalized-id error_type=" + + expected_type.__name__, + ] + finally: + conn.close() + assert len(_ws_errors(ws_error_logs)) == 1 + + +def test_shared_heartbeat_failure_logs_each_task_without_reader_duplicate( + ws_error_logs, +): + fail_ping = threading.Event() + + class BrokenPingSocket(QueueWebSocket): + def ping(self): + if fail_ping.is_set(): + raise OSError("private-error-message Bearer private-token") + + ws = BrokenPingSocket() + conn = WsConnection( + url=URL, + connector=lambda url, headers: ws, + timeout=0.5, + ping_interval=0.01, + ) + try: + first, second = conn.open_task(), conn.open_task() + _started(conn, first, request_id="first-id") + _started(conn, second, request_id="second-id") + fail_ping.set() + errors = [] + for task in (first, second): + with pytest.raises(APIConnectionError) as caught: + task.recv() + errors.append(caught.value) + assert errors[0] is errors[1] + assert errors[0].request_id is None + assert conn.closed and ws.closed + finally: + conn.close() + assert _ws_errors(ws_error_logs) == [ + "websocket heartbeat failed request_id=first-id " + "error_type=APIConnectionError", + "websocket heartbeat failed request_id=second-id " + "error_type=APIConnectionError", + ] + + +@pytest.mark.parametrize( + "transport, failure_timeout", + [(WsSession, 0.5), (WsConnection, 0.5), (WsConnection, None)], +) +def test_local_socket_initialization_failure_does_not_log_request_id( + transport, + failure_timeout, + ws_error_logs, +): + failure = OSError("local initialization failed") + + class BrokenTimeoutSocket(QueueWebSocket): + def settimeout(self, seconds): + if seconds == failure_timeout: + raise failure + super().settimeout(seconds) + + ws = BrokenTimeoutSocket() + with pytest.raises(OSError) as caught: + transport(url=URL, connector=lambda url, headers: ws, timeout=0.5) + assert caught.value is failure + assert ws.closed + assert _ws_errors(ws_error_logs) == []