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
|
"""Tests for rpc_device.wsrpc module."""
import pytest
from aioshelly.exceptions import InvalidAuthError
from . import load_device_fixture
from .conftest import WsRPCMocker
@pytest.mark.asyncio
async def test_device_wscall_get_config(ws_rpc: WsRPCMocker) -> None:
"""Test wscall."""
config_response = await load_device_fixture("shellyplugus", "Shelly.GetConfig")
calls = [("Shelly.GetConfig", None)]
responses = [config_response]
results = await ws_rpc.calls_with_mocked_responses(calls, responses)
assert results[0] == config_response["result"]
@pytest.mark.asyncio
async def test_device_wscall_no_auth_retry(ws_rpc: WsRPCMocker) -> None:
"""Test wscall when auth is not set."""
cover_close_auth_fail = await load_device_fixture(
"shellyplus2pm", "Cover.Close_auth_failure"
)
cover_close_success = await load_device_fixture(
"shellyplus2pm", "Cover.Close_success"
)
calls = [("Cover.Close", {"id": 0})]
responses = [cover_close_auth_fail, cover_close_success]
with pytest.raises(InvalidAuthError):
await ws_rpc.calls_with_mocked_responses(calls, responses)
@pytest.mark.asyncio
async def test_device_wscall_auth_retry(ws_rpc_with_auth: WsRPCMocker) -> None:
"""Test wscall when auth is set and a retry works."""
cover_close_auth_fail = await load_device_fixture(
"shellyplus2pm", "Cover.Close_auth_failure"
)
cover_close_success = await load_device_fixture(
"shellyplus2pm", "Cover.Close_success"
)
calls = [("Cover.Close", {"id": 0})]
responses = [cover_close_auth_fail, cover_close_success]
results = await ws_rpc_with_auth.calls_with_mocked_responses(calls, responses)
assert results[0] == cover_close_success["result"]
|
