Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions src/agents/realtime/audio_formats.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
RealtimeAudioFormats,
)

from .. import _debug
from ..logger import logger


Expand All @@ -25,6 +26,8 @@ def to_realtime_audio_format(
format = AudioPCMU(type="audio/pcmu")
elif input_audio_format in ["g711_alaw", "audio/pcma", "pcma"]:
format = AudioPCMA(type="audio/pcma")
elif _debug.DONT_LOG_MODEL_DATA:
logger.debug("Unknown input audio format")
else:
logger.debug("Unknown input_audio_format: %s", input_audio_format)
elif isinstance(input_audio_format, Mapping):
Expand All @@ -37,15 +40,21 @@ def to_realtime_audio_format(
elif rate is None:
pcm_rate = 24000
else:
logger.debug(
"Unknown pcm rate in input_audio_format mapping: %s", input_audio_format
)
if _debug.DONT_LOG_MODEL_DATA:
logger.debug("Unknown PCM rate in input audio format mapping")
else:
logger.debug(
"Unknown pcm rate in input_audio_format mapping: %s",
input_audio_format,
)
pcm_rate = 24000
format = AudioPCM(type="audio/pcm", rate=pcm_rate)
elif fmt_type == "audio/pcmu":
format = AudioPCMU(type="audio/pcmu")
elif fmt_type == "audio/pcma":
format = AudioPCMA(type="audio/pcma")
elif _debug.DONT_LOG_MODEL_DATA:
logger.debug("Unknown input audio format mapping")
else:
logger.debug("Unknown input_audio_format mapping: %s", input_audio_format)
else:
Expand Down
98 changes: 98 additions & 0 deletions tests/realtime/test_audio_formats_unit.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
import logging
from typing import Any

import pytest
from openai.types.realtime.realtime_audio_formats import AudioPCM, AudioPCMA, AudioPCMU

from agents import _debug
from agents.realtime.audio_formats import to_realtime_audio_format


Expand Down Expand Up @@ -51,3 +56,96 @@ def test_to_realtime_audio_format_from_mapping():
assert alaw.type == "audio/pcma"

assert to_realtime_audio_format({"type": "audio/unknown", "rate": 8000}) is None


@pytest.mark.parametrize("tool_data_redacted", [False, True])
@pytest.mark.parametrize(
("input_audio_format", "expected_message", "expected_type"),
[
("format-secret", "Unknown input audio format", None),
(
{"type": "audio/pcm", "rate": "rate-secret"},
"Unknown PCM rate in input audio format mapping",
AudioPCM,
),
(
{"type": "format-secret", "nested": "mapping-secret"},
"Unknown input audio format mapping",
None,
),
],
)
def test_to_realtime_audio_format_redacts_unknown_values(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
tool_data_redacted: bool,
input_audio_format: Any,
expected_message: str,
expected_type: type[AudioPCM] | None,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_data_redacted)
caplog.set_level(logging.DEBUG, logger="openai.agents")

result = to_realtime_audio_format(input_audio_format)

if expected_type is None:
assert result is None
else:
assert isinstance(result, expected_type)
record = caplog.records[-1]
assert record.msg == expected_message
assert record.args == ()
assert record.exc_info is None
assert record.exc_text is None
assert all(value is not input_audio_format for value in record.__dict__.values())
assert logging.Formatter().format(record) == expected_message


@pytest.mark.parametrize("tool_data_redacted", [False, True])
def test_to_realtime_audio_format_preserves_diagnostic_mapping(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
tool_data_redacted: bool,
) -> None:
input_audio_format = {"type": "format-secret", "nested": "mapping-secret"}
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_data_redacted)
caplog.set_level(logging.DEBUG, logger="openai.agents")

assert to_realtime_audio_format(input_audio_format) is None

record = caplog.records[-1]
assert record.msg == "Unknown input_audio_format mapping: %s"
assert record.args is input_audio_format
assert record.exc_info is None
assert record.exc_text is None
assert "format-secret" in logging.Formatter().format(record)
assert "mapping-secret" in logging.Formatter().format(record)


def test_to_realtime_audio_format_redaction_does_not_render_hostile_mapping(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
class HostileMapping(dict[str, object]):
def __str__(self) -> str:
raise AssertionError("redacted logging must not call __str__")

def __repr__(self) -> str:
raise AssertionError("redacted logging must not call __repr__")

input_audio_format = HostileMapping(type="unknown")
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
caplog.set_level(logging.DEBUG, logger="openai.agents")

assert to_realtime_audio_format(input_audio_format) is None

record = caplog.records[-1]
assert record.msg == "Unknown input audio format mapping"
assert record.args == ()
assert record.exc_info is None
assert record.exc_text is None
assert all(value is not input_audio_format for value in record.__dict__.values())
assert logging.Formatter().format(record) == "Unknown input audio format mapping"