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 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222
|
"""Tests for the Source module."""
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Any
from unittest.mock import Mock, patch
import pytest
from asusrouter.modules import source
from asusrouter.modules.source import (
ARDataSource,
ARDataState,
ARDataStateDynamic,
ARDataStateStatic,
ARDataType,
)
datetime_value = datetime(2025, 8, 1, 12, 1, 5)
class TestARDataState:
"""Class for testing ARDataState."""
@pytest.mark.parametrize(
("source", "success"),
[
# Valid types
(ARDataSource(), True),
(ARDataType.UNKNOWN, True),
# Invalid types
(object(), False),
("string", False),
(None, False),
],
)
def test_init(self, source: Any, success: bool) -> None:
"""Test the initialization."""
if success:
instance = ARDataState(source)
assert instance._source == source
assert instance._content is None
assert instance._last_update is None
assert instance._callback is None
return
with pytest.raises(
TypeError,
match="A valid `ARDataSource` or `ARDataType` is required",
):
ARDataState(source)
@pytest.mark.parametrize(
"content",
[
{"key": "value"},
["item1", "item2"],
None,
"string",
object(),
],
)
def test_update(
self, content: Any, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test the update method."""
mock_now = Mock(return_value=datetime_value)
monkeypatch.setattr(source, "datetime", SimpleNamespace(now=mock_now))
instance = ARDataState(ARDataType.UNKNOWN)
instance.update(content)
mock_now.assert_called_once_with(UTC)
assert instance._content == content
assert instance._last_update == datetime_value
@pytest.mark.parametrize("is_fresh", [True, False, None])
def test_is_fresh(
self, is_fresh: bool | None, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test the is_fresh method."""
instance = ARDataState(ARDataType.UNKNOWN)
threshold = timedelta(seconds=5)
# No last update
if is_fresh is None:
result = instance.is_fresh(threshold=threshold)
assert result is False
# With last update
mock_now = Mock(return_value=datetime_value)
monkeypatch.setattr(source, "datetime", SimpleNamespace(now=mock_now))
# Fresh
if is_fresh is True:
instance._last_update = (
datetime_value - threshold + timedelta(seconds=1)
)
result = instance.is_fresh(threshold=threshold)
assert result is True
return
# Stale
instance._last_update = (
datetime_value - threshold - timedelta(seconds=1)
)
result = instance.is_fresh(threshold=threshold)
assert result is False
@pytest.mark.parametrize(
"threshold",
[
"string",
5,
None,
object(),
],
)
def test_is_fresh_invalid_threshold(self, threshold: Any) -> None:
"""Test the is_fresh method with invalid threshold."""
instance = ARDataState(ARDataType.UNKNOWN)
with pytest.raises(TypeError, match="A valid `timedelta` is required"):
instance.is_fresh(threshold=threshold)
def test_properties(self) -> None:
"""Test the properties."""
instance = ARDataState(ARDataType.UNKNOWN)
async def mock_async_callback() -> None:
"""Mock an async callback."""
# Mock the properties
instance._content = "content"
instance._last_update = datetime_value
instance._callback = mock_async_callback
assert instance.content == "content"
assert instance.last_update == datetime_value
assert instance.callback == mock_async_callback
def test_setter_callback(self) -> None:
"""Test the setter for the callback property."""
instance = ARDataState(ARDataType.UNKNOWN)
async def mock_async_callback() -> None:
"""Mock an async callback."""
instance.callback = mock_async_callback
assert instance.callback == mock_async_callback
class TestARDataStateStatic:
"""Class for testing ARDataStateStatic."""
def test_init(self) -> None:
"""Test the initialization."""
inst_source = ARDataType.UNKNOWN
instance = ARDataStateStatic(inst_source)
assert isinstance(instance, ARDataStateStatic)
assert issubclass(instance.__class__, ARDataState)
assert instance._source == inst_source
assert instance._content is None
assert instance._last_update is None
assert instance._callback is None
def test_property_source(self) -> None:
"""Test the source property."""
inst_source = ARDataType.UNKNOWN
instance = ARDataStateStatic(inst_source)
with patch(
"asusrouter.modules.source.ARDataType.from_value",
return_value=inst_source,
) as mock_from_value:
result = instance.source
mock_from_value.assert_called_once_with(instance._source)
assert result == inst_source
class TestARDataStateDynamic:
"""Class for testing ARDataStateDynamic."""
def test_init(self) -> None:
"""Test the initialization."""
inst_source = ARDataSource()
instance = ARDataStateDynamic(inst_source)
assert isinstance(instance, ARDataStateDynamic)
assert issubclass(instance.__class__, ARDataState)
assert instance._source == inst_source
assert instance._content is None
assert instance._last_update is None
assert instance._callback is None
def test_property_source(self) -> None:
"""Test the source property."""
inst_source = ARDataSource()
instance = ARDataStateDynamic(inst_source)
result = instance.source
assert result == inst_source
inst_source_wrong = ARDataType.UNKNOWN
instance = ARDataStateDynamic(inst_source_wrong) # type: ignore[arg-type]
result = instance.source
# It should return a new instance of ARDataSource
assert isinstance(result, ARDataSource)
|