core/tests/components/motion_blinds/test_gateway.py at d494ee9c08d655ecf62e89eee17b0897a21374c0 · home-assistant/core · GitHub

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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""Tests for Motion Gateway interface detection."""
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from homeassistant.components.motion_blindds.gateway import ConnectMotionGateway
from homeassistant.components.motion_blinds.const import DEFAULT_INTERFACE


MOCK_ADAPTERS_DUAL_NIC = [
    {
        "ipv4": [{"address": "192.168.1.10"}],
        "enabled": True,
        "default": True,
    },
    {
        "ipv4": [{"address": "192.168.20.10"}],
        "enabled": True,
        "default": False,
    },
]

MOCK_ADAPTERS_SINGLE_NIC = [
    {
        "ipv4": [{"address": "192.168.1.10"}],
        "enabled": True,
        "default": True,
    },
]


async def test_get_interfaces_dual_nic_default_prioritized(hass):
    """Test that default interface is first in list with multiple NICs."""
    gateway = ConnectMotionGateway(hass)

    with patch(
        "homeassistant.components.motion_blinds.gateway.network.async_get_adapters",
        return_value=MOCK_ADAPTERS_DUAL_NIC,
    ):
        interfaces = await gateway.async_get_interfaces()

    # Default interface (192.168.1.10) must be first
    assert interfaces[0] == "192.168.1.10"


async def test_get_interfaces_single_nic(hass):
    """Test that single NIC setup still works correctly."""
    gateway = ConnectMotionGateway(hass)

    with patch(
        "homeassistant.components.motion_blinds.gateway.network.async_get_adapters",
        return_value=MOCK_ADAPTERS_SINGLE_NIC,
    ):
        interfaces = await gateway.async_get_interfaces()

    assert interfaces[0] == "192.168.1.10"


async def test_check_interface_timeout_moves_to_next(hass):
    """Test that a timed out interface check moves to the next interface."""
    gateway = ConnectMotionGateway(hass)
    successful_interface = "192.168.1.10"

    with patch(
        "homeassistant.components.motion_blinds.gateway.network.async_get_adapters",
        return_value=MOCK_ADAPTERS_DUAL_NIC,
    ), patch(
        "homeassistant.components.motion_blinds.gateway.AsyncMotionMulticast"
    ) as mock_multicast, patch(
        "homeassistant.components.motion_blinds.gateway.MotionGateway"
    ):
        mock_instance = MagicMock()
        mock_instance.Start_listen = AsyncMock()
        mock_instance.Stop_listen = MagicMock()
        mock_multicast.return_value = mock_instance

        call_count = 0

        async def check_interface_side_effect():
            nonlocal call_count
            call_count += 1
            # First interface times out, second succeeds
            if call_count == 1:
                await asyncio.sleep(10)  # Will be cancelled by timeout
            return True

        with patch.object(
            gateway,
            "check_interface",
            side_effect=check_interface_side_effect,
        ):
            result = await gateway.async_check_interface("192.168.1.1", "testkey")

    assert result == successful_interface


async def test_check_interface_found_on_first_try(hass):
    """Test that correct interface on first try returns immediately."""
    gateway = ConnectMotionGateway(hass)

    with patch(
        "homeassistant.components.motion_blinds.gateway.network.async_get_adapters",
        return_value=MOCK_ADAPTERS_DUAL_NIC,
    ), patch(
        "homeassistant.components.motion_blinds.gateway.AsyncMotionMulticast"
    ) as mock_multicast, patch(
        "homeassistant.components.motion_blinds.gateway.MotionGateway"
    ):
        mock_instance = MagicMock()
        mock_instance.Start_listen = AsyncMock()
        mock_instance.Stop_listen = MagicMock()
        mock_multicast.return_value = mock_instance

        with patch.object(gateway, "check_interface", return_value=True):
            result = await gateway.async_check_interface("192.168.1.1", "testkey")

    # Should return the first (default) interface immediately
    assert result == "192.168.1.10"


async def test_check_interface_none_working_falls_back(hass):
    """Test fallback to stored interface when none work."""
    gateway = ConnectMotionGateway(hass, interface="0.0.0.0")

    with patch(
        "homeassistant.components.motion_blinds.gateway.network.async_get_adapters",
        return_value=MOCK_ADAPTERS_DUAL_NIC,
    ), patch(
        "homeassistant.components.motion_blinds.gateway.AsyncMotionMulticast"
    ) as mock_multicast, patch(
        "homeassistant.components.motion_blinds.gateway.MotionGateway"
    ):
        mock_instance = MagicMock()
        mock_instance.Start_listen = AsyncMock()
        mock_instance.Stop_listen = MagicMock()
        mock_multicast.return_value = mock_instance

        with patch.object(gateway, "check_interface", return_value=False):
            result = await gateway.async_check_interface("192.168.1.1", "testkey")

    # Should fall back to stored interface
    assert result == "0.0.0.0"