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/camera_signaling.py b/roborock/camera_signaling.py new file mode 100644 index 00000000..16a7efb7 --- /dev/null +++ b/roborock/camera_signaling.py @@ -0,0 +1,598 @@ +"""Camera signaling models, ICE parsing utilities, and payload serialization for Roborock vacuums. + +This module provides data models and helpers for Roborock camera live view WebRTC +signaling, including TURN server mapping, ICE candidate parsing/formatting, +password hashing, and base64-JSON payload encoding/decoding. +""" + +from __future__ import annotations + +import base64 +import binascii +import hashlib +import json +import logging +import re +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any, Self + +from .data.containers import RoborockBase + +_LOGGER = logging.getLogger(__name__) + +__all__ = [ + "CameraIceCandidate", + "CameraQuality", + "CameraTurnServer", + "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-char (ALPHA / DIGIT / "+" / "/" / "-" / "_") +_ICE_FOUNDATION_PATTERN = re.compile(r"^[A-Za-z0-9+/_\-]{1,32}$") +_VALID_TCP_TYPES = frozenset({"active", "passive", "so"}) +_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.""" + if not isinstance(value, str): + raise TypeError(f"Expected str for camera quality, got {type(value).__name__}") + 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, + ) + + def to_dict(self) -> dict[str, Any]: + """Convert to standard WebRTC RTCIceServer configuration dictionary.""" + result: dict[str, Any] = {"urls": list(self.urls)} + if self.username is not None: + result["username"] = self.username + if self.credential is not None: + result["credential"] = self.credential + return result + + +@dataclass +class CameraIceCandidate: + """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, Any]: + """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() + + +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. + """ + if not isinstance(candidate_str, str): + raise TypeError(f"Expected str for ICE candidate, got {type(candidate_str).__name__}") + + 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'). + """ + if not isinstance(password, str): + raise TypeError(f"Camera password must be a str, got {type(password).__name__}") + + 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(f"Camera password must contain only ASCII digits, got {password!r}") + + return hashlib.md5(password_str.encode("utf-8"), usedforsecurity=False).hexdigest() + + +def encode_sdp_offer(sdp: str | dict[str, Any]) -> str: + """Package an SDP offer into a base64-encoded UTF-8 JSON payload for send_sdp_to_robot. + + Accepts an SDP description string or a dict containing an 'sdp' key. + Returns the base64-encoded string to be passed under the 'app_sdp' parameter. + """ + sdp_str: str + if isinstance(sdp, str): + if not sdp.strip(): + raise ValueError("SDP offer string cannot be empty") + sdp_str = sdp + elif isinstance(sdp, dict): + if "sdp" not in sdp or not isinstance(sdp["sdp"], str): + raise ValueError("SDP dictionary must contain a non-empty string 'sdp' key") + if not sdp["sdp"].strip(): + raise ValueError("SDP offer string cannot be empty") + offer_type = sdp.get("type", "offer") + if not isinstance(offer_type, str) or offer_type.lower() != "offer": + raise ValueError(f"SDP offer payload type must be 'offer', got {offer_type!r}") + sdp_str = sdp["sdp"] + else: + raise TypeError(f"Expected str or dict for SDP offer, got {type(sdp).__name__}") + + payload = {"sdp": sdp_str, "type": "offer"} + json_bytes = json.dumps(payload, separators=(",", ":")).encode("utf-8") + return base64.b64encode(json_bytes).decode("ascii") + + +def decode_sdp_answer(payload: str | dict[str, Any] | list[Any]) -> dict[str, str] | None: + """Unpackage a base64-encoded UTF-8 JSON SDP answer payload from get_device_sdp. + + Accepts the raw device response list/dict (e.g. [{'dev_sdp': '...'}]) or a base64 string. + Returns a dictionary with 'sdp' and 'type' keys, or None if the device returned 'retry'. + """ + raw_b64: str + 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 + if isinstance(payload, dict): + if "dev_sdp" in payload: + val = payload["dev_sdp"] + if not isinstance(val, str): + raise TypeError(f"Expected str for 'dev_sdp', got {type(val).__name__}") + raw_b64 = val + elif "sdp" in payload: + sdp_val = payload["sdp"] + if not isinstance(sdp_val, str) or not sdp_val.strip(): + raise ValueError(f"SDP dictionary missing or empty 'sdp' string: {payload}") + ans_type = payload.get("type", "answer") + if not isinstance(ans_type, str) or ans_type.lower() not in ("answer", "pranswer"): + raise ValueError(f"SDP answer has invalid 'type': {ans_type!r}") + return {"sdp": sdp_val, "type": ans_type.lower()} + else: + raise ValueError(f"Dictionary missing 'dev_sdp' or 'sdp' key: {payload}") + 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 + + try: + json_bytes = base64.b64decode(raw_b64, validate=True) + data = json.loads(json_bytes.decode("utf-8")) + except (binascii.Error, ValueError, json.JSONDecodeError, UnicodeDecodeError) as err: + raise ValueError(f"Failed to decode base64 JSON SDP answer: {err}") from err + + if not isinstance(data, dict): + raise ValueError(f"Decoded SDP answer JSON must be an object, got {type(data).__name__}") + + if "sdp" not in data or not isinstance(data["sdp"], str) or not data["sdp"].strip(): + raise ValueError(f"Decoded SDP answer missing or empty 'sdp' field: {data}") + + answer_type = data.get("type", "answer") + if not isinstance(answer_type, str) or answer_type.lower() not in ("answer", "pranswer"): + raise ValueError(f"Decoded SDP answer has invalid 'type': {answer_type!r}") + + return { + "sdp": data["sdp"], + "type": answer_type.lower(), + } + + +def encode_ice_candidate( + candidate: CameraIceCandidate | str | dict[str, Any], + sdp_mid: str | None = None, + sdp_mline_index: int | None = None, +) -> str: + """Package an ICE candidate into a base64-encoded UTF-8 JSON payload for send_ice_to_robot. + + Accepts a CameraIceCandidate, raw candidate string, or dict. + Returns the base64-encoded string to be passed under the 'app_ice' parameter. + """ + if isinstance(candidate, CameraIceCandidate): + # Override mid/mline if explicitly passed to encode_ice_candidate + payload = candidate.to_signaling_dict() + if sdp_mid is not None: + payload["sdpMid"] = str(sdp_mid) + if sdp_mline_index is not None: + if not isinstance(sdp_mline_index, int): + try: + sdp_mline_index = int(sdp_mline_index) + except (ValueError, TypeError) as err: + raise ValueError(f"Invalid sdpMLineIndex {sdp_mline_index!r}: must be an integer") from err + payload["sdpMLineIndex"] = sdp_mline_index + elif isinstance(candidate, dict): + cand_val = candidate.get("candidate") + if not isinstance(cand_val, str) or not cand_val.strip(): + raise ValueError(f"Candidate dict missing or invalid 'candidate' string: {candidate}") + candidate_str = cand_val.strip() + mid = sdp_mid if sdp_mid is not None else candidate.get("sdpMid") + mline = sdp_mline_index if sdp_mline_index is not None else candidate.get("sdpMLineIndex") + if not candidate_str.startswith("candidate:"): + candidate_str = f"candidate:{candidate_str}" + if mline is not None and not isinstance(mline, int): + try: + mline = int(mline) + except (ValueError, TypeError) as err: + raise ValueError(f"Invalid sdpMLineIndex {mline!r}: must be an integer") from err + payload = { + "candidate": candidate_str, + "sdpMid": str(mid) if mid is not None else "0", + "sdpMLineIndex": mline if mline is not None else 0, + } + elif isinstance(candidate, str): + candidate_str = candidate.strip() + if not candidate_str: + raise ValueError("ICE candidate string cannot be empty") + if not candidate_str.startswith("candidate:"): + candidate_str = f"candidate:{candidate_str}" + mline = sdp_mline_index + if mline is not None and not isinstance(mline, int): + try: + mline = int(mline) + except (ValueError, TypeError) as err: + raise ValueError(f"Invalid sdpMLineIndex {mline!r}: must be an integer") from err + payload = { + "candidate": candidate_str, + "sdpMid": str(sdp_mid) if sdp_mid is not None else "0", + "sdpMLineIndex": mline if mline is not None else 0, + } + else: + raise TypeError(f"Expected CameraIceCandidate, dict, or str, got {type(candidate).__name__}") + + json_bytes = json.dumps(payload, 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]: + """Unpackage base64-encoded UTF-8 JSON remote ICE candidates from get_device_ice. + + Accepts the raw device response list/dict (e.g. [{'dev_ice': ['...']}]) or a list of base64 strings. + Returns a list of parsed CameraIceCandidate objects. + If ignore_invalid is True, invalid candidates are logged and skipped rather than raising ValueError or TypeError. + """ + raw_list: list[Any] + if isinstance(payload, list): + if len(payload) == 1 and isinstance(payload[0], dict) and "dev_ice" in payload[0]: + dev_ice_val = payload[0]["dev_ice"] + if not isinstance(dev_ice_val, list): + raise TypeError(f"Expected list in 'dev_ice', got {type(dev_ice_val).__name__}") + raw_list = dev_ice_val + else: + raw_list = payload + elif isinstance(payload, dict): + if "dev_ice" in payload: + dev_ice_val = payload["dev_ice"] + if not isinstance(dev_ice_val, list): + raise TypeError(f"Expected list in 'dev_ice', got {type(dev_ice_val).__name__}") + raw_list = dev_ice_val + else: + 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 item in raw_list: + if isinstance(item, CameraIceCandidate): + candidates.append(item) + continue + if isinstance(item, dict): + cand_val = item.get("candidate") + if not isinstance(cand_val, str) or not cand_val.strip(): + if ignore_invalid: + _LOGGER.warning("Skipping candidate dict missing 'candidate' string: %s", item) + continue + raise ValueError(f"Candidate dict missing 'candidate' string: {item}") + mid = item.get("sdpMid") + mline = item.get("sdpMLineIndex") + try: + mline_int: int | None = None + if mline is not None: + mline_int = int(mline) + candidates.append( + parse_ice_candidate( + cand_val, + sdp_mid=str(mid) if mid is not None else None, + sdp_mline_index=mline_int, + ) + ) + except (ValueError, TypeError) as err: + if ignore_invalid: + _LOGGER.warning("Skipping invalid ICE candidate: %s", err) + continue + raise + continue + if isinstance(item, str): + item_str = item.strip() + if not item_str: + continue + cand_str_to_parse: str = item_str + sdp_mid_to_parse: str | None = None + sdp_mline_to_parse: int | None = None + try: + decoded_bytes = base64.b64decode(item_str, validate=True) + data = json.loads(decoded_bytes.decode("utf-8")) + if not isinstance(data, dict): + raise ValueError(f"Decoded candidate JSON must be a dict object, got {type(data).__name__}") + if "candidate" not in data or not isinstance(data["candidate"], str): + raise ValueError(f"Decoded candidate JSON missing 'candidate' string: {data}") + cand_str_to_parse = data["candidate"] + mid = data.get("sdpMid") + mline = data.get("sdpMLineIndex") + sdp_mid_to_parse = str(mid) if mid is not None else None + sdp_mline_to_parse = int(mline) if mline is not None else None + except (binascii.Error, ValueError, json.JSONDecodeError, UnicodeDecodeError, TypeError) as err: + # If this was clearly an attempted base64 JSON payload (starts with 'ey'), + # or valid base64 that decoded to non-dict, do not fall back to treating it as raw candidate + if item_str.startswith("ey") or ( + not isinstance(err, binascii.Error) and not isinstance(err, UnicodeDecodeError) + ): + if ignore_invalid: + _LOGGER.warning("Skipping invalid base64 JSON ICE candidate payload: %s", err) + continue + raise ValueError(f"Failed to decode base64 JSON ICE candidate: {err}") from err + cand_str_to_parse = item_str + + try: + candidates.append( + parse_ice_candidate( + cand_str_to_parse, + sdp_mid=sdp_mid_to_parse, + sdp_mline_index=sdp_mline_to_parse, + ) + ) + except (ValueError, TypeError) as err: + if ignore_invalid: + _LOGGER.warning("Skipping invalid ICE candidate: %s", err) + continue + raise + continue + + raise TypeError(f"Unexpected candidate item type: {type(item).__name__}") + + return candidates 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..ee90cc05 --- /dev/null +++ b/roborock/data/camera.py @@ -0,0 +1,27 @@ +"""Camera data models and utilities.""" + +from __future__ import annotations + +from roborock.camera_signaling import ( + CameraIceCandidate, + CameraQuality, + CameraTurnServer, + decode_ice_candidates, + decode_sdp_answer, + encode_ice_candidate, + encode_sdp_offer, + hash_camera_password, + parse_ice_candidate, +) + +__all__ = [ + "CameraIceCandidate", + "CameraQuality", + "CameraTurnServer", + "decode_ice_candidates", + "decode_sdp_answer", + "encode_ice_candidate", + "encode_sdp_offer", + "hash_camera_password", + "parse_ice_candidate", +] diff --git a/tests/test_camera_signaling.py b/tests/test_camera_signaling.py new file mode 100644 index 00000000..fdcbe859 --- /dev/null +++ b/tests/test_camera_signaling.py @@ -0,0 +1,731 @@ +"""Unit tests for camera signaling models, ICE parsing utilities, and payload serialization.""" + +from __future__ import annotations + +import base64 +import json + +import pytest + +from roborock.camera_signaling import ( + CameraIceCandidate, + CameraQuality, + CameraTurnServer, + decode_ice_candidates, + decode_sdp_answer, + encode_ice_candidate, + encode_sdp_offer, + hash_camera_password, + parse_ice_candidate, +) +from roborock.data import ( + CameraIceCandidate as DataCameraIceCandidate, +) +from roborock.data import ( + CameraQuality as DataCameraQuality, +) +from roborock.data import ( + CameraTurnServer as DataCameraTurnServer, +) +from roborock.data.camera import ( + CameraIceCandidate as CameraModIceCandidate, +) +from roborock.data.camera import ( + CameraQuality as CameraModQuality, +) +from roborock.data.camera import ( + CameraTurnServer as CameraModTurnServer, +) + +# ============================================================================ +# Re-export and import smoke tests +# ============================================================================ + + +def test_reexports_match() -> None: + """Verify camera signaling items are correctly re-exported across modules.""" + assert DataCameraQuality is CameraQuality + assert DataCameraTurnServer is CameraTurnServer + assert DataCameraIceCandidate is CameraIceCandidate + + assert CameraModQuality is CameraQuality + assert CameraModTurnServer is CameraTurnServer + assert CameraModIceCandidate is CameraIceCandidate + + +# ============================================================================ +# CameraQuality tests +# ============================================================================ + + +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") + + with pytest.raises(TypeError, match="Expected str"): + CameraQuality.from_str(123) # type: ignore[arg-type] + + +# ============================================================================ +# 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.to_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.to_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(TypeError, match="Expected dict inside TURN server list"): + CameraTurnServer.from_dict(["not_a_dict"]) # type: ignore[list-item] + + with pytest.raises(TypeError, match="Expected dict or list"): + CameraTurnServer.from_dict("not_a_dict_or_list") # type: ignore[arg-type] + + 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}) # type: ignore[arg-type] + + 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]}) # type: ignore[list-item] + + 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}) # type: ignore[dict-item] + + with pytest.raises(TypeError, match="Expected str for credential"): + CameraTurnServer.from_dict({"url": "turn:example.com", "pwd": 456}) # type: ignore[dict-item] + + +# ============================================================================ +# parse_ice_candidate tests +# ============================================================================ + + +def test_parse_ice_candidate_host_udp() -> None: + """Test parsing standard host UDP candidate.""" + cand_str = "candidate:4234997325 1 udp 2122260223 192.168.1.100 54321 typ host" + cand = parse_ice_candidate(cand_str, sdp_mid="0", sdp_mline_index=0) + + assert cand.foundation == "4234997325" + assert cand.component == 1 + assert cand.protocol == "udp" + assert cand.priority == 2122260223 + assert cand.ip == "192.168.1.100" + assert cand.port == 54321 + assert cand.candidate_type == "host" + assert cand.related_address is None + assert cand.related_port is None + assert cand.tcp_type is None + assert cand.sdp_mid == "0" + assert cand.sdp_mline_index == 0 + assert cand.to_sdp() == cand_str + assert str(cand) == cand_str + + +def test_parse_ice_candidate_srflx_with_raddr() -> None: + """Test parsing srflx candidate with raddr and rport.""" + cand_str = "candidate:123456 1 udp 1686052607 203.0.113.1 54321 typ srflx raddr 192.168.1.100 rport 54321" + cand = parse_ice_candidate(cand_str) + + assert cand.foundation == "123456" + assert cand.component == 1 + assert cand.protocol == "udp" + assert cand.priority == 1686052607 + assert cand.ip == "203.0.113.1" + assert cand.port == 54321 + assert cand.candidate_type == "srflx" + assert cand.related_address == "192.168.1.100" + assert cand.related_port == 54321 + assert cand.tcp_type is None + assert cand.to_sdp() == cand_str + + +def test_parse_ice_candidate_relay() -> None: + """Test parsing relay candidate.""" + cand_str = "candidate:789012 1 udp 33562367 198.51.100.2 3478 typ relay raddr 203.0.113.1 rport 54321" + cand = parse_ice_candidate(cand_str) + + assert cand.foundation == "789012" + assert cand.protocol == "udp" + assert cand.candidate_type == "relay" + assert cand.ip == "198.51.100.2" + assert cand.port == 3478 + assert cand.related_address == "203.0.113.1" + assert cand.related_port == 54321 + assert cand.to_sdp() == cand_str + + +def test_parse_ice_candidate_tcp_with_tcptype() -> None: + """Test parsing TCP candidate with tcptype.""" + cand_str = "candidate:345678 1 tcp 1518280447 192.168.1.100 9 typ host tcptype active" + cand = parse_ice_candidate(cand_str) + + assert cand.foundation == "345678" + assert cand.protocol == "tcp" + assert cand.candidate_type == "host" + assert cand.port == 9 + assert cand.tcp_type == "active" + assert cand.to_sdp() == cand_str + + +def test_parse_ice_candidate_ipv6() -> None: + """Test parsing IPv6 candidate.""" + cand_str = "candidate:1 1 udp 2122260223 2001:db8::1 50000 typ host" + cand = parse_ice_candidate(cand_str) + + assert cand.ip == "2001:db8::1" + assert cand.port == 50000 + assert cand.to_sdp() == cand_str + + +def test_parse_ice_candidate_mdns_hostname() -> None: + """Test parsing mDNS .local hostname candidate from modern browsers.""" + cand_str = "candidate:1 1 udp 2122260223 d79c09bf-323e-4610-8f92-5eb3fa7e4a1a.local 50000 typ host" + cand = parse_ice_candidate(cand_str) + + assert cand.ip == "d79c09bf-323e-4610-8f92-5eb3fa7e4a1a.local" + assert cand.to_sdp() == cand_str + + +def test_parse_ice_candidate_prefixes_and_extra_attrs() -> None: + """Test candidate with a= prefix, no prefix, CRLF, and extra SDP attributes like generation.""" + cand1 = parse_ice_candidate("a=candidate:1 1 udp 100 1.2.3.4 5000 typ host generation 0 network-id 1\r\n") + assert cand1.foundation == "1" + assert cand1.candidate_type == "host" + assert cand1.ip == "1.2.3.4" + assert cand1.extensions == {"generation": "0", "network-id": "1"} + + cand2 = parse_ice_candidate("1 1 udp 100 1.2.3.4 5000 typ host") + assert cand2.foundation == "1" + assert cand2.candidate_type == "host" + + +def test_parse_ice_candidate_numeric_and_lexical_bounds() -> None: + """Test strict RFC 8839 / RFC 8445 component, priority, and foundation bounds.""" + c1 = parse_ice_candidate("candidate:1 1 udp 100 1.1.1.1 200 typ host") + assert c1.component == 1 + c256 = parse_ice_candidate("candidate:1 256 udp 100 1.1.1.1 200 typ host") + assert c256.component == 256 + + with pytest.raises(ValueError, match="RFC 8839 range"): + parse_ice_candidate("candidate:1 0 udp 100 1.1.1.1 200 typ host") + with pytest.raises(ValueError, match="RFC 8839 range"): + parse_ice_candidate("candidate:1 257 udp 100 1.1.1.1 200 typ host") + + p1 = parse_ice_candidate("candidate:1 1 udp 1 1.1.1.1 200 typ host") + assert p1.priority == 1 + p_max = parse_ice_candidate("candidate:1 1 udp 2147483647 1.1.1.1 200 typ host") + assert p_max.priority == 2147483647 + + with pytest.raises(ValueError, match="RFC 8445 range"): + parse_ice_candidate("candidate:1 1 udp 0 1.1.1.1 200 typ host") + with pytest.raises(ValueError, match="RFC 8445 range"): + parse_ice_candidate("candidate:1 1 udp 2147483648 1.1.1.1 200 typ host") + + with pytest.raises(ValueError, match="ASCII decimal digits"): + parse_ice_candidate("candidate:1 +1 udp 100 1.1.1.1 200 typ host") + with pytest.raises(ValueError, match="ASCII decimal digits"): + parse_ice_candidate("candidate:1 1 udp 1_000 1.1.1.1 200 typ host") + + with pytest.raises(ValueError, match="Invalid ICE foundation"): + parse_ice_candidate(f"candidate:{'a' * 33} 1 udp 100 1.1.1.1 200 typ host") + with pytest.raises(ValueError, match="Invalid ICE foundation"): + parse_ice_candidate("candidate:invalid@char 1 udp 100 1.1.1.1 200 typ host") + + +def test_parse_ice_candidate_invariants() -> None: + """Test RFC 6544 TCP and RFC 8839 related address invariants.""" + with pytest.raises(ValueError, match="TCP ICE candidate must specify 'tcptype'"): + parse_ice_candidate("candidate:1 1 tcp 100 1.1.1.1 9 typ host") + + with pytest.raises(ValueError, match="Invalid ICE TCP type"): + parse_ice_candidate("candidate:1 1 tcp 100 1.1.1.1 9 typ host tcptype invalid") + + with pytest.raises(ValueError, match="must have both 'raddr' and 'rport' specified together"): + parse_ice_candidate("candidate:1 1 udp 100 1.1.1.1 200 typ srflx raddr 2.2.2.2") + + with pytest.raises(ValueError, match="must have both 'raddr' and 'rport' specified together"): + parse_ice_candidate("candidate:1 1 udp 100 1.1.1.1 200 typ srflx rport 5000") + + +def test_candidate_to_signaling_dict() -> None: + """Test converting CameraIceCandidate to signaling dict.""" + cand = CameraIceCandidate( + foundation="1", + component=1, + protocol="udp", + priority=100, + ip="1.2.3.4", + port=5000, + candidate_type="host", + sdp_mid="video", + sdp_mline_index=1, + ) + sig_dict = cand.to_signaling_dict() + assert sig_dict == { + "candidate": "candidate:1 1 udp 100 1.2.3.4 5000 typ host", + "sdpMid": "video", + "sdpMLineIndex": 1, + } + + +def test_candidate_mutation_reflected_in_to_sdp() -> None: + """Verify that mutating a parsed candidate updates to_sdp output (no stale raw cached).""" + cand = parse_ice_candidate("candidate:1 1 udp 100 192.168.1.100 5000 typ host") + assert cand.port == 5000 + assert cand.to_sdp() == "candidate:1 1 udp 100 192.168.1.100 5000 typ host" + + cand.port = 9999 + assert cand.to_sdp() == "candidate:1 1 udp 100 192.168.1.100 9999 typ host" + + +def test_candidate_equality_and_to_sdp_consistency() -> None: + """Verify candidates that compare equal produce identical to_sdp output.""" + cand1 = parse_ice_candidate("candidate:1 1 udp 100 1.2.3.4 5000 typ host") + cand2 = parse_ice_candidate("candidate:1 1 udp 100 1.2.3.4 5000 typ host") + assert cand1 == cand2 + assert cand1.to_sdp() == cand2.to_sdp() + + +# ============================================================================ +# hash_camera_password tests +# ============================================================================ + + +def test_hash_camera_password_valid() -> None: + """Test deriving lowercase hex MD5 of digits.""" + # 1234 -> 81dc9bdb52d04dc20036dbd8313ed055 + assert hash_camera_password("1234") == "81dc9bdb52d04dc20036dbd8313ed055" + # 9876 -> 912e79cd13c64069d91da65d62fbb78c + assert hash_camera_password("9876") == "912e79cd13c64069d91da65d62fbb78c" + # Leading zero in digits preserved + assert hash_camera_password("0123") == "eb62f6b9306db575c2d596b1279627a4" + + +def test_hash_camera_password_invalid() -> None: + """Test invalid password formats raise ValueError or TypeError.""" + with pytest.raises(ValueError, match="cannot be empty"): + hash_camera_password("") + + with pytest.raises(ValueError, match="only ASCII digits"): + hash_camera_password("12a4") + + # Full-width digits should be rejected + with pytest.raises(ValueError, match="only ASCII digits"): + hash_camera_password("１２３４") + + with pytest.raises(ValueError, match="only ASCII digits"): + hash_camera_password("password") + + with pytest.raises(TypeError, match="must be a str"): + hash_camera_password(None) # type: ignore[arg-type] + + with pytest.raises(TypeError, match="must be a str"): + hash_camera_password(1234) # type: ignore[arg-type] + + with pytest.raises(TypeError, match="must be a str"): + hash_camera_password(["1", "2", "3", "4"]) # type: ignore[arg-type] + + +# ============================================================================ +# SDP offer/answer packaging tests +# ============================================================================ + + +def test_encode_sdp_offer_and_decode_sdp_answer_roundtrip() -> None: + """Test round-trip SDP offer encoding and answer decoding.""" + raw_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" + encoded_offer = encode_sdp_offer(raw_sdp) + + decoded_json = json.loads(base64.b64decode(encoded_offer).decode("utf-8")) + assert decoded_json == {"sdp": raw_sdp, "type": "offer"} + + answer_payload = { + "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", + "type": "answer", + } + answer_b64 = base64.b64encode(json.dumps(answer_payload).encode("utf-8")).decode("ascii") + + res_str = decode_sdp_answer(answer_b64) + assert res_str == answer_payload + + res_dict = decode_sdp_answer({"dev_sdp": answer_b64}) + assert res_dict == answer_payload + + res_list = decode_sdp_answer([{"dev_sdp": answer_b64}]) + assert res_list == answer_payload + + +def test_encode_sdp_offer_from_dict() -> None: + """Test encoding SDP offer from dictionary.""" + offer_dict = {"sdp": "v=0\r\no=- 1 1 IN IP4 0.0.0.0", "type": "offer"} + b64_str = encode_sdp_offer(offer_dict) + decoded = json.loads(base64.b64decode(b64_str).decode("utf-8")) + assert decoded["sdp"] == offer_dict["sdp"] + assert decoded["type"] == "offer" + + +def test_decode_sdp_answer_retry() -> None: + """Test that retry string returns None.""" + assert decode_sdp_answer("retry") is None + assert decode_sdp_answer({"dev_sdp": "retry"}) is None + assert decode_sdp_answer([{"dev_sdp": "retry"}]) is None + + +def test_encode_sdp_offer_errors() -> None: + """Test error handling in encode_sdp_offer.""" + with pytest.raises(ValueError, match="cannot be empty"): + encode_sdp_offer("") + + with pytest.raises(ValueError, match="must contain a non-empty string 'sdp' key"): + encode_sdp_offer({}) + + with pytest.raises(ValueError, match="must contain a non-empty string 'sdp' key"): + encode_sdp_offer({"sdp": None}) # type: ignore[dict-item] + + with pytest.raises(ValueError, match="cannot be empty"): + encode_sdp_offer({"sdp": " "}) + + with pytest.raises(ValueError, match="type must be 'offer'"): + encode_sdp_offer({"sdp": "v=0...", "type": "answer"}) + + with pytest.raises(TypeError, match="Expected str or dict"): + encode_sdp_offer(12345) # type: ignore[arg-type] + + +def test_decode_sdp_answer_already_decoded_dict() -> None: + """Test decoding when already a dict with sdp key.""" + res = decode_sdp_answer({"sdp": "v=0...", "type": "answer"}) + assert res == {"sdp": "v=0...", "type": "answer"} + + with pytest.raises(ValueError, match="invalid 'type'"): + decode_sdp_answer({"sdp": "v=0...", "type": "offer"}) + + with pytest.raises(ValueError, match="missing or empty 'sdp'"): + decode_sdp_answer({"sdp": "", "type": "answer"}) + + +def test_decode_sdp_answer_errors_and_edge_cases() -> None: + """Test invalid payloads in decode_sdp_answer raise ValueError or TypeError.""" + with pytest.raises(ValueError, match="Empty response list"): + decode_sdp_answer([]) + + with pytest.raises(TypeError, match="Expected dict inside SDP answer list"): + decode_sdp_answer(["not_dict"]) # type: ignore[list-item] + + with pytest.raises(ValueError, match="missing 'dev_sdp' or 'sdp'"): + decode_sdp_answer({"invalid_key": "abc"}) + + with pytest.raises(TypeError, match="Expected str for 'dev_sdp'"): + decode_sdp_answer({"dev_sdp": 12345}) # type: ignore[dict-item] + + with pytest.raises(TypeError, match="Expected str, dict, or list"): + decode_sdp_answer(12345) # type: ignore[arg-type] + + with pytest.raises(ValueError, match="Failed to decode base64 JSON"): + decode_sdp_answer("invalid_base64!@#$") + + invalid_json_b64 = base64.b64encode(b"not json").decode("ascii") + with pytest.raises(ValueError, match="Failed to decode base64 JSON"): + decode_sdp_answer(invalid_json_b64) + + not_obj_b64 = base64.b64encode(b"[1, 2, 3]").decode("ascii") + with pytest.raises(ValueError, match="must be an object"): + decode_sdp_answer(not_obj_b64) + + missing_sdp_b64 = base64.b64encode(b'{"type": "answer"}').decode("ascii") + with pytest.raises(ValueError, match="missing or empty 'sdp' field"): + decode_sdp_answer(missing_sdp_b64) + + +# ============================================================================ +# ICE candidate packaging tests +# ============================================================================ + + +def test_encode_ice_candidate_and_decode_roundtrip() -> None: + """Test round-trip encoding and decoding of ICE candidates.""" + cand = CameraIceCandidate( + foundation="cand1", + component=1, + protocol="udp", + priority=2122260223, + ip="192.168.1.50", + port=55555, + candidate_type="host", + sdp_mid="0", + sdp_mline_index=0, + ) + + encoded = encode_ice_candidate(cand) + + obj = json.loads(base64.b64decode(encoded).decode("utf-8")) + assert obj["candidate"] == cand.to_sdp() + assert obj["sdpMid"] == "0" + assert obj["sdpMLineIndex"] == 0 + + decoded_list = decode_ice_candidates([encoded]) + assert len(decoded_list) == 1 + assert decoded_list[0] == cand + + decoded_dict = decode_ice_candidates({"dev_ice": [encoded]}) + assert len(decoded_dict) == 1 + assert decoded_dict[0] == cand + + decoded_resp = decode_ice_candidates([{"dev_ice": [encoded]}]) + assert len(decoded_resp) == 1 + assert decoded_resp[0] == cand + + +def test_encode_ice_candidate_from_str_and_dict() -> None: + """Test encoding from raw candidate string and dictionary.""" + raw_sdp = "candidate:1 1 udp 100 1.1.1.1 200 typ host" + b64_str = encode_ice_candidate(raw_sdp, sdp_mid="1", sdp_mline_index=1) + obj = json.loads(base64.b64decode(b64_str).decode("utf-8")) + assert obj["candidate"] == raw_sdp + assert obj["sdpMid"] == "1" + assert obj["sdpMLineIndex"] == 1 + + b64_str2 = encode_ice_candidate("1 1 udp 100 1.1.1.1 200 typ host") + obj2 = json.loads(base64.b64decode(b64_str2).decode("utf-8")) + assert obj2["candidate"] == raw_sdp + + dict_cand = {"candidate": raw_sdp, "sdpMid": "0", "sdpMLineIndex": 0} + b64_dict = encode_ice_candidate(dict_cand) + obj3 = json.loads(base64.b64decode(b64_dict).decode("utf-8")) + assert obj3["candidate"] == raw_sdp + + +def test_decode_ice_candidates_empty_or_multiple() -> None: + """Test decoding empty list and multiple candidates.""" + assert decode_ice_candidates([]) == [] + assert decode_ice_candidates({"dev_ice": []}) == [] + assert decode_ice_candidates([{"dev_ice": []}]) == [] + + cand1 = "candidate:1 1 udp 100 10.0.0.1 5000 typ host" + cand2 = "candidate:2 1 udp 200 10.0.0.2 5000 typ host" + enc1 = encode_ice_candidate(cand1, sdp_mid="0", sdp_mline_index=0) + enc2 = encode_ice_candidate(cand2, sdp_mid="0", sdp_mline_index=0) + + results = decode_ice_candidates([{"dev_ice": [enc1, enc2]}]) + assert len(results) == 2 + assert results[0].foundation == "1" + assert results[1].foundation == "2" + + +def test_decode_ice_candidates_errors() -> None: + """Test invalid payloads in decode_ice_candidates raise TypeError or ValueError.""" + with pytest.raises(TypeError, match="Expected str, list, or dict"): + decode_ice_candidates(12345) # type: ignore[arg-type] + + with pytest.raises(TypeError, match="Expected list in 'dev_ice'"): + decode_ice_candidates({"dev_ice": "not_a_list"}) + + with pytest.raises(TypeError, match="Unexpected candidate item type"): + decode_ice_candidates([12345]) # type: ignore[list-item] + + with pytest.raises(ValueError, match="missing 'candidate' string"): + decode_ice_candidates([{"sdpMid": "0"}]) + + +def test_encode_ice_candidate_errors() -> None: + """Test error handling in encode_ice_candidate.""" + with pytest.raises(ValueError, match="cannot be empty"): + encode_ice_candidate("") + + with pytest.raises(ValueError, match="missing or invalid 'candidate'"): + encode_ice_candidate({}) + + with pytest.raises(TypeError, match="Expected CameraIceCandidate, dict, or str"): + encode_ice_candidate(12345) # type: ignore[arg-type] + + +def test_decode_ice_candidates_edge_cases() -> None: + """Test edge cases in decode_ice_candidates.""" + cand_str = "candidate:1 1 udp 100 1.2.3.4 5000 typ host" + res1 = decode_ice_candidates(cand_str) + assert len(res1) == 1 + assert res1[0].foundation == "1" + + res2 = decode_ice_candidates({"candidate": cand_str, "sdpMid": "0", "sdpMLineIndex": 0}) + assert len(res2) == 1 + assert res2[0].foundation == "1" + + existing_cand = parse_ice_candidate(cand_str) + res3 = decode_ice_candidates([existing_cand]) + assert res3 == [existing_cand] + + res4 = decode_ice_candidates([{"candidate": cand_str, "sdpMid": "video", "sdpMLineIndex": 1}]) + assert len(res4) == 1 + assert res4[0].sdp_mid == "video" + assert res4[0].sdp_mline_index == 1 + + res5 = decode_ice_candidates(["", " "]) + assert res5 == [] + + res6 = decode_ice_candidates([cand_str]) + assert len(res6) == 1 + assert res6[0].foundation == "1" + + bad_base64_json = "eyJjYW5kaWRhdGUiOiA=" + with pytest.raises(ValueError, match="Failed to decode base64 JSON"): + decode_ice_candidates([bad_base64_json]) + + res7 = decode_ice_candidates([bad_base64_json, cand_str], ignore_invalid=True) + assert len(res7) == 1 + assert res7[0].foundation == "1" + + +def test_parse_ice_candidate_dangling_token_rejected() -> None: + """Test candidate with an odd trailing token raises ValueError.""" + with pytest.raises(ValueError, match="odd number of tokens"): + parse_ice_candidate("candidate:1 1 udp 100 1.2.3.4 5000 typ host dangling_token") + + +def test_encode_ice_candidate_mline_string_and_errors() -> None: + """Test sdp_mline_index coercion and error handling.""" + raw_sdp = "candidate:1 1 udp 100 1.1.1.1 200 typ host" + b64_str = encode_ice_candidate(raw_sdp, sdp_mline_index="1") # type: ignore[arg-type] + obj = json.loads(base64.b64decode(b64_str).decode("utf-8")) + assert obj["sdpMLineIndex"] == 1 + + with pytest.raises(ValueError, match="Invalid sdpMLineIndex"): + encode_ice_candidate(raw_sdp, sdp_mline_index="not_a_number") # type: ignore[arg-type] + + +def test_decode_ice_candidates_ignore_invalid_variations(caplog: pytest.LogCaptureFixture) -> None: + """Test ignore_invalid handles bad candidate dicts, bad strings, and TypeError from malformed mline.""" + res1 = decode_ice_candidates([{"sdpMid": "0"}], ignore_invalid=True) + assert res1 == [] + + res2 = decode_ice_candidates([{"candidate": "not a candidate"}], ignore_invalid=True) + assert res2 == [] + + res3 = decode_ice_candidates(["not a candidate"], ignore_invalid=True) + assert res3 == [] + + # TypeError from non-int coercible sdpMLineIndex (e.g. list) + res4 = decode_ice_candidates( + [{"candidate": "candidate:1 1 udp 100 1.2.3.4 5000 typ host", "sdpMLineIndex": [1]}], + ignore_invalid=True, + ) + assert res4 == [] + + # Valid base64 that decodes to non-dict JSON (e.g. JSON list) + non_dict_json_b64 = base64.b64encode(b"[1, 2, 3]").decode("ascii") + with pytest.raises(ValueError, match="Failed to decode base64 JSON"): + decode_ice_candidates([non_dict_json_b64]) + + res5 = decode_ice_candidates([non_dict_json_b64], ignore_invalid=True) + assert res5 == [] + + # Base64 JSON missing candidate key + no_cand_json = base64.b64encode(b'{"sdpMid": "0"}').decode("ascii") + with pytest.raises(ValueError, match="Failed to decode base64 JSON"): + decode_ice_candidates([no_cand_json]) + + # Check that caplog captured warnings + assert any("Skipping" in record.message for record in caplog.records)