GitHub Viewer
import json
import os
import re
from datetime import datetime
from unittest.mock import ANY, PropertyMock, patch
from zoneinfo import ZoneInfo
import asyncclick as click
import pytest
from asyncclick.testing import CliRunner
from pytest_mock import MockerFixture
from kasa import (
AuthenticationError,
Credentials,
Device,
DeviceError,
DeviceType,
EmeterStatus,
KasaException,
Module,
)
from kasa.cli.device import (
alias,
factory_reset,
led,
reboot,
state,
sysinfo,
toggle,
update_credentials,
)
from kasa.cli.light import (
brightness,
effect,
hsv,
presets,
presets_modify,
temperature,
)
from kasa.cli.main import TYPES, _legacy_type_to_class, cli, cmd_command, raw_command
from kasa.cli.time import time
from kasa.cli.usage import energy
from kasa.cli.wifi import wifi
from kasa.discover import Discover, DiscoveryResult
from kasa.iot import IotDevice
from kasa.smart import SmartDevice
from kasa.smartcam import SmartCamDevice
from .conftest import (
device_smart,
get_device_for_fixture_protocol,
handle_turn_on,
new_discovery,
turn_on,
)
# The cli tests should be testing the cli logic rather than a physical device
# so mark the whole file for skipping with real devices.
pytestmark = [pytest.mark.requires_dummy]
@pytest.fixture
def runner():
"""Runner fixture that unsets the KASA_ environment variables for tests."""
KASA_VARS = {k: None for k, v in os.environ.items() if k.startswith("KASA_")}
runner = CliRunner(env=KASA_VARS)
return runner
async def test_help(runner):
"""Test that all the lazy modules are correctly names."""
res = await runner.invoke(cli, ["--help"])
assert res.exit_code == 0, "--help failed, check lazy module names"
@pytest.mark.parametrize(
("device_family", "encrypt_type"),
[
pytest.param(None, None, id="No connect params"),
pytest.param("SMART.TAPOPLUG", None, id="Only device_family"),
],
)
async def test_update_called_by_cli(dev, mocker, runner, device_family, encrypt_type):
"""Test that device update is called on main."""
update = mocker.patch.object(dev, "update")
# These will mock the features to avoid accessing non-existing
mocker.patch("kasa.device.Device.features", return_value={})
mocker.patch("kasa.iot.iotdevice.IotDevice.features", return_value={})
mocker.patch("kasa.discover.Discover.discover_single", return_value=dev)
res = await runner.invoke(
cli,
[
"--host",
"127.0.0.1",
"--username",
"foo",
"--password",
"bar",
"--device-family",
device_family,
"--encrypt-type",
encrypt_type,
],
catch_exceptions=False,
)
assert res.exit_code == 0
update.assert_called()
async def test_list_devices(discovery_mock, runner):
"""Test that device update is called on main."""
res = await runner.invoke(
cli,
["--username", "foo", "--password", "bar", "discover", "list"],
catch_exceptions=False,
)
assert res.exit_code == 0
header = f"{'HOST':