refactor!: drop run_device_action; the media controls are methods (0.2.0) #4
No files matched your search
@@ -1,13 +1,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import socket
|
||||
from typing import Any
|
||||
|
||||
from napalm.base.exceptions import ConnectionException
|
||||
|
||||
from napalm_device_types import FingerprintRule, MediaDriver, PortSpec
|
||||
from napalm_device_types.models import DeviceActionResultDict
|
||||
|
||||
"""NAPALM driver for Sonos speakers.
|
||||
|
||||
@@ -16,7 +14,6 @@ SoCo library, exposing NAPALM standard methods alongside Sonos-specific
|
||||
media-playback controls.
|
||||
"""
|
||||
|
||||
_VOLUME_RE = re.compile(r"^\d+$")
|
||||
|
||||
|
||||
class SonosDriver(MediaDriver):
|
||||
@@ -437,52 +434,3 @@ class SonosDriver(MediaDriver):
|
||||
except Exception:
|
||||
# Broad catch: speaker_info dict access is safe, but keep for safety
|
||||
return {}
|
||||
|
||||
def run_device_action(self, action: str) -> DeviceActionResultDict:
|
||||
"""Run a device action (play, pause, stop, next_track, previous_track, set_volume:N, toggle_mute)."""
|
||||
result: DeviceActionResultDict = {
|
||||
"success": False,
|
||||
"action": action,
|
||||
"message": "",
|
||||
}
|
||||
try:
|
||||
if action == "play":
|
||||
self.play()
|
||||
result["success"] = True
|
||||
result["message"] = "Playback started"
|
||||
elif action == "pause":
|
||||
self.pause()
|
||||
result["success"] = True
|
||||
result["message"] = "Playback paused"
|
||||
elif action == "stop":
|
||||
self.stop()
|
||||
result["success"] = True
|
||||
result["message"] = "Playback stopped"
|
||||
elif action == "next_track":
|
||||
self.next_track()
|
||||
result["success"] = True
|
||||
result["message"] = "Skipped to next track"
|
||||
elif action == "previous_track":
|
||||
self.previous_track()
|
||||
result["success"] = True
|
||||
result["message"] = "Skipped to previous track"
|
||||
elif action.startswith("set_volume:"):
|
||||
parts = action.split(":", 1)
|
||||
if len(parts) == 2 and _VOLUME_RE.match(parts[1]):
|
||||
vol = int(parts[1])
|
||||
self.set_volume(vol)
|
||||
result["success"] = True
|
||||
result["message"] = f"Volume set to {vol}"
|
||||
else:
|
||||
result["message"] = f"Invalid volume value: {action}"
|
||||
elif action == "toggle_mute":
|
||||
current = self.get_mute()
|
||||
self.set_mute(not current)
|
||||
result["success"] = True
|
||||
result["message"] = f"Mute toggled to {not current}"
|
||||
else:
|
||||
result["message"] = f"Unknown action: {action}"
|
||||
except Exception as e:
|
||||
# Broad catch: action methods wrap SoCo calls that can fail variously
|
||||
result["message"] = f"Action '{action}' failed: {e}"
|
||||
return result
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "napalm-sonos"
|
||||
version = "0.1.0"
|
||||
version = "0.2.0"
|
||||
description = "NAPALM driver for Sonos speakers (SoCo)."
|
||||
readme = "README.md"
|
||||
license = { text = "Apache-2.0" }
|
||||
|
||||
+6
-32
@@ -171,37 +171,11 @@ class TestSonosDriver:
|
||||
assert zone["model_name"] == "Sonos One"
|
||||
assert zone["ip_address"] == "192.168.1.100"
|
||||
|
||||
def test_run_device_action_play(self):
|
||||
result = self.driver.run_device_action("play")
|
||||
assert result["success"] is True
|
||||
assert result["action"] == "play"
|
||||
def test_the_media_controls_are_methods_not_a_string_dispatch(self):
|
||||
"""run_device_action returned another shape than its contract and is gone;
|
||||
play(), set_volume() and the others are called directly (netOrk #1105)."""
|
||||
assert not hasattr(type(self.driver), "run_device_action")
|
||||
self.driver.play()
|
||||
self.driver.set_volume(42)
|
||||
self.mock_device.play.assert_called_once()
|
||||
|
||||
def test_run_device_action_pause(self):
|
||||
result = self.driver.run_device_action("pause")
|
||||
assert result["success"] is True
|
||||
self.mock_device.pause.assert_called_once()
|
||||
|
||||
def test_run_device_action_stop(self):
|
||||
result = self.driver.run_device_action("stop")
|
||||
assert result["success"] is True
|
||||
self.mock_device.stop.assert_called_once()
|
||||
|
||||
def test_run_device_action_set_volume(self):
|
||||
result = self.driver.run_device_action("set_volume:42")
|
||||
assert result["success"] is True
|
||||
assert self.mock_device.volume == 42
|
||||
|
||||
def test_run_device_action_set_volume_invalid(self):
|
||||
result = self.driver.run_device_action("set_volume:abc")
|
||||
assert result["success"] is False
|
||||
|
||||
def test_run_device_action_toggle_mute(self):
|
||||
result = self.driver.run_device_action("toggle_mute")
|
||||
assert result["success"] is True
|
||||
assert self.mock_device.mute is True
|
||||
|
||||
def test_run_device_action_unknown(self):
|
||||
result = self.driver.run_device_action("foobar")
|
||||
assert result["success"] is False
|
||||
assert "Unknown" in result["message"]
|
||||
Reference in new issue
Block a user