pigway-frame-flow/tests/test_client.py

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