Build Pigway Frame Flow with bilingual UI, complete KoKonna driver and safe automation

tests/test_client.py

@@ -0,0 +1,120 @@
+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