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) @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]