pigway-frame-flow/tests/test_client.py

from types import SimpleNamespace
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


@pytest.fixture(autouse=True)
def transport_clock(monkeypatch):
    import pigway_frame_flow.drivers.kokonna.client as module

    clock = SimpleNamespace(now=0.0, waits=[])

    def sleep(seconds):
        clock.waits.append(seconds)
        clock.now += seconds

    monkeypatch.setattr(
        module, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=sleep)
    )
    return clock


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"
    s.request.return_value.headers = {}
    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 if write else 3)


@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(transport_clock):
    s = session({"nickname": "x"})
    c = KokonnaClient("secret", session=s)
    for _ in range(18):
        c.get_device()
    c.get_device()
    assert s.request.call_count == 19
    assert transport_clock.waits == [60]


@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 == (
        "CLOUD_UNAVAILABLE" if category == "connect_timeout" else "WRITE_RESULT_UNKNOWN"
    )
    assert error.value.details["transport_error"] == category
    assert error.value.details["attempts"] == (
        3 if category == "connect_timeout" else 1
    )
    assert "secret" not in str(error.value.details)
    assert s.request.call_count == (3 if category == "connect_timeout" else 1)


def test_connection_establishment_failure_is_unsent_and_bounded(transport_clock):
    from urllib3.exceptions import MaxRetryError, NewConnectionError

    s = session()
    s.request.side_effect = requests.ConnectionError(
        MaxRetryError(None, "/private", NewConnectionError(None, "private"))
    )
    c = KokonnaClient("secret", True, s)
    with pytest.raises(FlowError) as error:
        c.display(1)
    assert error.value.code == "CLOUD_UNAVAILABLE"
    assert error.value.details["write_outcome"] == "not_sent"
    assert error.value.details["transport_error"] == "connect_error"
    assert s.request.call_count == 3
    assert "private" not in str(error.value.details)


@pytest.mark.parametrize("write", [False, True])
def test_connect_timeout_retries_then_succeeds(write, transport_clock):
    s = session({"success": True})
    response = s.request.return_value
    s.request.side_effect = [requests.ConnectTimeout("private"), response]
    c = KokonnaClient("private", True, s)
    c.display(1) if write else c.get_device()
    assert s.request.call_count == 2
    assert transport_clock.waits == [1]


@pytest.mark.parametrize(
    "message,code,kind",
    [
        ("image not found", "NOT_FOUND", "image_not_found"),
        ("imageId is empty", "INVALID_REQUEST", "invalid_image_id"),
        ("can not find robot private", "CONFIG_REQUIRED", "device_key_invalid"),
    ],
)
def test_documented_500_rejection_does_not_lock_or_retry(message, code, kind):
    s = session({"message": message}, 500)
    c = KokonnaClient("private", True, s)
    with pytest.raises(FlowError) as error:
        c.display(1)
    assert error.value.code == code
    assert error.value.details["write_outcome"] == "rejected"
    assert error.value.details["failure_kind"] == kind
    assert "private" not in str(error.value.details)
    assert s.request.call_count == 1


@pytest.mark.parametrize("status", [500, 502, 503, 530])
def test_generic_5xx_read_retries_but_write_never_replays(status):
    for write in [False, True]:
        s = session({"message": "unknown failure private"}, status)
        c = KokonnaClient("private", True, s)
        with pytest.raises(FlowError) as error:
            c.display(1) if write else c.get_device()
        assert s.request.call_count == (1 if write else 3)
        assert error.value.code == (
            "WRITE_RESULT_UNKNOWN" if write else "CLOUD_UNAVAILABLE"
        )
        assert "private" not in str(error.value.details)


def test_rate_limit_honors_retry_after_and_closes_response(transport_clock):
    s = session({"message": "rate limited"}, 429)
    first = s.request.return_value
    first.headers = {"Retry-After": "7"}
    second = session({"success": True}).request.return_value
    s.request.side_effect = [first, second]
    KokonnaClient("private", True, s).display(1)
    assert s.request.call_count == 2
    assert transport_clock.waits == [7]
    first.close.assert_called_once()
    second.close.assert_called_once()


def test_long_retry_after_preserves_cooldown_without_early_request(transport_clock):
    s = session({"message": "rate limited"}, 429)
    s.request.return_value.headers = {"Retry-After": "120"}
    c = KokonnaClient("private", True, s)
    for _ in range(2):
        with pytest.raises(FlowError) as error:
            c.display(1)
        assert error.value.code == "RATE_LIMITED"
    assert s.request.call_count == 1
    assert transport_clock.waits == []


@pytest.mark.parametrize("header", [None, "bad", "nan", "inf"])
def test_invalid_retry_after_defaults_to_one_minute(header):
    from pigway_frame_flow.drivers.kokonna.client import SafeTransport

    assert SafeTransport._retry_after(header) == 60


def test_retry_after_accepts_http_date():
    from datetime import UTC, datetime, timedelta
    from email.utils import format_datetime

    from pigway_frame_flow.drivers.kokonna.client import SafeTransport

    future = format_datetime(datetime.now(UTC) + timedelta(seconds=30), usegmt=True)
    assert 28 <= SafeTransport._retry_after(future) <= 30


def test_invalid_write_response_is_never_retried():
    s = session({"success": False, "message": "private"})
    with pytest.raises(FlowError) as error:
        KokonnaClient("private", True, s).display(1)
    assert error.value.code == "WRITE_RESULT_UNKNOWN"
    assert error.value.details["failure_kind"] == "invalid_response"
    assert s.request.call_count == 1


@pytest.mark.parametrize(
    "failure,expected",
    [
        ("connect_timeout", "CLOUD_UNAVAILABLE"),
        ("business_rejection", "CONFIG_REQUIRED"),
        ("rate_limit", "RATE_LIMITED"),
    ],
)
def test_proven_upload_failures_end_failed_without_device_lock(
    service, monkeypatch, failure, expected
):
    from test_display import submit

    s = session({"message": "can not find robot private"}, 500)
    if failure == "rate_limit":
        s.request.return_value.status_code = 429
    if failure == "connect_timeout":
        s.request.side_effect = requests.ConnectTimeout("private")
    c = KokonnaClient("private", True, s)
    monkeypatch.setattr(service.registry.get("living-room"), "upload", c.upload)
    operation = submit(service)
    service.process_one()
    op = service.operation(operation)
    assert op["status"] == "FAILED"
    assert op["error"]["code"] == expected
    assert not service.blocked("living-room")


@pytest.mark.parametrize("value", [True, False, 0, -1, "12", None])
def test_invalid_upload_id_never_becomes_confirmed(tmp_path, value):
    path = tmp_path / "image.jpg"
    path.write_bytes(b"jpeg")
    s = session({"id": value})
    with pytest.raises(FlowError) as error:
        KokonnaClient("private", True, s).upload(path, "image.jpg")
    assert error.value.code == "WRITE_RESULT_UNKNOWN"
    assert s.request.call_count == 1


def test_ambiguous_failure_after_safe_rejection_stops_retrying(transport_clock):
    s = session({"message": "limited"}, 429)
    first = s.request.return_value
    first.headers = {"Retry-After": "1"}
    s.request.side_effect = [first, requests.ReadTimeout("private")]
    with pytest.raises(FlowError) as error:
        KokonnaClient("private", True, s).display(1)
    assert error.value.code == "WRITE_RESULT_UNKNOWN"
    assert error.value.details["attempts"] == 2
    assert s.request.call_count == 2
    assert transport_clock.waits == [1]