1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
"""Unit test for XKNX Module."""
import logging
from pathlib import Path
import tempfile
from unittest.mock import AsyncMock, Mock, patch
import pytest
from xknx import XKNX
from xknx.exceptions import CommunicationError
from xknx.io import ConnectionConfig, ConnectionType
class TestXknxModule:
"""Test class for XKNX."""
def test_log_to_file(self) -> None:
"""Test logging enable."""
with tempfile.TemporaryDirectory() as tmpdir:
XKNX(log_directory=tmpdir)
_path = Path(f"{tmpdir}/xknx.log")
assert _path.is_file()
# Needed for Windows to release the logging file
logging.shutdown()
_path.unlink()
def test_log_to_file_when_dir_does_not_exist(self) -> None:
"""Test logging enable with non existent directory."""
XKNX(log_directory="/xknx/is/fun")
assert not Path("/xknx/is/fun/xknx.log").is_file()
def test_register_telegram_cb(self) -> None:
"""Test register telegram callback."""
xknx = XKNX(telegram_received_cb=AsyncMock())
assert len(xknx.telegram_queue.telegram_received_cbs) == 1
def test_register_device_cb(self) -> None:
"""Test register telegram callback."""
xknx = XKNX(device_updated_cb=AsyncMock())
assert len(xknx.devices.device_updated_cbs) == 1
def test_register_connection_state_change_cb(self) -> None:
"""Test register con state callback."""
xknx = XKNX(connection_state_changed_cb=Mock())
assert len(xknx.connection_manager._connection_state_changed_cbs) == 1
@patch("xknx.io.KNXIPInterface._start", new_callable=AsyncMock)
async def test_xknx_start(self, start_mock: AsyncMock) -> None:
"""Test xknx start."""
xknx = XKNX(state_updater=True)
await xknx.start()
start_mock.assert_called_once()
await xknx.stop()
@patch("xknx.io.KNXIPInterface._start", new_callable=AsyncMock)
async def test_xknx_start_as_context_manager(
self, ipinterface_mock: AsyncMock
) -> None:
"""Test xknx start."""
async with XKNX(state_updater=True) as xknx:
assert xknx.started.is_set()
ipinterface_mock.assert_called_once()
@patch("xknx.io.KNXIPInterface._start", new_callable=AsyncMock)
async def test_xknx_start_and_stop_with_dedicated_connection_config(
self, start_mock: AsyncMock
) -> None:
"""Test xknx start and stop with connection config."""
connection_config = ConnectionConfig(connection_type=ConnectionType.TUNNELING)
xknx = XKNX(connection_config=connection_config)
await xknx.start()
start_mock.assert_called_once()
assert xknx.knxip_interface.connection_config == connection_config
await xknx.stop()
assert xknx.knxip_interface._interface is None
assert xknx.telegram_queue._consumer_task.done()
assert not xknx.state_updater.started
@pytest.mark.parametrize(
"connection_config",
[
ConnectionConfig(
connection_type=ConnectionType.ROUTING, local_ip="127.0.0.1"
),
ConnectionConfig(
connection_type=ConnectionType.TUNNELING, gateway_ip="127.0.0.2"
),
],
)
@patch(
"xknx.io.transport.UDPTransport.connect",
new_callable=AsyncMock,
side_effect=OSError,
)
async def test_xknx_start_initial_connection_error(
self, transport_connect_mock: AsyncMock, connection_config: ConnectionConfig
) -> None:
"""Test xknx start raising when socket can't be set up."""
xknx = XKNX(
state_updater=True,
connection_config=connection_config,
)
with pytest.raises(CommunicationError):
await xknx.start()
transport_connect_mock.assert_called_once()
assert xknx.telegram_queue._consumer_task is None # not started
assert not xknx.state_updater.started
assert not xknx.started.is_set()
|
