mirror of
https://github.com/home-assistant/core.git
synced 2025-07-14 16:57:10 +00:00
Fix aprs tests with python 3.11 (#88189)
* Fix aprs tests with python 3.11 * hints
This commit is contained in:
parent
80e2f96097
commit
c8b827b46b
@ -1,7 +1,9 @@
|
|||||||
"""Test APRS device tracker."""
|
"""Test APRS device tracker."""
|
||||||
from unittest.mock import Mock, patch
|
from collections.abc import Generator
|
||||||
|
from unittest.mock import MagicMock, Mock, patch
|
||||||
|
|
||||||
import aprslib
|
import aprslib
|
||||||
|
from aprslib import IS
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
import homeassistant.components.aprs.device_tracker as device_tracker
|
import homeassistant.components.aprs.device_tracker as device_tracker
|
||||||
@ -16,6 +18,13 @@ TEST_HOST = "testhost"
|
|||||||
TEST_PASSWORD = "testpass"
|
TEST_PASSWORD = "testpass"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(name="mock_ais")
|
||||||
|
def mock_ais() -> Generator[MagicMock, None, None]:
|
||||||
|
"""Mock aprslib."""
|
||||||
|
with patch("aprslib.IS") as mock_ais:
|
||||||
|
yield mock_ais
|
||||||
|
|
||||||
|
|
||||||
def test_make_filter() -> None:
|
def test_make_filter() -> None:
|
||||||
"""Test filter."""
|
"""Test filter."""
|
||||||
callsigns = ["CALLSIGN1", "callsign2"]
|
callsigns = ["CALLSIGN1", "callsign2"]
|
||||||
@ -86,35 +95,34 @@ def test_gps_accuracy_invalid_float() -> None:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener() -> None:
|
def test_aprs_listener(mock_ais: MagicMock) -> None:
|
||||||
"""Test listener thread."""
|
"""Test listener thread."""
|
||||||
with patch("aprslib.IS") as mock_ais:
|
callsign = TEST_CALLSIGN
|
||||||
callsign = TEST_CALLSIGN
|
password = TEST_PASSWORD
|
||||||
password = TEST_PASSWORD
|
host = TEST_HOST
|
||||||
host = TEST_HOST
|
server_filter = TEST_FILTER
|
||||||
server_filter = TEST_FILTER
|
port = DEFAULT_PORT
|
||||||
port = DEFAULT_PORT
|
see = Mock()
|
||||||
see = Mock()
|
|
||||||
|
|
||||||
listener = device_tracker.AprsListenerThread(
|
listener = device_tracker.AprsListenerThread(
|
||||||
callsign, password, host, server_filter, see
|
callsign, password, host, server_filter, see
|
||||||
)
|
)
|
||||||
listener.run()
|
listener.run()
|
||||||
|
|
||||||
assert listener.callsign == callsign
|
assert listener.callsign == callsign
|
||||||
assert listener.host == host
|
assert listener.host == host
|
||||||
assert listener.server_filter == server_filter
|
assert listener.server_filter == server_filter
|
||||||
assert listener.see == see
|
assert listener.see == see
|
||||||
assert listener.start_event.is_set()
|
assert listener.start_event.is_set()
|
||||||
assert listener.start_success
|
assert listener.start_success
|
||||||
assert listener.start_message == "Connected to testhost with callsign testcall."
|
assert listener.start_message == "Connected to testhost with callsign testcall."
|
||||||
mock_ais.assert_called_with(callsign, passwd=password, host=host, port=port)
|
mock_ais.assert_called_with(callsign, passwd=password, host=host, port=port)
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener_start_fail() -> None:
|
def test_aprs_listener_start_fail() -> None:
|
||||||
"""Test listener thread start failure."""
|
"""Test listener thread start failure."""
|
||||||
with patch(
|
with patch.object(
|
||||||
"aprslib.IS.connect", side_effect=aprslib.ConnectionError("Unable to connect.")
|
IS, "connect", side_effect=aprslib.ConnectionError("Unable to connect.")
|
||||||
):
|
):
|
||||||
callsign = TEST_CALLSIGN
|
callsign = TEST_CALLSIGN
|
||||||
password = TEST_PASSWORD
|
password = TEST_PASSWORD
|
||||||
@ -136,166 +144,161 @@ def test_aprs_listener_start_fail() -> None:
|
|||||||
assert listener.start_message == "Unable to connect."
|
assert listener.start_message == "Unable to connect."
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener_stop() -> None:
|
def test_aprs_listener_stop(mock_ais: MagicMock) -> None:
|
||||||
"""Test listener thread stop."""
|
"""Test listener thread stop."""
|
||||||
with patch("aprslib.IS"):
|
callsign = TEST_CALLSIGN
|
||||||
callsign = TEST_CALLSIGN
|
password = TEST_PASSWORD
|
||||||
password = TEST_PASSWORD
|
host = TEST_HOST
|
||||||
host = TEST_HOST
|
server_filter = TEST_FILTER
|
||||||
server_filter = TEST_FILTER
|
see = Mock()
|
||||||
see = Mock()
|
|
||||||
|
|
||||||
listener = device_tracker.AprsListenerThread(
|
listener = device_tracker.AprsListenerThread(
|
||||||
callsign, password, host, server_filter, see
|
callsign, password, host, server_filter, see
|
||||||
)
|
)
|
||||||
listener.ais.close = Mock()
|
listener.ais.close = Mock()
|
||||||
listener.run()
|
listener.run()
|
||||||
listener.stop()
|
listener.stop()
|
||||||
|
|
||||||
assert listener.callsign == callsign
|
assert listener.callsign == callsign
|
||||||
assert listener.host == host
|
assert listener.host == host
|
||||||
assert listener.server_filter == server_filter
|
assert listener.server_filter == server_filter
|
||||||
assert listener.see == see
|
assert listener.see == see
|
||||||
assert listener.start_event.is_set()
|
assert listener.start_event.is_set()
|
||||||
assert listener.start_message == "Connected to testhost with callsign testcall."
|
assert listener.start_message == "Connected to testhost with callsign testcall."
|
||||||
assert listener.start_success
|
assert listener.start_success
|
||||||
listener.ais.close.assert_called_with()
|
listener.ais.close.assert_called_with()
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener_rx_msg() -> None:
|
def test_aprs_listener_rx_msg(mock_ais: MagicMock) -> None:
|
||||||
"""Test rx_msg."""
|
"""Test rx_msg."""
|
||||||
with patch("aprslib.IS"):
|
callsign = TEST_CALLSIGN
|
||||||
callsign = TEST_CALLSIGN
|
password = TEST_PASSWORD
|
||||||
password = TEST_PASSWORD
|
host = TEST_HOST
|
||||||
host = TEST_HOST
|
server_filter = TEST_FILTER
|
||||||
server_filter = TEST_FILTER
|
see = Mock()
|
||||||
see = Mock()
|
|
||||||
|
|
||||||
sample_msg = {
|
sample_msg = {
|
||||||
device_tracker.ATTR_FORMAT: "uncompressed",
|
device_tracker.ATTR_FORMAT: "uncompressed",
|
||||||
device_tracker.ATTR_FROM: "ZZ0FOOBAR-1",
|
device_tracker.ATTR_FROM: "ZZ0FOOBAR-1",
|
||||||
device_tracker.ATTR_LATITUDE: 0.0,
|
device_tracker.ATTR_LATITUDE: 0.0,
|
||||||
device_tracker.ATTR_LONGITUDE: 0.0,
|
device_tracker.ATTR_LONGITUDE: 0.0,
|
||||||
device_tracker.ATTR_ALTITUDE: 0,
|
device_tracker.ATTR_ALTITUDE: 0,
|
||||||
}
|
}
|
||||||
|
|
||||||
listener = device_tracker.AprsListenerThread(
|
listener = device_tracker.AprsListenerThread(
|
||||||
callsign, password, host, server_filter, see
|
callsign, password, host, server_filter, see
|
||||||
)
|
)
|
||||||
listener.run()
|
listener.run()
|
||||||
listener.rx_msg(sample_msg)
|
listener.rx_msg(sample_msg)
|
||||||
|
|
||||||
assert listener.callsign == callsign
|
assert listener.callsign == callsign
|
||||||
assert listener.host == host
|
assert listener.host == host
|
||||||
assert listener.server_filter == server_filter
|
assert listener.server_filter == server_filter
|
||||||
assert listener.see == see
|
assert listener.see == see
|
||||||
assert listener.start_event.is_set()
|
assert listener.start_event.is_set()
|
||||||
assert listener.start_success
|
assert listener.start_success
|
||||||
assert listener.start_message == "Connected to testhost with callsign testcall."
|
assert listener.start_message == "Connected to testhost with callsign testcall."
|
||||||
see.assert_called_with(
|
see.assert_called_with(
|
||||||
dev_id=device_tracker.slugify("ZZ0FOOBAR-1"),
|
dev_id=device_tracker.slugify("ZZ0FOOBAR-1"),
|
||||||
gps=(0.0, 0.0),
|
gps=(0.0, 0.0),
|
||||||
attributes={"altitude": 0},
|
attributes={"altitude": 0},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener_rx_msg_ambiguity() -> None:
|
def test_aprs_listener_rx_msg_ambiguity(mock_ais: MagicMock) -> None:
|
||||||
"""Test rx_msg with posambiguity."""
|
"""Test rx_msg with posambiguity."""
|
||||||
with patch("aprslib.IS"):
|
callsign = TEST_CALLSIGN
|
||||||
callsign = TEST_CALLSIGN
|
password = TEST_PASSWORD
|
||||||
password = TEST_PASSWORD
|
host = TEST_HOST
|
||||||
host = TEST_HOST
|
server_filter = TEST_FILTER
|
||||||
server_filter = TEST_FILTER
|
see = Mock()
|
||||||
see = Mock()
|
|
||||||
|
|
||||||
sample_msg = {
|
sample_msg = {
|
||||||
device_tracker.ATTR_FORMAT: "uncompressed",
|
device_tracker.ATTR_FORMAT: "uncompressed",
|
||||||
device_tracker.ATTR_FROM: "ZZ0FOOBAR-1",
|
device_tracker.ATTR_FROM: "ZZ0FOOBAR-1",
|
||||||
device_tracker.ATTR_LATITUDE: 0.0,
|
device_tracker.ATTR_LATITUDE: 0.0,
|
||||||
device_tracker.ATTR_LONGITUDE: 0.0,
|
device_tracker.ATTR_LONGITUDE: 0.0,
|
||||||
device_tracker.ATTR_POS_AMBIGUITY: 1,
|
device_tracker.ATTR_POS_AMBIGUITY: 1,
|
||||||
}
|
}
|
||||||
|
|
||||||
listener = device_tracker.AprsListenerThread(
|
listener = device_tracker.AprsListenerThread(
|
||||||
callsign, password, host, server_filter, see
|
callsign, password, host, server_filter, see
|
||||||
)
|
)
|
||||||
listener.run()
|
listener.run()
|
||||||
listener.rx_msg(sample_msg)
|
listener.rx_msg(sample_msg)
|
||||||
|
|
||||||
assert listener.callsign == callsign
|
assert listener.callsign == callsign
|
||||||
assert listener.host == host
|
assert listener.host == host
|
||||||
assert listener.server_filter == server_filter
|
assert listener.server_filter == server_filter
|
||||||
assert listener.see == see
|
assert listener.see == see
|
||||||
assert listener.start_event.is_set()
|
assert listener.start_event.is_set()
|
||||||
assert listener.start_success
|
assert listener.start_success
|
||||||
assert listener.start_message == "Connected to testhost with callsign testcall."
|
assert listener.start_message == "Connected to testhost with callsign testcall."
|
||||||
see.assert_called_with(
|
see.assert_called_with(
|
||||||
dev_id=device_tracker.slugify("ZZ0FOOBAR-1"),
|
dev_id=device_tracker.slugify("ZZ0FOOBAR-1"),
|
||||||
gps=(0.0, 0.0),
|
gps=(0.0, 0.0),
|
||||||
attributes={device_tracker.ATTR_GPS_ACCURACY: 186},
|
attributes={device_tracker.ATTR_GPS_ACCURACY: 186},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener_rx_msg_ambiguity_invalid() -> None:
|
def test_aprs_listener_rx_msg_ambiguity_invalid(mock_ais: MagicMock) -> None:
|
||||||
"""Test rx_msg with invalid posambiguity."""
|
"""Test rx_msg with invalid posambiguity."""
|
||||||
with patch("aprslib.IS"):
|
callsign = TEST_CALLSIGN
|
||||||
callsign = TEST_CALLSIGN
|
password = TEST_PASSWORD
|
||||||
password = TEST_PASSWORD
|
host = TEST_HOST
|
||||||
host = TEST_HOST
|
server_filter = TEST_FILTER
|
||||||
server_filter = TEST_FILTER
|
see = Mock()
|
||||||
see = Mock()
|
|
||||||
|
|
||||||
sample_msg = {
|
sample_msg = {
|
||||||
device_tracker.ATTR_FORMAT: "uncompressed",
|
device_tracker.ATTR_FORMAT: "uncompressed",
|
||||||
device_tracker.ATTR_FROM: "ZZ0FOOBAR-1",
|
device_tracker.ATTR_FROM: "ZZ0FOOBAR-1",
|
||||||
device_tracker.ATTR_LATITUDE: 0.0,
|
device_tracker.ATTR_LATITUDE: 0.0,
|
||||||
device_tracker.ATTR_LONGITUDE: 0.0,
|
device_tracker.ATTR_LONGITUDE: 0.0,
|
||||||
device_tracker.ATTR_POS_AMBIGUITY: 5,
|
device_tracker.ATTR_POS_AMBIGUITY: 5,
|
||||||
}
|
}
|
||||||
|
|
||||||
listener = device_tracker.AprsListenerThread(
|
listener = device_tracker.AprsListenerThread(
|
||||||
callsign, password, host, server_filter, see
|
callsign, password, host, server_filter, see
|
||||||
)
|
)
|
||||||
listener.run()
|
listener.run()
|
||||||
listener.rx_msg(sample_msg)
|
listener.rx_msg(sample_msg)
|
||||||
|
|
||||||
assert listener.callsign == callsign
|
assert listener.callsign == callsign
|
||||||
assert listener.host == host
|
assert listener.host == host
|
||||||
assert listener.server_filter == server_filter
|
assert listener.server_filter == server_filter
|
||||||
assert listener.see == see
|
assert listener.see == see
|
||||||
assert listener.start_event.is_set()
|
assert listener.start_event.is_set()
|
||||||
assert listener.start_success
|
assert listener.start_success
|
||||||
assert listener.start_message == "Connected to testhost with callsign testcall."
|
assert listener.start_message == "Connected to testhost with callsign testcall."
|
||||||
see.assert_called_with(
|
see.assert_called_with(
|
||||||
dev_id=device_tracker.slugify("ZZ0FOOBAR-1"), gps=(0.0, 0.0), attributes={}
|
dev_id=device_tracker.slugify("ZZ0FOOBAR-1"), gps=(0.0, 0.0), attributes={}
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_aprs_listener_rx_msg_no_position() -> None:
|
def test_aprs_listener_rx_msg_no_position(mock_ais: MagicMock) -> None:
|
||||||
"""Test rx_msg with non-position report."""
|
"""Test rx_msg with non-position report."""
|
||||||
with patch("aprslib.IS"):
|
callsign = TEST_CALLSIGN
|
||||||
callsign = TEST_CALLSIGN
|
password = TEST_PASSWORD
|
||||||
password = TEST_PASSWORD
|
host = TEST_HOST
|
||||||
host = TEST_HOST
|
server_filter = TEST_FILTER
|
||||||
server_filter = TEST_FILTER
|
see = Mock()
|
||||||
see = Mock()
|
|
||||||
|
|
||||||
sample_msg = {device_tracker.ATTR_FORMAT: "invalid"}
|
sample_msg = {device_tracker.ATTR_FORMAT: "invalid"}
|
||||||
|
|
||||||
listener = device_tracker.AprsListenerThread(
|
listener = device_tracker.AprsListenerThread(
|
||||||
callsign, password, host, server_filter, see
|
callsign, password, host, server_filter, see
|
||||||
)
|
)
|
||||||
listener.run()
|
listener.run()
|
||||||
listener.rx_msg(sample_msg)
|
listener.rx_msg(sample_msg)
|
||||||
|
|
||||||
assert listener.callsign == callsign
|
assert listener.callsign == callsign
|
||||||
assert listener.host == host
|
assert listener.host == host
|
||||||
assert listener.server_filter == server_filter
|
assert listener.server_filter == server_filter
|
||||||
assert listener.see == see
|
assert listener.see == see
|
||||||
assert listener.start_event.is_set()
|
assert listener.start_event.is_set()
|
||||||
assert listener.start_success
|
assert listener.start_success
|
||||||
assert listener.start_message == "Connected to testhost with callsign testcall."
|
assert listener.start_message == "Connected to testhost with callsign testcall."
|
||||||
see.assert_not_called()
|
see.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
async def test_setup_scanner(hass: HomeAssistant) -> None:
|
async def test_setup_scanner(hass: HomeAssistant) -> None:
|
||||||
@ -324,7 +327,7 @@ async def test_setup_scanner(hass: HomeAssistant) -> None:
|
|||||||
|
|
||||||
async def test_setup_scanner_timeout(hass: HomeAssistant) -> None:
|
async def test_setup_scanner_timeout(hass: HomeAssistant) -> None:
|
||||||
"""Test setup_scanner failure from timeout."""
|
"""Test setup_scanner failure from timeout."""
|
||||||
with patch("aprslib.IS.connect", side_effect=TimeoutError):
|
with patch.object(IS, "connect", side_effect=TimeoutError):
|
||||||
config = {
|
config = {
|
||||||
"username": TEST_CALLSIGN,
|
"username": TEST_CALLSIGN,
|
||||||
"password": TEST_PASSWORD,
|
"password": TEST_PASSWORD,
|
||||||
|
Loading…
x
Reference in New Issue
Block a user