wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve version format check

Browse Source
This commit is contained in:
mattie ruth backman
2026-03-31 12:24:11 -04:00
committed by Mattie Ruth
parent 565b9b961d
commit 3e255f3d21
2 changed files with 16 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/pipecat/processors/frameworks/rtvi/processor.py
View File

@@ -304,7 +304,10 @@ class RTVIProcessor(FrameProcessor):
version_error = None
if version:
try:
self._client_version = [int(v) for v in version.split(".")]
parts = [int(v) for v in version.split(".")]
if len(parts) != 3:
raise ValueError
self._client_version = parts
protocol_major = int(RTVI.PROTOCOL_VERSION.split(".")[0])
if self._client_version[0] != protocol_major:
version_error = f"RTVI version {version} is not compatible with server protocol {RTVI.PROTOCOL_VERSION}."

21
tests/test_rtvi_processor.py
View File

@@ -73,15 +73,18 @@ class TestRTVIClientReadyVersionHandling(unittest.IsolatedAsyncioTestCase):
self.assertIn("unknown", error_msg)
async def test_invalid_version_format_sends_error(self):
data = RTVI.ClientReadyData(
version="not-a-version",
about=RTVI.AboutClientData(library="test-client"),
)
await self._call_handle_client_ready(data)
self.processor._send_error_response.assert_called_once()
error_msg = self.processor._send_error_response.call_args[0][1]
self.assertIn("Invalid client version format", error_msg)
self.assertIn("not-a-version", error_msg)
bad_versions = ["not-a-version", "123", "1.2.3.0", "junk", "1.2"]
for version in bad_versions:
with self.subTest(version=version):
data = RTVI.ClientReadyData(
version=version,
about=RTVI.AboutClientData(library="test-client"),
)
await self._call_handle_client_ready(data)
self.processor._send_error_response.assert_called_once()
error_msg = self.processor._send_error_response.call_args[0][1]
self.assertIn("Invalid client version format", error_msg)
self.assertIn(version, error_msg)
async def test_error_message_includes_compatibility_warning(self):
"""All version errors should append the compatibility warning."""