Skip to content

Commit

Permalink
improve probing test
Browse files Browse the repository at this point in the history
  • Loading branch information
Sébastien RAMAGE committed Nov 20, 2020
1 parent 62d4144 commit 6f6d085
Show file tree
Hide file tree
Showing 2 changed files with 20 additions and 9 deletions.
24 changes: 17 additions & 7 deletions tests/test_api.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import asyncio

import sys
import pytest
import serial
import serial_asyncio
Expand All @@ -12,6 +12,7 @@
DEVICE_CONFIG = config.SCHEMA_DEVICE({config.CONF_DEVICE_PATH: "/dev/null"})



@pytest.fixture
def api():
api = zigate_api.ZiGate(DEVICE_CONFIG)
Expand Down Expand Up @@ -58,17 +59,26 @@ async def test_api_new(conn_mck):

@pytest.mark.asyncio
@patch.object(zigate_api.ZiGate, "set_raw_mode", new_callable=AsyncMock)
@patch.object(zigpy_zigate.uart, "connect")
async def test_probe_success(mock_connect, mock_raw_mode):
@pytest.mark.parametrize(
"port",
('/dev/null', 'pizigate:/dev/ttyAMA0'),
)
async def test_probe_success(mock_raw_mode, port, monkeypatch):
"""Test device probing."""

async def mock_conn(loop, protocol_factory, **kwargs):
protocol = protocol_factory()
loop.call_soon(protocol.connection_made, None)
return None, protocol
monkeypatch.setattr(serial_asyncio, "create_serial_connection", mock_conn)
DEVICE_CONFIG = zigpy_zigate.config.SCHEMA_DEVICE(
{zigpy_zigate.config.CONF_DEVICE_PATH: port}
)
sys.modules['RPi'] = MagicMock()
sys.modules['RPi.GPIO'] = MagicMock()
res = await zigate_api.ZiGate.probe(DEVICE_CONFIG)
assert res is True
assert mock_connect.call_count == 1
assert mock_connect.await_count == 1
assert mock_connect.call_args[0][0] == DEVICE_CONFIG
assert mock_raw_mode.call_count == 1
assert mock_connect.return_value.close.call_count == 1


@pytest.mark.asyncio
Expand Down
5 changes: 3 additions & 2 deletions zigpy_zigate/uart.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,8 @@ def connection_made(self, transport):
self._connected_future.set_result(True)

def close(self):
self._transport.close()
if self._transport:
self._transport.close()

def send(self, cmd, data=b''):
"""Send data, taking care of escaping and framing"""
Expand Down Expand Up @@ -145,7 +146,7 @@ async def connect(device_config: Dict[str, Any], api, loop=None):
# in case of pizigate:/dev/ttyAMA0 syntax
if port.startswith('pizigate:'):
port = port[9:]
elif c.is_zigate_din:
elif c.is_zigate_din(port):
LOGGER.debug('ZiGate USB DIN detected')
await c.set_zigatedin_running_mode()

Expand Down

0 comments on commit 6f6d085

Please sign in to comment.