mirror of
https://github.com/python-kasa/python-kasa.git
synced 2024-12-23 03:33:35 +00:00
Avoid calling pformat unless debug logging is enabled (#217)
* Avoid calling pformat unless debug logging is enabled * add logging test * isort * check for debug logging * formatting
This commit is contained in:
parent
85f17ab91c
commit
76c1264dc9
@ -90,8 +90,10 @@ class TPLinkSmartHomeProtocol:
|
||||
"""Execute a query on the device and wait for the response."""
|
||||
assert self.writer is not None
|
||||
assert self.reader is not None
|
||||
debug_log = _LOGGER.isEnabledFor(logging.DEBUG)
|
||||
|
||||
_LOGGER.debug("> (%i) %s", len(request), request)
|
||||
if debug_log:
|
||||
_LOGGER.debug("> (%i) %s", len(request), request)
|
||||
self.writer.write(TPLinkSmartHomeProtocol.encrypt(request))
|
||||
await self.writer.drain()
|
||||
|
||||
@ -101,7 +103,8 @@ class TPLinkSmartHomeProtocol:
|
||||
buffer = await self.reader.readexactly(length)
|
||||
response = TPLinkSmartHomeProtocol.decrypt(buffer)
|
||||
json_payload = json.loads(response)
|
||||
_LOGGER.debug("< (%i) %s", len(response), pf(json_payload))
|
||||
if debug_log:
|
||||
_LOGGER.debug("< (%i) %s", len(response), pf(json_payload))
|
||||
return json_payload
|
||||
|
||||
async def close(self):
|
||||
|
@ -1,4 +1,5 @@
|
||||
import json
|
||||
import logging
|
||||
import struct
|
||||
import sys
|
||||
|
||||
@ -64,6 +65,39 @@ async def test_protocol_reconnect(mocker, retry_count):
|
||||
assert response == {"great": "success"}
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info < (3, 8), reason="3.8 is first one with asyncmock")
|
||||
@pytest.mark.parametrize("log_level", [logging.WARNING, logging.DEBUG])
|
||||
async def test_protocol_logging(mocker, caplog, log_level):
|
||||
caplog.set_level(log_level)
|
||||
logging.getLogger("kasa").setLevel(log_level)
|
||||
encrypted = TPLinkSmartHomeProtocol.encrypt('{"great":"success"}')[
|
||||
TPLinkSmartHomeProtocol.BLOCK_SIZE :
|
||||
]
|
||||
|
||||
async def _mock_read(byte_count):
|
||||
nonlocal encrypted
|
||||
if byte_count == TPLinkSmartHomeProtocol.BLOCK_SIZE:
|
||||
return struct.pack(">I", len(encrypted))
|
||||
if byte_count == len(encrypted):
|
||||
return encrypted
|
||||
raise ValueError(f"No mock for {byte_count}")
|
||||
|
||||
def aio_mock_writer(_, __):
|
||||
reader = mocker.patch("asyncio.StreamReader")
|
||||
writer = mocker.patch("asyncio.StreamWriter")
|
||||
mocker.patch.object(reader, "readexactly", _mock_read)
|
||||
return reader, writer
|
||||
|
||||
protocol = TPLinkSmartHomeProtocol("127.0.0.1")
|
||||
mocker.patch("asyncio.open_connection", side_effect=aio_mock_writer)
|
||||
response = await protocol.query({})
|
||||
assert response == {"great": "success"}
|
||||
if log_level == logging.DEBUG:
|
||||
assert "success" in caplog.text
|
||||
else:
|
||||
assert "success" not in caplog.text
|
||||
|
||||
|
||||
def test_encrypt():
|
||||
d = json.dumps({"foo": 1, "bar": 2})
|
||||
encrypted = TPLinkSmartHomeProtocol.encrypt(d)
|
||||
|
Loading…
Reference in New Issue
Block a user