Files
python-kasa/tests/cli/test_vacuum.py
ZeliardM 1a84242a05 tests: add type annotations to CLI, protocol, and smartcam tests (#1687)
Add type annotations (`-> None` return types and parameter types) to
test files in the `tests/cli/`, `tests/protocols/`, and
`tests/smartcam/` directories. This enables mypy to check test function
bodies, catching type errors that were previously hidden.
2026-07-05 18:46:01 +02:00

124 lines
4.0 KiB
Python

from asyncclick.testing import CliRunner
from pytest_mock import MockerFixture
from kasa import Device, DeviceType, Module
from kasa.cli.vacuum import vacuum
from ..device_fixtures import plug_iot
from ..device_fixtures import vacuum as vacuum_devices
@vacuum_devices
async def test_vacuum_records_group(
dev: Device, mocker: MockerFixture, runner: CliRunner
) -> None:
"""Test that vacuum records calls the expected methods."""
rec = dev.modules.get(Module.CleanRecords)
assert rec
res = await runner.invoke(vacuum, ["records"], obj=dev, catch_exceptions=False)
latest = rec.parsed_data.last_clean
expected = (
f"Totals: {rec.total_clean_area} {rec.area_unit} in {rec.total_clean_time} "
f"(cleaned {rec.total_clean_count} times)\n"
f"Last clean: {latest.clean_area} {rec.area_unit} @ {latest.clean_time}"
)
assert expected in res.output
assert res.exit_code == 0
@vacuum_devices
async def test_vacuum_records_list(
dev: Device, mocker: MockerFixture, runner: CliRunner
) -> None:
"""Test that vacuum records list calls the expected methods."""
rec = dev.modules.get(Module.CleanRecords)
assert rec
res = await runner.invoke(
vacuum, ["records", "list"], obj=dev, catch_exceptions=False
)
data = rec.parsed_data
for record in data.records:
expected = (
f"* {record.timestamp}: cleaned {record.clean_area} {rec.area_unit}"
f" in {record.clean_time}"
)
assert expected in res.output
assert res.exit_code == 0
@vacuum_devices
async def test_vacuum_consumables(dev: Device, runner: CliRunner) -> None:
"""Test that vacuum consumables calls the expected methods."""
cons = dev.modules.get(Module.Consumables)
assert cons
res = await runner.invoke(vacuum, ["consumables"], obj=dev, catch_exceptions=False)
expected = ""
for c in cons.consumables.values():
expected += f"{c.name} ({c.id}): {c.used} used, {c.remaining} remaining\n"
assert expected in res.output
assert res.exit_code == 0
@vacuum_devices
async def test_vacuum_consumables_reset(
dev: Device, mocker: MockerFixture, runner: CliRunner
) -> None:
"""Test that vacuum consumables reset calls the expected methods."""
cons = dev.modules.get(Module.Consumables)
assert cons
reset_consumable_mock = mocker.spy(cons, "reset_consumable")
for c_id in cons.consumables:
reset_consumable_mock.reset_mock()
res = await runner.invoke(
vacuum, ["consumables", "reset", c_id], obj=dev, catch_exceptions=False
)
reset_consumable_mock.assert_awaited_once_with(c_id)
assert f"Consumable {c_id} reset" in res.output
assert res.exit_code == 0
res = await runner.invoke(
vacuum, ["consumables", "reset", "foobar"], obj=dev, catch_exceptions=False
)
expected = (
"Consumable foobar not found in "
f"device consumables: {', '.join(cons.consumables.keys())}."
)
assert expected in res.output.replace("\n", "")
assert res.exit_code != 0
@plug_iot
async def test_non_vacuum(
dev: Device, mocker: MockerFixture, runner: CliRunner
) -> None:
"""Test that vacuum commands return an error if executed on a non-vacuum."""
assert dev.device_type is not DeviceType.Vacuum
res = await runner.invoke(vacuum, ["records"], obj=dev, catch_exceptions=False)
assert "This device does not support records" in res.output
assert res.exit_code != 0
res = await runner.invoke(
vacuum, ["records", "list"], obj=dev, catch_exceptions=False
)
assert "This device does not support records" in res.output
assert res.exit_code != 0
res = await runner.invoke(vacuum, ["consumables"], obj=dev, catch_exceptions=False)
assert "This device does not support consumables" in res.output
assert res.exit_code != 0
res = await runner.invoke(
vacuum, ["consumables", "reset", "foobar"], obj=dev, catch_exceptions=False
)
assert "This device does not support consumables" in res.output
assert res.exit_code != 0