Add kwargs to send_magic_packet() service individually (#37387)

This commit is contained in:
Sava Tshontikidis 2020-07-08 17:25:21 -04:00 committed by GitHub
parent 05f93f27df
commit c1de781a23
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 124 additions and 32 deletions

View file

@ -31,23 +31,23 @@ async def async_setup(hass, config):
mac_address = call.data.get(CONF_MAC)
broadcast_address = call.data.get(CONF_BROADCAST_ADDRESS)
broadcast_port = call.data.get(CONF_BROADCAST_PORT)
service_kwargs = {}
if broadcast_address is not None:
service_kwargs["ip_address"] = broadcast_address
if broadcast_port is not None:
service_kwargs["port"] = broadcast_port
_LOGGER.info(
"Send magic packet to mac %s (broadcast: %s, port: %s)",
mac_address,
broadcast_address,
broadcast_port,
)
if broadcast_address is not None:
await hass.async_add_job(
partial(
wakeonlan.send_magic_packet,
mac_address,
ip_address=broadcast_address,
port=broadcast_port,
)
)
else:
await hass.async_add_job(partial(wakeonlan.send_magic_packet, mac_address))
await hass.async_add_job(
partial(wakeonlan.send_magic_packet, mac_address, **service_kwargs)
)
hass.services.async_register(
DOMAIN,

View file

@ -96,14 +96,20 @@ class WolSwitch(SwitchEntity):
def turn_on(self, **kwargs):
"""Turn the device on."""
if self._broadcast_address:
wakeonlan.send_magic_packet(
self._mac_address,
ip_address=self._broadcast_address,
port=self._broadcast_port,
)
else:
wakeonlan.send_magic_packet(self._mac_address)
service_kwargs = {}
if self._broadcast_address is not None:
service_kwargs["ip_address"] = self._broadcast_address
if self._broadcast_port is not None:
service_kwargs["port"] = self._broadcast_port
_LOGGER.info(
"Send magic packet to mac %s (broadcast: %s, port: %s)",
self._mac_address,
self._broadcast_address,
self._broadcast_port,
)
wakeonlan.send_magic_packet(self._mac_address, **service_kwargs)
def turn_off(self, **kwargs):
"""Turn the device off if an off action is present."""

View file

@ -28,6 +28,28 @@ async def test_send_magic_packet(hass):
assert mocked_wakeonlan.mock_calls[-1][2]["ip_address"] == bc_ip
assert mocked_wakeonlan.mock_calls[-1][2]["port"] == bc_port
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_MAGIC_PACKET,
{"mac": mac, "broadcast_address": bc_ip},
blocking=True,
)
assert len(mocked_wakeonlan.mock_calls) == 2
assert mocked_wakeonlan.mock_calls[-1][1][0] == mac
assert mocked_wakeonlan.mock_calls[-1][2]["ip_address"] == bc_ip
assert "port" not in mocked_wakeonlan.mock_calls[-1][2]
await hass.services.async_call(
DOMAIN,
SERVICE_SEND_MAGIC_PACKET,
{"mac": mac, "broadcast_port": bc_port},
blocking=True,
)
assert len(mocked_wakeonlan.mock_calls) == 3
assert mocked_wakeonlan.mock_calls[-1][1][0] == mac
assert mocked_wakeonlan.mock_calls[-1][2]["port"] == bc_port
assert "ip_address" not in mocked_wakeonlan.mock_calls[-1][2]
with pytest.raises(vol.Invalid):
await hass.services.async_call(
DOMAIN,
@ -35,11 +57,11 @@ async def test_send_magic_packet(hass):
{"broadcast_address": bc_ip},
blocking=True,
)
assert len(mocked_wakeonlan.mock_calls) == 1
assert len(mocked_wakeonlan.mock_calls) == 3
await hass.services.async_call(
DOMAIN, SERVICE_SEND_MAGIC_PACKET, {"mac": mac}, blocking=True
)
assert len(mocked_wakeonlan.mock_calls) == 2
assert len(mocked_wakeonlan.mock_calls) == 4
assert mocked_wakeonlan.mock_calls[-1][1][0] == mac
assert not mocked_wakeonlan.mock_calls[-1][2]

View file

@ -5,18 +5,13 @@ import homeassistant.components.switch as switch
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.setup import setup_component
from tests.async_mock import patch
from tests.async_mock import Mock, patch
from tests.common import get_test_home_assistant, mock_service
from tests.components.switch import common
TEST_STATE = None
def send_magic_packet(*macs, **kwargs):
"""Fake call for sending magic packets."""
return
def call(cmd, stdout, stderr):
"""Return fake subprocess return codes."""
if cmd[5] == "validhostname" and TEST_STATE:
@ -32,6 +27,8 @@ def system():
class TestWolSwitch(unittest.TestCase):
"""Test the wol switch."""
send_magic_packet = Mock(return_value=None)
def setUp(self):
"""Set up things to be run when tests are started."""
self.hass = get_test_home_assistant()
@ -116,17 +113,22 @@ class TestWolSwitch(unittest.TestCase):
@patch("wakeonlan.send_magic_packet", new=send_magic_packet)
@patch("subprocess.call", new=call)
def test_broadcast_config(self):
"""Test with broadcast address config."""
def test_broadcast_config_ip_and_port(self):
"""Test with broadcast address and broadcast port config."""
mac = "00-01-02-03-04-05"
broadcast_address = "255.255.255.255"
port = 999
assert setup_component(
self.hass,
switch.DOMAIN,
{
"switch": {
"platform": "wake_on_lan",
"mac": "00-01-02-03-04-05",
"broadcast_address": "255.255.255.255",
"broadcast_port": 999,
"mac": mac,
"broadcast_address": broadcast_address,
"broadcast_port": port,
}
},
)
@ -138,6 +140,68 @@ class TestWolSwitch(unittest.TestCase):
common.turn_on(self.hass, "switch.wake_on_lan")
self.hass.block_till_done()
self.send_magic_packet.assert_called_with(
mac, ip_address=broadcast_address, port=port
)
@patch("wakeonlan.send_magic_packet", new=send_magic_packet)
@patch("subprocess.call", new=call)
def test_broadcast_config_ip(self):
"""Test with only broadcast address."""
mac = "00-01-02-03-04-05"
broadcast_address = "255.255.255.255"
assert setup_component(
self.hass,
switch.DOMAIN,
{
"switch": {
"platform": "wake_on_lan",
"mac": mac,
"broadcast_address": broadcast_address,
}
},
)
self.hass.block_till_done()
state = self.hass.states.get("switch.wake_on_lan")
assert STATE_OFF == state.state
common.turn_on(self.hass, "switch.wake_on_lan")
self.hass.block_till_done()
self.send_magic_packet.assert_called_with(mac, ip_address=broadcast_address)
@patch("wakeonlan.send_magic_packet", new=send_magic_packet)
@patch("subprocess.call", new=call)
def test_broadcast_config_port(self):
"""Test with only broadcast port config."""
mac = "00-01-02-03-04-05"
port = 999
assert setup_component(
self.hass,
switch.DOMAIN,
{
"switch": {
"platform": "wake_on_lan",
"mac": mac,
"broadcast_port": port,
}
},
)
self.hass.block_till_done()
state = self.hass.states.get("switch.wake_on_lan")
assert STATE_OFF == state.state
common.turn_on(self.hass, "switch.wake_on_lan")
self.hass.block_till_done()
self.send_magic_packet.assert_called_with(mac, port=port)
@patch("wakeonlan.send_magic_packet", new=send_magic_packet)
@patch("subprocess.call", new=call)
def test_off_script(self):