diff --git a/docs/CAMERA_PROTOCOL.md b/docs/CAMERA_PROTOCOL.md index 756c256b..298e7937 100644 --- a/docs/CAMERA_PROTOCOL.md +++ b/docs/CAMERA_PROTOCOL.md @@ -141,7 +141,7 @@ See also [V1_API_COMMANDS.md](V1_API_COMMANDS.md) for general V1 device RPC refe | `check_homesec_password` | `{"password": ""}` | `["ok"]` | Validates the camera security PIN/pattern hash. | | `start_camera_preview` | `{"client_id": "...", "quality": "HD", "password": ""}` | `["ok"]` | Activates camera preview mode on the robot. | | `stop_camera_preview` | `[]` | `["ok"]` | Deactivates camera preview and powers down the camera sensor. | -| `switch_video_quality` | `{"quality": "HD"}` or `{"quality": "SD"}` | `["ok"]` | Switches streaming resolution (`HD` or `SD`). | +| `switch_video_quality` | `{"quality": "HD"}` / `{"quality": "SD"}` / `{"quality": "FHD"}` / `{"quality": "auto"}` | `["ok"]` | Switches streaming resolution (`HD`, `SD`, `FHD`, or `auto`). | ### WebRTC Signaling diff --git a/roborock/data/__init__.py b/roborock/data/__init__.py index 121f8b3c..82be80c4 100644 --- a/roborock/data/__init__.py +++ b/roborock/data/__init__.py @@ -2,6 +2,7 @@ from .b01_q7 import * from .b01_q10 import * +from .camera import * from .code_mappings import * from .containers import * from .dyad import * diff --git a/roborock/data/camera.py b/roborock/data/camera.py new file mode 100644 index 00000000..2ea63eb2 --- /dev/null +++ b/roborock/data/camera.py @@ -0,0 +1,158 @@ +"""Data models for Roborock camera live view.""" + +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any, Self + +from .containers import RoborockBase + +__all__ = ["CameraIceCandidate", "CameraQuality", "CameraSdpDescription", "CameraTurnServer"] + +_VALID_TURN_SCHEMES = ("turn:", "turns:", "stun:", "stuns:") + + +class CameraQuality(StrEnum): + """Camera video streaming resolution setting.""" + + HD = "HD" + SD = "SD" + FHD = "FHD" + AUTO = "auto" + + @classmethod + def from_str(cls, value: str) -> Self: + """Parse CameraQuality case-insensitively.""" + normalized = value.strip() + # "auto" is lowercase in the Roborock app enum, while HD/SD/FHD are uppercase + for member in cls: + if member.value.lower() == normalized.lower(): + return member + valid = ", ".join(repr(m.value) for m in cls) + raise ValueError(f"Invalid camera quality: {value!r}. Expected one of {valid}.") + + +@dataclass +class CameraTurnServer(RoborockBase): + """Ephemeral TURN server credentials for WebRTC peer connection relay.""" + + urls: list[str] + username: str | None = None + credential: str | None = field(default=None, repr=False) + + @classmethod + def from_dict(cls, data: dict[str, Any] | list[dict[str, Any]]) -> Self: + """Create CameraTurnServer from get_turn_server response or dict. + + Handles both device RPC responses (single-item list or dict with 'url', 'user', 'pwd') + and standard WebRTC configuration dicts ('urls', 'username', 'credential'). + """ + if isinstance(data, list): + if not data: + raise ValueError("Empty TURN server data list") + item = data[0] + if not isinstance(item, dict): + raise TypeError(f"Expected dict inside TURN server list, got {type(item).__name__}") + data = item + if not isinstance(data, dict): + raise TypeError(f"Expected dict or list of dicts for TURN server, got {type(data).__name__}") + + if "urls" in data: + urls_val = data["urls"] + elif "url" in data: + urls_val = data["url"] + else: + raise ValueError("Missing 'url' or 'urls' in TURN server response") + + urls: list[str] = [] + if isinstance(urls_val, str): + urls_clean = urls_val.strip() + if not urls_clean: + raise ValueError("TURN server URL cannot be empty") + urls = [urls_clean] + elif isinstance(urls_val, list): + if not urls_val: + raise ValueError("TURN server 'urls' list cannot be empty") + for u in urls_val: + if not isinstance(u, str) or not u.strip(): + raise ValueError(f"Each URL in 'urls' must be a non-empty string, got {u!r}") + urls.append(u.strip()) + else: + raise TypeError(f"Expected str or list for urls, got {type(urls_val).__name__}") + + # Validate URL scheme (stun:, stuns:, turn:, turns:) + for u in urls: + if not any(u.lower().startswith(scheme) for scheme in _VALID_TURN_SCHEMES): + raise ValueError(f"Invalid ICE server URL scheme in {u!r}. Expected turn:, turns:, stun:, or stuns:") + + user_val = data.get("username") if "username" in data else data.get("user") + pwd_val = data.get("credential") if "credential" in data else data.get("pwd") + + if user_val is not None and not isinstance(user_val, str): + raise TypeError(f"Expected str for username, got {type(user_val).__name__}") + if pwd_val is not None and not isinstance(pwd_val, str): + raise TypeError(f"Expected str for credential, got {type(pwd_val).__name__}") + + return cls( + urls=urls, + username=user_val, + credential=pwd_val, + ) + + +@dataclass +class CameraIceCandidate(RoborockBase): + """Parsed WebRTC ICE candidate representation (RFC 5245 / RFC 8839).""" + + foundation: str + component: int + protocol: str + priority: int + ip: str + port: int + candidate_type: str + related_address: str | None = None + related_port: int | None = None + tcp_type: str | None = None + sdp_mid: str | None = None + sdp_mline_index: int | None = None + extensions: dict[str, str] = field(default_factory=dict) + + def to_sdp(self) -> str: + """Format candidate as an SDP attribute line value (RFC 5245 / RFC 8839).""" + parts = [ + f"candidate:{self.foundation}", + str(self.component), + self.protocol.lower(), + str(self.priority), + self.ip, + str(self.port), + "typ", + self.candidate_type.lower(), + ] + if self.related_address is not None and self.related_port is not None: + parts.extend(["raddr", self.related_address, "rport", str(self.related_port)]) + if self.tcp_type is not None: + parts.extend(["tcptype", self.tcp_type.lower()]) + for ext_k, ext_v in self.extensions.items(): + parts.extend([ext_k, ext_v]) + return " ".join(parts) + + def to_signaling_dict(self) -> dict[str, str | int]: + """Convert to Roborock camera signaling dictionary format.""" + return { + "candidate": self.to_sdp(), + "sdpMid": self.sdp_mid if self.sdp_mid is not None else "0", + "sdpMLineIndex": self.sdp_mline_index if self.sdp_mline_index is not None else 0, + } + + def __str__(self) -> str: + """Return the SDP candidate string.""" + return self.to_sdp() + + +@dataclass +class CameraSdpDescription(RoborockBase): + """WebRTC session description exchanged during camera signaling.""" + + sdp: str + type: str diff --git a/roborock/protocols/camera_signaling.py b/roborock/protocols/camera_signaling.py new file mode 100644 index 00000000..a72a280d --- /dev/null +++ b/roborock/protocols/camera_signaling.py @@ -0,0 +1,311 @@ +"""V1 camera WebRTC signaling, ICE parsing, and payload serialization.""" + +import base64 +import hashlib +import json +import logging +import re +from typing import Any + +from roborock.data.camera import CameraIceCandidate, CameraSdpDescription +from roborock.exceptions import RoborockParsingException +from roborock.roborock_typing import RoborockCommand + +_LOGGER = logging.getLogger(__name__) + +__all__ = [ + "decode_ice_candidates", + "decode_sdp_answer", + "encode_ice_candidate", + "encode_sdp_offer", + "hash_camera_password", + "parse_ice_candidate", +] + +# RFC 5245 / RFC 8839: foundation is 1 to 32 ice characters. +_ICE_FOUNDATION_PATTERN = re.compile(r"^[A-Za-z0-9+/_\-]{1,32}$") +_VALID_TCP_TYPES = frozenset({"active", "passive", "so"}) + + +def parse_ice_candidate( + candidate_str: str, + sdp_mid: str | None = None, + sdp_mline_index: int | None = None, +) -> CameraIceCandidate: + """Parse a raw SDP candidate string into a CameraIceCandidate object. + + Handles standard candidate strings with optional 'a=' or 'candidate:' prefixes, + IPv4, IPv6, and FQDN/mDNS addresses, transport protocols, candidate types, and + optional extension attributes such as raddr, rport, tcptype, generation, etc. + """ + cand_clean = candidate_str.rstrip("\r\n").strip() + if not cand_clean: + raise ValueError("Empty ICE candidate string") + + tokens_str = cand_clean + if tokens_str.startswith("a="): + tokens_str = tokens_str[2:].strip() + if tokens_str.startswith("candidate:"): + tokens_str = tokens_str[10:].strip() + + tokens = tokens_str.split() + if len(tokens) < 8: + raise ValueError(f"Malformed ICE candidate string (expected at least 8 tokens): {candidate_str!r}") + + foundation = tokens[0] + if not _ICE_FOUNDATION_PATTERN.match(foundation): + raise ValueError(f"Invalid ICE foundation {foundation!r}: must be 1-32 ice characters") + + component_str = tokens[1] + if not (component_str.isascii() and component_str.isdigit()): + raise ValueError(f"Invalid ICE component {component_str!r}: must be ASCII decimal digits") + component = int(component_str) + if not (1 <= component <= 256): + raise ValueError(f"ICE component {component} out of valid RFC 8839 range [1, 256]") + + protocol = tokens[2].lower() + + priority_str = tokens[3] + if not (priority_str.isascii() and priority_str.isdigit()): + raise ValueError(f"Invalid ICE priority {priority_str!r}: must be ASCII decimal digits") + priority = int(priority_str) + if not (1 <= priority <= 2147483647): + raise ValueError(f"ICE priority {priority} out of valid RFC 8445 range [1, 2147483647]") + + ip = tokens[4] + + port_str = tokens[5] + if not (port_str.isascii() and port_str.isdigit()): + raise ValueError(f"Invalid ICE port {port_str!r}: must be ASCII decimal digits") + port = int(port_str) + if not (0 <= port <= 65535): + raise ValueError(f"ICE port {port} out of valid range [0, 65535]") + + typ_keyword = tokens[6].lower() + if typ_keyword != "typ": + raise ValueError( + f"Malformed ICE candidate string (expected 'typ' keyword at position 7, got {tokens[6]!r}): " + f"{candidate_str!r}" + ) + + cand_type = tokens[7].lower() + + related_address: str | None = None + related_port: int | None = None + tcp_type: str | None = None + extensions: dict[str, str] = {} + + if (len(tokens) - 8) % 2 != 0: + raise ValueError( + f"Malformed ICE candidate extension attributes: odd number of tokens ({len(tokens) - 8}) " + f"after candidate type: {candidate_str!r}" + ) + + i = 8 + while i < len(tokens): + attr_name = tokens[i].lower() + attr_val = tokens[i + 1] + if attr_name == "raddr": + related_address = attr_val + elif attr_name == "rport": + if not (attr_val.isascii() and attr_val.isdigit()): + raise ValueError(f"Invalid ICE related port {attr_val!r}: must be ASCII decimal digits") + related_port = int(attr_val) + if not (0 <= related_port <= 65535): + raise ValueError(f"ICE related port {related_port} out of valid range [0, 65535]") + elif attr_name == "tcptype": + tcp_type = attr_val.lower() + if tcp_type not in _VALID_TCP_TYPES: + raise ValueError(f"Invalid ICE TCP type {attr_val!r}: expected 'active', 'passive', or 'so'") + else: + extensions[tokens[i]] = attr_val + i += 2 + + # Invariants for related address / related port: + if (related_address is not None and related_port is None) or (related_address is None and related_port is not None): + raise ValueError("ICE candidate must have both 'raddr' and 'rport' specified together") + + # Invariants for TCP candidates: + if protocol == "tcp" and tcp_type is None: + raise ValueError("TCP ICE candidate must specify 'tcptype' attribute (RFC 6544)") + + return CameraIceCandidate( + foundation=foundation, + component=component, + protocol=protocol, + priority=priority, + ip=ip, + port=port, + candidate_type=cand_type, + related_address=related_address, + related_port=related_port, + tcp_type=tcp_type, + sdp_mid=sdp_mid, + sdp_mline_index=sdp_mline_index, + extensions=extensions, + ) + + +def hash_camera_password(password: str) -> str: + """Derive the lowercase hexadecimal MD5 digest of camera password digits. + + The Roborock camera password is a PIN or 3x3 pattern lock represented as + a sequence of numeric ASCII digits (e.g. '1234'). + """ + password_str = password.strip() + if not password_str: + raise ValueError("Camera password cannot be empty") + if not (password_str.isascii() and password_str.isdigit()): + raise ValueError("Camera password must contain only ASCII digits") + + return hashlib.md5(password_str.encode("utf-8"), usedforsecurity=False).hexdigest() + + +def encode_sdp_offer(sdp: str) -> str: + """Encode an SDP offer for the ``send_sdp_to_robot`` app_sdp parameter.""" + if not sdp.strip(): + raise ValueError("SDP offer string cannot be empty") + payload = CameraSdpDescription(sdp=sdp, type="offer") + json_bytes = json.dumps(payload.as_dict(), separators=(",", ":")).encode("utf-8") + return base64.b64encode(json_bytes).decode("ascii") + + +def decode_sdp_answer(payload: str | dict[str, Any] | list[Any]) -> CameraSdpDescription | None: + """Decode a get_device_sdp response, returning None for a pending ``retry``. + + Accept a base64 JSON string, its dev_sdp RPC envelope, or an already decoded + description. Preserve the SDP exactly, including CRLF line endings. + Malformed responses raise RoborockParsingException with redacted context. + """ + context = {"type": type(payload).__name__} + stage = "response envelope" + try: + if isinstance(payload, list): + if not payload: + raise ValueError("Empty response list for SDP answer") + item = payload[0] + if not isinstance(item, dict): + raise TypeError(f"Expected dict inside SDP answer list, got {type(item).__name__}") + payload = item + + data: Any + if isinstance(payload, dict) and "sdp" in payload and "dev_sdp" not in payload: + data = payload + else: + raw_b64: str + if isinstance(payload, dict): + if "dev_sdp" not in payload: + raise ValueError("Dictionary missing 'dev_sdp' or 'sdp' key") + raw = payload["dev_sdp"] + if not isinstance(raw, str): + raise TypeError(f"Expected str for 'dev_sdp', got {type(raw).__name__}") + raw_b64 = raw + elif isinstance(payload, str): + raw_b64 = payload + else: + raise TypeError(f"Expected str, dict, or list for SDP answer payload, got {type(payload).__name__}") + + raw_b64 = raw_b64.strip() + if raw_b64 == "retry": + return None + stage = "base64 JSON payload" + data = json.loads(base64.b64decode(raw_b64, validate=True).decode("utf-8")) + + stage = "SDP description fields" + if not isinstance(data, dict): + raise ValueError(f"Decoded SDP answer JSON must be an object, got {type(data).__name__}") + sdp = data.get("sdp") + if not isinstance(sdp, str) or not sdp.strip(): + raise ValueError("SDP answer missing or empty 'sdp' field") + answer_type = data.get("type", "answer") + if not isinstance(answer_type, str) or answer_type.lower() not in ("answer", "pranswer"): + raise ValueError("SDP answer has invalid 'type': expected answer or pranswer") + return CameraSdpDescription(sdp=sdp, type=answer_type.lower()) + except (ValueError, TypeError) as err: + raise RoborockParsingException( + trait_name="CameraSignaling", + command=RoborockCommand.GET_DEVICE_SDP, + payload=context, + inner_error=f"Invalid {stage} ({type(err).__name__})", + ) from err + + +def encode_ice_candidate(candidate: CameraIceCandidate) -> str: + """Encode a candidate for the ``send_ice_to_robot`` app_ice parameter. + + Set SDP media identifiers on the candidate before encoding. Missing identifiers + use the device defaults, sdpMid="0" and sdpMLineIndex=0. + """ + json_bytes = json.dumps(candidate.to_signaling_dict(), separators=(",", ":")).encode("utf-8") + return base64.b64encode(json_bytes).decode("ascii") + + +def decode_ice_candidates( + payload: str | list[Any] | dict[str, Any], + *, + ignore_invalid: bool = False, +) -> list[CameraIceCandidate]: + """Decode get_device_ice base64 JSON candidates or their dev_ice RPC envelope. + + Raw SDP belongs in parse_ice_candidate(). With ignore_invalid=True, skip + malformed candidate entries; malformed response envelopes still raise. + Malformed responses raise RoborockParsingException with redacted context. + """ + context: dict[str, str | int] = {"type": type(payload).__name__} + stage = "response envelope" + try: + if isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict) and "dev_ice" in payload[0]: + payload = payload[0] + + raw_list: list[Any] + if isinstance(payload, dict): + if "dev_ice" not in payload: + raise ValueError("ICE response dictionary missing 'dev_ice' list") + entries = payload["dev_ice"] + if not isinstance(entries, list): + raise TypeError(f"Expected list in 'dev_ice', got {type(entries).__name__}") + raw_list = entries + elif isinstance(payload, list): + raw_list = payload + elif isinstance(payload, str): + raw_list = [payload] + else: + raise TypeError(f"Expected str, list, or dict for ICE candidates payload, got {type(payload).__name__}") + + candidates: list[CameraIceCandidate] = [] + for index, item in enumerate(raw_list): + context["candidate_index"] = index + stage = "base64 JSON candidate" + try: + if not isinstance(item, str): + raise TypeError(f"Expected base64 JSON ICE candidate string, got {type(item).__name__}") + data = json.loads(base64.b64decode(item.strip(), validate=True).decode("utf-8")) + stage = "ICE candidate fields" + if not isinstance(data, dict): + raise ValueError("Decoded ICE candidate JSON must be an object") + candidate = data.get("candidate") + if not isinstance(candidate, str) or not candidate.strip(): + raise ValueError("Decoded ICE candidate JSON missing or empty 'candidate' string") + mid = data.get("sdpMid") + if mid is not None and not isinstance(mid, str): + raise TypeError("ICE candidate sdpMid must be a string or null") + mline = data.get("sdpMLineIndex") + if mline is not None and (not isinstance(mline, int) or isinstance(mline, bool)): + raise TypeError("ICE candidate sdpMLineIndex must be an integer or null") + if mline is not None and mline < 0: + raise ValueError("ICE candidate sdpMLineIndex cannot be negative") + stage = "SDP candidate string" + candidates.append(parse_ice_candidate(candidate, sdp_mid=mid, sdp_mline_index=mline)) + except (ValueError, TypeError) as err: + if not ignore_invalid: + raise + _LOGGER.warning("Skipping invalid ICE candidate at index %d (%s)", index, type(err).__name__) + + return candidates + except (ValueError, TypeError) as err: + raise RoborockParsingException( + trait_name="CameraSignaling", + command=RoborockCommand.GET_DEVICE_ICE, + payload=context, + inner_error=f"Invalid {stage} ({type(err).__name__})", + ) from err diff --git a/tests/data/test_camera.py b/tests/data/test_camera.py new file mode 100644 index 00000000..18d15c0d --- /dev/null +++ b/tests/data/test_camera.py @@ -0,0 +1,277 @@ +"""Tests for camera data models.""" + +import pytest + +from roborock.data.camera import CameraIceCandidate, CameraQuality, CameraSdpDescription, CameraTurnServer +from roborock.data.containers import RoborockBase + + +def test_camera_quality_values() -> None: + """Test CameraQuality enum values and string behavior.""" + assert CameraQuality.HD == "HD" + assert CameraQuality.SD == "SD" + assert CameraQuality.FHD == "FHD" + assert CameraQuality.AUTO == "auto" + assert str(CameraQuality.HD) == "HD" + assert CameraQuality("HD") is CameraQuality.HD + assert CameraQuality("SD") is CameraQuality.SD + assert CameraQuality("FHD") is CameraQuality.FHD + assert CameraQuality("auto") is CameraQuality.AUTO + + +@pytest.mark.parametrize( + ("input_str", "expected"), + [ + ("hd", CameraQuality.HD), + ("HD", CameraQuality.HD), + ("sd", CameraQuality.SD), + (" SD ", CameraQuality.SD), + ("fhd", CameraQuality.FHD), + ("FHD", CameraQuality.FHD), + ("auto", CameraQuality.AUTO), + ("AUTO", CameraQuality.AUTO), + (" auto ", CameraQuality.AUTO), + ], +) +def test_camera_quality_from_str(input_str: str, expected: CameraQuality) -> None: + """Test CameraQuality.from_str case insensitivity and whitespace trimming.""" + assert CameraQuality.from_str(input_str) is expected + + +def test_camera_quality_from_str_invalid() -> None: + """Test CameraQuality.from_str error handling for invalid values.""" + with pytest.raises(ValueError, match="Invalid camera quality"): + CameraQuality.from_str("4K") + + +# ============================================================================ +# CameraTurnServer tests +# ============================================================================ + + +def test_camera_turn_server_from_device_rpc_list() -> None: + """Test parsing get_turn_server response from single-item list.""" + payload = [{"url": "turn:turn-us.roborock.com:3478", "user": "roborock_user", "pwd": "secret_password"}] + turn = CameraTurnServer.from_dict(payload) + + assert turn.urls == ["turn:turn-us.roborock.com:3478"] + assert turn.username == "roborock_user" + assert turn.credential == "secret_password" + + # Verify credential redaction in repr + assert "secret_password" not in repr(turn) + + rtc_dict = turn.as_dict() + assert rtc_dict == { + "urls": ["turn:turn-us.roborock.com:3478"], + "username": "roborock_user", + "credential": "secret_password", + } + + +def test_camera_turn_server_from_device_rpc_dict() -> None: + """Test parsing get_turn_server response from bare dict.""" + payload = {"url": "turn:turn-eu.roborock.com:3478", "user": "eu_user", "pwd": "eu_password"} + turn = CameraTurnServer.from_dict(payload) + + assert turn.urls == ["turn:turn-eu.roborock.com:3478"] + assert turn.username == "eu_user" + assert turn.credential == "eu_password" + assert "eu_password" not in repr(turn) + + +def test_camera_turn_server_from_standard_webrtc_dict() -> None: + """Test parsing standard WebRTC RTCIceServer dict format.""" + payload = { + "urls": ["stun:stun.l.google.com:19302", "turns:turn.example.com:443"], + "username": "user123", + "credential": "pass123", + } + turn = CameraTurnServer.from_dict(payload) + + assert turn.urls == ["stun:stun.l.google.com:19302", "turns:turn.example.com:443"] + assert turn.username == "user123" + assert turn.credential == "pass123" + + +def test_camera_turn_server_without_credentials() -> None: + """Test TURN server with optional credentials absent.""" + payload = {"urls": "turn:turn.example.com:3478"} + turn = CameraTurnServer.from_dict(payload) + + assert turn.urls == ["turn:turn.example.com:3478"] + assert turn.username is None + assert turn.credential is None + assert turn.as_dict() == {"urls": ["turn:turn.example.com:3478"]} + + +def test_camera_turn_server_invalid_payloads() -> None: + """Test invalid payloads raise appropriate errors.""" + with pytest.raises(ValueError, match="Empty TURN server data list"): + CameraTurnServer.from_dict([]) + + with pytest.raises(ValueError, match="Missing 'url' or 'urls'"): + CameraTurnServer.from_dict({"user": "user_only"}) + + with pytest.raises(TypeError, match="Expected str or list for urls"): + CameraTurnServer.from_dict({"urls": 12345}) + + with pytest.raises(ValueError, match="non-empty string"): + CameraTurnServer.from_dict({"urls": [""]}) + + with pytest.raises(ValueError, match="non-empty string"): + CameraTurnServer.from_dict({"urls": [None]}) + + with pytest.raises(ValueError, match="Invalid ICE server URL scheme"): + CameraTurnServer.from_dict({"url": "http://invalidscheme.com"}) + + with pytest.raises(ValueError, match="TURN server URL cannot be empty"): + CameraTurnServer.from_dict({"url": ""}) + + with pytest.raises(ValueError, match="TURN server 'urls' list cannot be empty"): + CameraTurnServer.from_dict({"urls": []}) + + with pytest.raises(TypeError, match="Expected str for username"): + CameraTurnServer.from_dict({"url": "turn:example.com", "user": 123}) + + with pytest.raises(TypeError, match="Expected str for credential"): + CameraTurnServer.from_dict({"url": "turn:example.com", "pwd": 456}) + + +@pytest.mark.parametrize("model", [CameraIceCandidate, CameraQuality, CameraSdpDescription, CameraTurnServer]) +def test_camera_models_defined_in_data_module(model: type) -> None: + """Camera models are defined in their discoverable data module.""" + assert model.__module__ == "roborock.data.camera" + + +@pytest.mark.parametrize("include_optional_fields", [False, True]) +def test_camera_ice_candidate_model_roundtrip(include_optional_fields: bool) -> None: + """The inherited serializer preserves optional values and arbitrary ICE extensions.""" + candidate = CameraIceCandidate( + foundation="1", + component=1, + protocol="tcp" if include_optional_fields else "udp", + priority=100, + ip="1.2.3.4", + port=5000, + candidate_type="srflx" if include_optional_fields else "host", + related_address="10.0.0.1" if include_optional_fields else None, + related_port=4000 if include_optional_fields else None, + tcp_type="active" if include_optional_fields else None, + sdp_mid="video" if include_optional_fields else None, + sdp_mline_index=0 if include_optional_fields else None, + extensions={"generation": "0", "network-id": "1"} if include_optional_fields else {}, + ) + serialized = candidate.as_dict() + assert isinstance(candidate, RoborockBase) + assert serialized["candidateType"] == candidate.candidate_type + assert serialized["extensions"] == candidate.extensions + if include_optional_fields: + assert serialized["relatedAddress"] == "10.0.0.1" + assert serialized["relatedPort"] == 4000 + assert serialized["tcpType"] == "active" + assert serialized["sdpMid"] == "video" + assert serialized["sdpMlineIndex"] == 0 + else: + assert "relatedAddress" not in serialized + assert "relatedPort" not in serialized + assert "tcpType" not in serialized + assert "sdpMid" not in serialized + assert "sdpMlineIndex" not in serialized + assert CameraIceCandidate.from_dict(serialized) == candidate + + +def test_camera_ice_candidate_from_dict_converts_field_types() -> None: + """Inherited deserialization uses concrete annotations for nested and optional values.""" + candidate = CameraIceCandidate.from_dict( + { + "foundation": "1", + "component": "1", + "protocol": "udp", + "priority": "100", + "ip": "1.2.3.4", + "port": "5000", + "candidateType": "host", + "sdpMlineIndex": "0", + "extensions": {"generation": 0}, + } + ) + assert candidate.component == 1 + assert candidate.priority == 100 + assert candidate.port == 5000 + assert candidate.sdp_mline_index == 0 + assert candidate.extensions == {"generation": "0"} + + +@pytest.mark.parametrize("sdp_mid, sdp_mline_index", [(None, None), ("video", 1)]) +def test_camera_ice_candidate_signaling_dict(sdp_mid: str | None, sdp_mline_index: int | None) -> None: + """Signaling uses protocol field names and default media identifiers.""" + candidate = CameraIceCandidate( + foundation="1", + component=1, + protocol="UDP", + priority=100, + ip="1.2.3.4", + port=5000, + candidate_type="HOST", + sdp_mid=sdp_mid, + sdp_mline_index=sdp_mline_index, + ) + assert candidate.to_sdp() == "candidate:1 1 udp 100 1.2.3.4 5000 typ host" + assert str(candidate) == candidate.to_sdp() + assert candidate.to_signaling_dict() == { + "candidate": candidate.to_sdp(), + "sdpMid": sdp_mid if sdp_mid is not None else "0", + "sdpMLineIndex": sdp_mline_index if sdp_mline_index is not None else 0, + } + + +def test_camera_ice_candidate_sdp_optional_attributes() -> None: + """Formatting preserves related address, TCP type, and extension attributes.""" + candidate = CameraIceCandidate( + foundation="1", + component=1, + protocol="tcp", + priority=100, + ip="1.2.3.4", + port=5000, + candidate_type="srflx", + related_address="10.0.0.1", + related_port=4000, + tcp_type="ACTIVE", + extensions={"generation": "0", "network-id": "1"}, + ) + assert candidate.to_sdp() == ( + "candidate:1 1 tcp 100 1.2.3.4 5000 typ srflx " + "raddr 10.0.0.1 rport 4000 tcptype active generation 0 network-id 1" + ) + candidate.port = 9999 + assert "1.2.3.4 9999 typ srflx" in candidate.to_sdp() + + +@pytest.mark.parametrize("description_type", ["offer", "answer"]) +def test_camera_sdp_description_roundtrip(description_type: str) -> None: + """SDP results use the common typed model serialization contract.""" + description = CameraSdpDescription(sdp="v=0\r\n", type=description_type) + assert isinstance(description, RoborockBase) + assert description.as_dict() == {"sdp": "v=0\r\n", "type": description_type} + assert CameraSdpDescription.from_dict(description.as_dict()) == description + + +def test_camera_turn_server_as_dict_roundtrip() -> None: + """The inherited serializer exports standard WebRTC configuration without alias helpers.""" + server = CameraTurnServer( + urls=["turn:turn.example.com:3478"], + username="user", + credential="secret_password", + ) + serialized = server.as_dict() + assert serialized == { + "urls": ["turn:turn.example.com:3478"], + "username": "user", + "credential": "secret_password", + } + assert CameraTurnServer.from_dict(serialized) == server + assert "secret_password" not in repr(server) + serialized["urls"].append("stun:stun.example.com:3478") + assert server.urls == ["turn:turn.example.com:3478"] diff --git a/tests/protocols/test_camera_signaling.py b/tests/protocols/test_camera_signaling.py new file mode 100644 index 00000000..43eddb77 --- /dev/null +++ b/tests/protocols/test_camera_signaling.py @@ -0,0 +1,479 @@ +"""Tests for camera signaling parsing, hashing, and wire payload serialization.""" + +import base64 +import binascii +import json +from typing import Any + +import pytest + +from roborock.data.camera import CameraIceCandidate, CameraSdpDescription +from roborock.exceptions import RoborockException, RoborockParsingException +from roborock.protocols.camera_signaling import ( + decode_ice_candidates, + decode_sdp_answer, + encode_ice_candidate, + encode_sdp_offer, + hash_camera_password, + parse_ice_candidate, +) +from roborock.roborock_typing import RoborockCommand + +HOST_CANDIDATE = "candidate:4234997325 1 udp 2122260223 192.168.1.100 54321 typ host" + + +def encode_wire_json(payload: Any) -> str: + """Encode a dynamic device payload independently from the signaling encoder.""" + return base64.b64encode(json.dumps(payload).encode("utf-8")).decode("ascii") + + +def test_parse_ice_candidate_host_udp() -> None: + """Parse the standard host candidate and its media identifiers.""" + candidate = parse_ice_candidate(HOST_CANDIDATE, sdp_mid="video", sdp_mline_index=1) + + assert candidate == CameraIceCandidate( + foundation="4234997325", + component=1, + protocol="udp", + priority=2122260223, + ip="192.168.1.100", + port=54321, + candidate_type="host", + sdp_mid="video", + sdp_mline_index=1, + ) + assert candidate.to_sdp() == HOST_CANDIDATE + + +@pytest.mark.parametrize( + ("candidate_string", "expected"), + [ + ( + "candidate:123456 1 udp 1686052607 203.0.113.1 54321 typ srflx raddr 192.168.1.100 rport 54321", + CameraIceCandidate( + foundation="123456", + component=1, + protocol="udp", + priority=1686052607, + ip="203.0.113.1", + port=54321, + candidate_type="srflx", + related_address="192.168.1.100", + related_port=54321, + ), + ), + ( + "candidate:789012 1 udp 33562367 198.51.100.2 3478 typ relay raddr 203.0.113.1 rport 54321", + CameraIceCandidate( + foundation="789012", + component=1, + protocol="udp", + priority=33562367, + ip="198.51.100.2", + port=3478, + candidate_type="relay", + related_address="203.0.113.1", + related_port=54321, + ), + ), + ( + "candidate:345678 1 tcp 1518280447 192.168.1.100 9 typ host tcptype active", + CameraIceCandidate( + foundation="345678", + component=1, + protocol="tcp", + priority=1518280447, + ip="192.168.1.100", + port=9, + candidate_type="host", + tcp_type="active", + ), + ), + ], +) +def test_parse_ice_candidate_types(candidate_string: str, expected: CameraIceCandidate) -> None: + """Preserve related address and TCP attributes when parsing and serializing.""" + candidate = parse_ice_candidate(candidate_string) + + assert candidate == expected + assert candidate.to_sdp() == candidate_string + + +@pytest.mark.parametrize("address", ["2001:db8::1", "d79c09bf-323e-4610-8f92-5eb3fa7e4a1a.local"]) +def test_parse_ice_candidate_addresses(address: str) -> None: + """Support IPv6 and mDNS names without changing the address.""" + candidate_string = f"candidate:1 1 udp 2122260223 {address} 50000 typ host" + candidate = parse_ice_candidate(candidate_string) + + assert candidate.ip == address + assert candidate.to_sdp() == candidate_string + + +@pytest.mark.parametrize("prefix", ["", "candidate:", "a=candidate:"]) +def test_parse_ice_candidate_prefixes_and_extensions(prefix: str) -> None: + """Allow SDP prefixes and CRLF while preserving extension attributes.""" + candidate = parse_ice_candidate(f"{prefix}1 1 UDP 100 1.2.3.4 5000 typ HOST generation 0 network-id 1\r\n") + + assert candidate.foundation == "1" + assert candidate.protocol == "udp" + assert candidate.candidate_type == "host" + assert candidate.extensions == {"generation": "0", "network-id": "1"} + + +@pytest.mark.parametrize("component", [1, 256]) +@pytest.mark.parametrize("priority", [1, 2147483647]) +@pytest.mark.parametrize("port", [0, 65535]) +def test_parse_ice_candidate_numeric_bounds(component: int, priority: int, port: int) -> None: + """Accept the inclusive component, priority, and port limits.""" + candidate = parse_ice_candidate(f"candidate:1 {component} udp {priority} 1.1.1.1 {port} typ host") + + assert candidate.component == component + assert candidate.priority == priority + assert candidate.port == port + + +@pytest.mark.parametrize( + ("candidate_string", "message"), + [ + ("", "Empty ICE candidate"), + (" \r\n", "Empty ICE candidate"), + ("candidate:1 1 udp", "expected at least 8 tokens"), + ("candidate:1 0 udp 100 1.1.1.1 200 typ host", "RFC 8839 range"), + ("candidate:1 257 udp 100 1.1.1.1 200 typ host", "RFC 8839 range"), + ("candidate:1 +1 udp 100 1.1.1.1 200 typ host", "ASCII decimal digits"), + ("candidate:1 １ udp 100 1.1.1.1 200 typ host", "ASCII decimal digits"), + ("candidate:1 1 udp 0 1.1.1.1 200 typ host", "RFC 8445 range"), + ("candidate:1 1 udp 2147483648 1.1.1.1 200 typ host", "RFC 8445 range"), + ("candidate:1 1 udp 1_000 1.1.1.1 200 typ host", "ASCII decimal digits"), + (f"candidate:{'a' * 33} 1 udp 100 1.1.1.1 200 typ host", "Invalid ICE foundation"), + ("candidate:invalid@char 1 udp 100 1.1.1.1 200 typ host", "Invalid ICE foundation"), + ("candidate:1 1 udp 100 1.1.1.1 -1 typ host", "ASCII decimal digits"), + ("candidate:1 1 udp 100 1.1.1.1 65536 typ host", "port.*out of valid range"), + ("candidate:1 1 udp 100 1.1.1.1 200 other host", "expected 'typ' keyword"), + ("candidate:1 1 udp 100 1.1.1.1 200 typ host dangling_token", "odd number of tokens"), + ("candidate:1 1 tcp 100 1.1.1.1 9 typ host", "TCP ICE candidate must specify 'tcptype'"), + ("candidate:1 1 tcp 100 1.1.1.1 9 typ host tcptype invalid", "Invalid ICE TCP type"), + ("candidate:1 1 udp 100 1.1.1.1 200 typ srflx raddr 2.2.2.2", "must have both 'raddr' and 'rport'"), + ("candidate:1 1 udp 100 1.1.1.1 200 typ srflx rport 5000", "must have both 'raddr' and 'rport'"), + ("candidate:1 1 udp 100 1.1.1.1 200 typ srflx raddr 2.2.2.2 rport x", "ASCII decimal digits"), + ("candidate:1 1 udp 100 1.1.1.1 200 typ srflx raddr 2.2.2.2 rport 65536", "port.*out of valid range"), + ], +) +def test_parse_ice_candidate_invalid_values(candidate_string: str, message: str) -> None: + """Reject invalid candidate values and RFC invariant violations.""" + with pytest.raises(ValueError, match=message): + parse_ice_candidate(candidate_string) + + +@pytest.mark.parametrize( + ("password", "expected"), + [ + ("1234", "81dc9bdb52d04dc20036dbd8313ed055"), + ("9876", "912e79cd13c64069d91da65d62fbb78c"), + ("0123", "eb62f6b9306db575c2d596b1279627a4"), + (" 1234 ", "81dc9bdb52d04dc20036dbd8313ed055"), + ], +) +def test_hash_camera_password_valid(password: str, expected: str) -> None: + """Hash ASCII PIN digits, preserving leading zeros and trimming whitespace.""" + assert hash_camera_password(password) == expected + + +@pytest.mark.parametrize("password", ["", " ", "12a4", "１２３４", "password", "12 34"]) +def test_hash_camera_password_invalid(password: str) -> None: + """Reject empty passwords and non-ASCII PIN digits.""" + with pytest.raises(ValueError): + hash_camera_password(password) + + +def test_hash_camera_password_error_redacts_password() -> None: + """Keep invalid camera passwords out of error messages and logs.""" + password = "private-camera-password" + with pytest.raises(ValueError) as error: + hash_camera_password(password) + + assert password not in str(error.value) + + +def test_encode_sdp_offer_preserves_crlf() -> None: + """Package the exact SDP string and the offer type as base64 JSON.""" + sdp = "v=0\r\no=- 12345 2 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\n" + + assert json.loads(base64.b64decode(encode_sdp_offer(sdp))) == {"sdp": sdp, "type": "offer"} + + +@pytest.mark.parametrize("sdp", ["", " ", "\r\n"]) +def test_encode_sdp_offer_empty(sdp: str) -> None: + """Reject an offer without SDP content.""" + with pytest.raises(ValueError, match="cannot be empty"): + encode_sdp_offer(sdp) + + +@pytest.mark.parametrize("answer_type", ["answer", "pranswer", "ANSWER", "PRANSWER"]) +def test_decode_sdp_answer_supported_formats(answer_type: str) -> None: + """Return a typed description from each supported device response envelope.""" + sdp = "v=0\r\no=- 67890 2 IN IP4 127.0.0.1\r\ns=-\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\n" + description = {"sdp": sdp, "type": answer_type} + encoded = encode_wire_json(description) + expected = CameraSdpDescription(sdp=sdp, type=answer_type.lower()) + + assert decode_sdp_answer(encoded) == expected + assert decode_sdp_answer({"dev_sdp": encoded}) == expected + assert decode_sdp_answer([{"dev_sdp": encoded}]) == expected + assert decode_sdp_answer(description) == expected + + +def test_decode_sdp_answer_default_type() -> None: + """Use the answer type when it is absent from the device response.""" + assert decode_sdp_answer({"sdp": "v=0\r\n"}) == CameraSdpDescription(sdp="v=0\r\n", type="answer") + assert decode_sdp_answer(encode_wire_json({"sdp": "v=0\r\n"})) == CameraSdpDescription(sdp="v=0\r\n", type="answer") + + +@pytest.mark.parametrize("payload", ["retry", {"dev_sdp": "retry"}, [{"dev_sdp": "retry"}]]) +def test_decode_sdp_answer_retry(payload: str | dict[str, Any] | list[Any]) -> None: + """Leave retry handling explicit for the caller.""" + assert decode_sdp_answer(payload) is None + + +@pytest.mark.parametrize( + ("payload", "error"), + [ + ([], ValueError), + (["not_dict"], TypeError), + ({"invalid_key": "abc"}, ValueError), + ({"dev_sdp": 12345}, TypeError), + ("invalid_base64!@#$", binascii.Error), + (base64.b64encode(b"not json").decode("ascii"), json.JSONDecodeError), + (base64.b64encode(b"\xff").decode("ascii"), UnicodeDecodeError), + (encode_wire_json([1, 2, 3]), ValueError), + (encode_wire_json({"type": "answer"}), ValueError), + (encode_wire_json({"sdp": ""}), ValueError), + (encode_wire_json({"sdp": 123}), ValueError), + (encode_wire_json({"sdp": "v=0", "type": "offer"}), ValueError), + (encode_wire_json({"sdp": "v=0", "type": 1}), ValueError), + ({"sdp": "v=0", "type": "offer"}, ValueError), + ({"sdp": "", "type": "answer"}, ValueError), + ], +) +def test_decode_sdp_answer_invalid_payloads(payload: str | dict[str, Any] | list[Any], error: type[Exception]) -> None: + """Normalize malformed responses while preserving command context and the cause.""" + with pytest.raises(RoborockParsingException) as raised: + decode_sdp_answer(payload) + + assert isinstance(raised.value, RoborockException) + assert isinstance(raised.value.__cause__, error) + assert RoborockCommand.GET_DEVICE_SDP.name in str(raised.value) + assert "CameraSignaling" in str(raised.value) + assert f"'type': '{type(payload).__name__}'" in str(raised.value) + + +@pytest.mark.parametrize("encoded", [False, True]) +def test_decode_sdp_answer_error_redacts_response(encoded: bool, caplog: pytest.LogCaptureFixture) -> None: + """Do not put SDP credentials or malformed type values in public errors.""" + secret = "private-camera-secret" + description = {"sdp": f"v=0\r\na=ice-pwd:{secret}\r\n", "type": secret} + encoded_description = encode_wire_json(description) + payload: str | dict[str, Any] = encoded_description if encoded else description + + with pytest.raises(RoborockParsingException) as raised: + decode_sdp_answer(payload) + + assert isinstance(raised.value, RoborockException) + assert isinstance(raised.value.__cause__, ValueError) + assert RoborockCommand.GET_DEVICE_SDP.name in str(raised.value) + assert secret not in str(raised.value) + assert secret not in repr(raised.value) + assert secret not in caplog.text + assert encoded_description not in str(raised.value) + + +def test_encode_ice_candidate_roundtrip() -> None: + """Package a typed candidate and decode each supported response envelope.""" + candidate = parse_ice_candidate(HOST_CANDIDATE, sdp_mid="video", sdp_mline_index=1) + encoded = encode_ice_candidate(candidate) + + assert json.loads(base64.b64decode(encoded)) == { + "candidate": HOST_CANDIDATE, + "sdpMid": "video", + "sdpMLineIndex": 1, + } + assert decode_ice_candidates(encoded) == [candidate] + assert decode_ice_candidates([encoded]) == [candidate] + assert decode_ice_candidates({"dev_ice": [encoded]}) == [candidate] + assert decode_ice_candidates([{"dev_ice": [encoded]}]) == [candidate] + + +def test_encode_ice_candidate_default_media_identifiers() -> None: + """Use the protocol's default media identifiers when the model omits them.""" + encoded = encode_ice_candidate(parse_ice_candidate(HOST_CANDIDATE)) + + assert json.loads(base64.b64decode(encoded)) == { + "candidate": HOST_CANDIDATE, + "sdpMid": "0", + "sdpMLineIndex": 0, + } + + +def test_decode_ice_candidates_empty_or_multiple() -> None: + """Preserve order and support empty candidate responses.""" + assert decode_ice_candidates([]) == [] + assert decode_ice_candidates({"dev_ice": []}) == [] + assert decode_ice_candidates([{"dev_ice": []}]) == [] + + first = parse_ice_candidate(HOST_CANDIDATE, sdp_mid="0", sdp_mline_index=0) + second = parse_ice_candidate("candidate:2 1 udp 200 10.0.0.2 5000 typ host", sdp_mid="0", sdp_mline_index=0) + encoded = [encode_ice_candidate(first), encode_ice_candidate(second)] + + assert decode_ice_candidates([{"dev_ice": encoded}]) == [first, second] + + +def test_decode_ice_candidates_no_base64_prefix_heuristic() -> None: + """Decode valid base64 JSON even when its first bytes are whitespace.""" + wire_json = " \n" + json.dumps({"candidate": HOST_CANDIDATE, "sdpMid": "video", "sdpMLineIndex": 1}) + encoded = base64.b64encode(wire_json.encode("utf-8")).decode("ascii") + assert not encoded.startswith("ey") + + assert decode_ice_candidates(encoded) == [parse_ice_candidate(HOST_CANDIDATE, sdp_mid="video", sdp_mline_index=1)] + + +def test_decode_ice_candidates_optional_media_identifiers() -> None: + """Leave omitted media identifiers unset when receiving candidates.""" + encoded = encode_wire_json({"candidate": HOST_CANDIDATE}) + + assert decode_ice_candidates(encoded) == [parse_ice_candidate(HOST_CANDIDATE)] + encoded_nulls = encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMid": None, "sdpMLineIndex": None}) + assert decode_ice_candidates(encoded_nulls) == [parse_ice_candidate(HOST_CANDIDATE)] + + +@pytest.mark.parametrize( + ("payload", "error"), + [ + pytest.param("invalid_base64!@#$", binascii.Error, id="invalid-base64"), + pytest.param("YWJjZA", binascii.Error, id="missing-base64-padding"), + pytest.param(HOST_CANDIDATE, ValueError, id="raw-sdp"), + pytest.param("", ValueError, id="empty-string"), + pytest.param(" ", ValueError, id="whitespace-string"), + pytest.param(base64.b64encode(b"\xff").decode("ascii"), UnicodeDecodeError, id="invalid-utf8"), + pytest.param(base64.b64encode(b"not json").decode("ascii"), json.JSONDecodeError, id="invalid-json"), + pytest.param(base64.b64encode(b'{"candidate":').decode("ascii"), json.JSONDecodeError, id="truncated-json"), + pytest.param(encode_wire_json([1, 2, 3]), ValueError, id="json-list"), + pytest.param(encode_wire_json(None), ValueError, id="json-null"), + pytest.param(encode_wire_json({"sdpMid": "0"}), ValueError, id="missing-candidate"), + pytest.param(encode_wire_json({"candidate": None}), ValueError, id="null-candidate"), + pytest.param(encode_wire_json({"candidate": ""}), ValueError, id="empty-candidate"), + pytest.param(encode_wire_json({"candidate": "not a candidate"}), ValueError, id="malformed-candidate"), + pytest.param(encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMid": 1}), TypeError, id="numeric-mid"), + pytest.param(encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMid": ["0"]}), TypeError, id="list-mid"), + pytest.param( + encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMLineIndex": "1"}), TypeError, id="string-mline" + ), + pytest.param(encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMLineIndex": [1]}), TypeError, id="list-mline"), + pytest.param( + encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMLineIndex": True}), TypeError, id="bool-mline" + ), + pytest.param( + encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMLineIndex": 1.5}), TypeError, id="float-mline" + ), + pytest.param( + encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMLineIndex": -1}), ValueError, id="negative-mline" + ), + ], +) +def test_decode_ice_candidates_invalid_wire_data(payload: str, error: type[Exception]) -> None: + """Normalize invalid base64, UTF-8, JSON, schema, and candidate errors.""" + with pytest.raises(RoborockParsingException) as raised: + decode_ice_candidates(payload) + + assert isinstance(raised.value, RoborockException) + assert isinstance(raised.value.__cause__, error) + assert RoborockCommand.GET_DEVICE_ICE.name in str(raised.value) + assert "CameraSignaling" in str(raised.value) + assert "'type': 'str'" in str(raised.value) + assert "'candidate_index': 0" in str(raised.value) + + +@pytest.mark.parametrize( + "payload", + [ + {"candidate": HOST_CANDIDATE}, + [{"candidate": HOST_CANDIDATE, "sdpMid": "0", "sdpMLineIndex": 0}], + {"dev_ice": "not_a_list"}, + {"dev_ice": [123]}, + [123], + [parse_ice_candidate(HOST_CANDIDATE)], + [{"dev_ice": []}, {"dev_ice": []}], + ], +) +def test_decode_ice_candidates_rejects_non_wire_inputs(payload: dict[str, Any] | list[Any]) -> None: + """Do not accept unencoded models, candidate dictionaries, or malformed envelopes.""" + with pytest.raises(RoborockParsingException) as raised: + decode_ice_candidates(payload) + + assert isinstance(raised.value, RoborockException) + assert isinstance(raised.value.__cause__, (ValueError, TypeError)) + assert RoborockCommand.GET_DEVICE_ICE.name in str(raised.value) + assert f"'type': '{type(payload).__name__}'" in str(raised.value) + + +def test_decode_ice_candidates_ignore_invalid(caplog: pytest.LogCaptureFixture) -> None: + """Skip malformed candidate items and retain all valid candidates in order.""" + first = parse_ice_candidate(HOST_CANDIDATE, sdp_mid="video", sdp_mline_index=1) + second = parse_ice_candidate("candidate:2 1 udp 200 10.0.0.2 5000 typ host", sdp_mid="video", sdp_mline_index=1) + payload = [ + "not base64!", + encode_ice_candidate(first), + base64.b64encode(b"\xff").decode("ascii"), + encode_wire_json([1, 2, 3]), + encode_wire_json({"sdpMid": "0"}), + encode_wire_json({"candidate": "not a candidate"}), + encode_wire_json({"candidate": HOST_CANDIDATE, "sdpMLineIndex": [1]}), + encode_ice_candidate(second), + ] + + assert decode_ice_candidates({"dev_ice": payload}, ignore_invalid=True) == [first, second] + assert any("Skipping" in record.message for record in caplog.records) + + +def test_decode_ice_candidates_ignore_invalid_does_not_hide_envelope_errors() -> None: + """Candidate leniency does not allow an invalid RPC response structure.""" + with pytest.raises(RoborockParsingException) as raised: + decode_ice_candidates({"dev_ice": "not_a_list"}, ignore_invalid=True) + + assert isinstance(raised.value, RoborockException) + assert isinstance(raised.value.__cause__, TypeError) + assert RoborockCommand.GET_DEVICE_ICE.name in str(raised.value) + assert "'type': 'dict'" in str(raised.value) + + +def test_decode_ice_candidates_error_identifies_candidate_and_redacts_response( + caplog: pytest.LogCaptureFixture, +) -> None: + """Locate a malformed candidate without exposing its raw SDP in the error.""" + secret = "private-camera-secret" + invalid = encode_wire_json({"candidate": f"{HOST_CANDIDATE} {secret}"}) + payload = [encode_wire_json({"candidate": HOST_CANDIDATE}), invalid] + + with pytest.raises(RoborockParsingException) as raised: + decode_ice_candidates(payload) + + assert isinstance(raised.value, RoborockException) + assert isinstance(raised.value.__cause__, ValueError) + assert RoborockCommand.GET_DEVICE_ICE.name in str(raised.value) + assert "'type': 'list'" in str(raised.value) + assert "'candidate_index': 1" in str(raised.value) + assert secret not in str(raised.value) + assert secret not in repr(raised.value) + assert invalid not in str(raised.value) + assert secret not in caplog.text + + +def test_decode_ice_candidates_ignore_invalid_log_redacts_response(caplog: pytest.LogCaptureFixture) -> None: + """Keep skipped candidate diagnostics useful without logging candidate secrets.""" + secret = "private-camera-secret" + invalid = encode_wire_json({"candidate": f"{HOST_CANDIDATE} {secret}"}) + valid = encode_wire_json({"candidate": HOST_CANDIDATE}) + + assert decode_ice_candidates([invalid, valid], ignore_invalid=True) == [parse_ice_candidate(HOST_CANDIDATE)] + assert any("Skipping" in record.message for record in caplog.records) + assert secret not in caplog.text + assert invalid not in caplog.text