144 lines
4.3 KiB
Python
144 lines
4.3 KiB
Python
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from pigway_frame_flow.drivers.kokonna.client import KokonnaClient
|
|
from pigway_frame_flow.errors import FlowError
|
|
|
|
|
|
def session(body=None, status=200):
|
|
s = Mock()
|
|
s.request.return_value.status_code = status
|
|
s.request.return_value.json.return_value = body or {}
|
|
s.request.return_value.content = b"image"
|
|
return s
|
|
|
|
|
|
def test_query():
|
|
s = session({"nickname": "Test"})
|
|
c = KokonnaClient("secret", session=s)
|
|
assert c.get_device()["nickname"] == "Test"
|
|
kw = s.request.call_args.kwargs
|
|
assert kw["headers"]["Authorization"] == "Bearer secret"
|
|
assert s.request.call_args.args[:2] == (
|
|
"POST",
|
|
"https://api.kokonna.art/openapi/device",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"endpoint,method,args",
|
|
[
|
|
("/listImages", "list_images", ()),
|
|
("/displayImageById", "display", (2,)),
|
|
("/deleteImage", "delete", (2,)),
|
|
("/displayImageByName", "display_by_name", ("art.jpg",)),
|
|
],
|
|
)
|
|
def test_endpoints(endpoint, method, args):
|
|
s = session({"success": True})
|
|
c = KokonnaClient("secret", True, s)
|
|
getattr(c, method)(*args)
|
|
assert s.request.call_args.args[1].endswith(endpoint)
|
|
|
|
|
|
def test_upload_payload(tmp_path):
|
|
path = tmp_path / "x.jpg"
|
|
path.write_bytes(b"jpeg")
|
|
s = session({"id": 10, "counter": 2})
|
|
c = KokonnaClient("secret", True, s)
|
|
assert c.upload(path, "art.jpg") == 10
|
|
assert s.request.call_args.kwargs["json"] == {
|
|
"base64": "anBlZw==",
|
|
"name": "art.jpg",
|
|
}
|
|
|
|
|
|
def test_download_not_leaked():
|
|
s = session()
|
|
c = KokonnaClient("super-secret", session=s)
|
|
assert c.download(12) == b"image"
|
|
assert s.request.call_args.kwargs["allow_redirects"] is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"method,args",
|
|
[
|
|
("upload", (None, "x")),
|
|
("display", (1,)),
|
|
("delete", (1,)),
|
|
("display_by_name", ("x",)),
|
|
],
|
|
)
|
|
def test_write_disabled(method, args):
|
|
s = session()
|
|
c = KokonnaClient("secret", False, s)
|
|
with pytest.raises(FlowError, match="authorization"):
|
|
getattr(c, method)(*args)
|
|
s.request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("write", [False, True])
|
|
def test_timeout_redaction_no_retry(write):
|
|
s = session()
|
|
s.request.side_effect = requests.Timeout("secret https://host/secret")
|
|
c = KokonnaClient("secret", True, s)
|
|
with pytest.raises(FlowError) as e:
|
|
c.display(1) if write else c.get_device()
|
|
assert "secret" not in str(e.value)
|
|
assert e.value.code == ("WRITE_RESULT_UNKNOWN" if write else "CLOUD_UNAVAILABLE")
|
|
assert s.request.call_count == 1
|
|
|
|
|
|
@pytest.mark.parametrize("status", [401, 429, 500, 530])
|
|
def test_cloud_error_sanitized(status):
|
|
s = session({"message": "secret"}, status)
|
|
c = KokonnaClient("secret", True, s)
|
|
with pytest.raises(FlowError) as e:
|
|
c.get_device()
|
|
assert "secret" not in str(e.value)
|
|
|
|
|
|
def test_invalid_upload_id(tmp_path):
|
|
path = tmp_path / "x.jpg"
|
|
path.write_bytes(b"jpeg")
|
|
c = KokonnaClient("secret", True, session({"counter": 2}))
|
|
with pytest.raises(FlowError) as e:
|
|
c.upload(path, "x")
|
|
assert e.value.code == "WRITE_RESULT_UNKNOWN"
|
|
|
|
|
|
def test_rate_budget():
|
|
s = session({"nickname": "x"})
|
|
c = KokonnaClient("secret", session=s)
|
|
for _ in range(18):
|
|
c.get_device()
|
|
with pytest.raises(FlowError) as e:
|
|
c.get_device()
|
|
assert e.value.code == "RATE_LIMITED"
|
|
assert s.request.call_count == 18
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"exception,category",
|
|
[
|
|
(requests.ConnectTimeout, "connect_timeout"),
|
|
(requests.ReadTimeout, "read_timeout"),
|
|
(requests.exceptions.SSLError, "tls_error"),
|
|
(requests.ConnectionError, "connection_error"),
|
|
(requests.Timeout, "timeout"),
|
|
(requests.RequestException, "request_error"),
|
|
],
|
|
)
|
|
def test_transport_failure_records_safe_category_without_replay(exception, category):
|
|
s = session()
|
|
s.request.side_effect = exception("secret https://host/secret")
|
|
c = KokonnaClient("secret", True, s)
|
|
with pytest.raises(FlowError) as error:
|
|
c.display(1)
|
|
assert error.value.code == "WRITE_RESULT_UNKNOWN"
|
|
assert error.value.details == {"transport_error": category}
|
|
assert "secret" not in str(error.value.details)
|
|
assert s.request.call_count == 1
|