mirror of
https://github.com/python-kasa/python-kasa.git
synced 2025-08-06 10:44:04 +00:00
Use ruff and ruff format (#534)
Replaces the previously used linting and code formatting tools with ruff.
This commit is contained in:
@@ -1,7 +1,8 @@
|
||||
"""Script that checks if README.md is missing devices that have fixtures."""
|
||||
from kasa.tests.conftest import ALL_DEVICES, BULBS, DIMMERS, LIGHT_STRIPS, PLUGS, STRIPS
|
||||
|
||||
readme = open("README.md").read()
|
||||
with open("README.md") as f:
|
||||
readme = f.read()
|
||||
|
||||
typemap = {
|
||||
"light strips": LIGHT_STRIPS,
|
||||
|
@@ -1,7 +1,8 @@
|
||||
"""This script generates devinfo files for the test suite.
|
||||
|
||||
If you have new, yet unsupported device or a device with no devinfo file under kasa/tests/fixtures,
|
||||
feel free to run this script and create a PR to add the file to the repository.
|
||||
If you have new, yet unsupported device or a device with no devinfo file under
|
||||
kasa/tests/fixtures, feel free to run this script and create a PR to add the file
|
||||
to the repository.
|
||||
|
||||
Executing this script will several modules and methods one by one,
|
||||
and finally execute a query to query all of them at once.
|
||||
@@ -84,13 +85,13 @@ def cli(host, debug):
|
||||
|
||||
for test_call in items:
|
||||
|
||||
async def _run_query():
|
||||
async def _run_query(test_call):
|
||||
protocol = TPLinkSmartHomeProtocol(host)
|
||||
return await protocol.query({test_call.module: {test_call.method: None}})
|
||||
|
||||
try:
|
||||
click.echo(f"Testing {test_call}..", nl=False)
|
||||
info = asyncio.run(_run_query())
|
||||
info = asyncio.run(_run_query(test_call))
|
||||
resp = info[test_call.module]
|
||||
except Exception as ex:
|
||||
click.echo(click.style(f"FAIL {ex}", fg="red"))
|
||||
|
@@ -15,7 +15,7 @@ from kasa.protocol import TPLinkSmartHomeProtocol
|
||||
def read_payloads_from_file(file):
|
||||
"""Read the given pcap file and yield json payloads."""
|
||||
pcap = dpkt.pcap.Reader(file)
|
||||
for ts, pkt in pcap:
|
||||
for _ts, pkt in pcap:
|
||||
eth = Ethernet(pkt)
|
||||
if eth.type != ETH_TYPE_IP:
|
||||
continue
|
||||
@@ -44,9 +44,8 @@ def read_payloads_from_file(file):
|
||||
|
||||
try:
|
||||
json_payload = json.loads(decrypted)
|
||||
except (
|
||||
Exception
|
||||
) as ex: # this can happen when the response is split into multiple tcp segments
|
||||
except Exception as ex:
|
||||
# this can happen when the response is split into multiple tcp segments
|
||||
echo(f"[red]Unable to parse payload '{decrypted}', ignoring: {ex}[/red]")
|
||||
continue
|
||||
|
||||
@@ -91,7 +90,8 @@ def parse_pcap(file):
|
||||
context_str = f" [ctx: {context}]" if context else ""
|
||||
|
||||
echo(
|
||||
f"[{is_success}] {direction}{context_str} {module}.{cmd}: {pf(response)}"
|
||||
f"[{is_success}] {direction}{context_str} {module}.{cmd}:"
|
||||
f" {pf(response)}"
|
||||
)
|
||||
|
||||
echo(pf(seen_items))
|
||||
|
@@ -59,13 +59,13 @@ async def main(addrs, rounds):
|
||||
|
||||
if test_gathered:
|
||||
print("=== Testing using gather on all devices ===")
|
||||
for i in range(rounds):
|
||||
for _i in range(rounds):
|
||||
data.append(await _update_concurrently(devs))
|
||||
await asyncio.sleep(2)
|
||||
|
||||
await asyncio.sleep(5)
|
||||
|
||||
for i in range(rounds):
|
||||
for _i in range(rounds):
|
||||
data.append(await _update_sequentially(devs))
|
||||
await asyncio.sleep(2)
|
||||
|
||||
@@ -77,7 +77,7 @@ async def main(addrs, rounds):
|
||||
futs = []
|
||||
data = []
|
||||
locks = {dev: asyncio.Lock() for dev in devs}
|
||||
for i in range(rounds):
|
||||
for _i in range(rounds):
|
||||
for dev in devs:
|
||||
futs.append(asyncio.ensure_future(_update(dev, locks[dev])))
|
||||
|
||||
|
Reference in New Issue
Block a user