Artemis-Eapil/tests/unit/clients/test_native_transport.py

"""Native input is one-shot and remains owned through uncertain transport cleanup."""

import json
from types import SimpleNamespace
import urllib.error
from unittest.mock import Mock

import pytest

from artemis.clients import accessibility_client as module
from artemis.clients.accessibility_client import AccessibilityClient
from artemis.clients.screen_client_factory import FallbackScreenClient
from artemis.runtime.helper_manager import HelperSession
from artemis.runtime.native_protocol import input_completion_seconds


@pytest.fixture
def native(monkeypatch):
    manager = Mock()
    client = AccessibilityClient("test-device", manager=manager)
    client._session = HelperSession(
        serial="test-device",
        local_port=40001,
        transport_id="1",
        version_code=12,
        version_name="1.3.0",
        owns_forward=True,
        token="test-only",
        protocol_version=3,
        capabilities=("inspect_control", "control_action", "touch_gesture", "cancel_gesture"),
    )
    clock = [0.0]
    sleeps = []

    def sleep(seconds):
        sleeps.append(seconds)
        clock[0] += seconds

    monkeypatch.setattr(module, "time", SimpleNamespace(monotonic=lambda: clock[0], sleep=sleep))
    client._http_once = Mock(return_value=b'{"success":true,"execution":"completed","code":"OK"}')
    return client, manager, clock, sleeps


def test_exactly_one_request_and_command_cannot_be_overridden(native):
    client, manager, _, sleeps = native
    receipt = {
        "success": True,
        "execution": "observed",
        "code": "OK",
        "node": {"selection_start": 3},
    }
    client._http_once.return_value = json.dumps(receipt).encode()
    assert client.native_request("inspect_control", {"cmd": "swipe"}, timeout=0.02) == receipt
    client._http_once.assert_called_once_with(
        client.session, "/action", {"cmd": "inspect_control"}, 0.02
    )
    assert sleeps == []
    manager.attach.assert_not_called()


@pytest.mark.parametrize(
    "status,execution,attempts,repairs",
    [
        (400, "not_started", 1, 0),
        (401, "not_started", 1, 0),
        (403, "not_started", 1, 0),
        (503, "unknown", 1, 0),
    ],
)
def test_http_rejection_never_replays_native_input(native, status, execution, attempts, repairs):
    client, manager, _, sleeps = native
    client._http_once.side_effect = urllib.error.HTTPError(
        "http://test", status, "rejected", {}, None
    )
    result = client.native_request("control_action", {"action": "click"}, timeout=0.02)
    assert result["execution"] == execution and not result["success"]
    assert client._http_once.call_count == attempts
    assert client._http_once.call_args.args[-1] >= 5
    assert manager.push_token.call_count == repairs
    manager.reattach.assert_not_called()
    assert bool(sleeps) == (execution == "unknown")


@pytest.mark.parametrize(
    "response", [b"broken", b"[]", b"{}", b'{"success":true,"execution":"unknown"}']
)
def test_uncertain_write_drains_without_retry(native, response):
    client, _, clock, sleeps = native
    client._http_once.return_value = response
    result = client.native_request("control_action", {"action": "click"})
    assert result["execution"] == "unknown"
    assert sleeps == [5.0] and clock[0] == 5.0
    client._http_once.assert_called_once()


def test_partial_transport_elapsed_time_counts_toward_input_drain(native):
    client, _, clock, sleeps = native

    def broken(*args):
        clock[0] = 2
        raise OSError("response lost after dispatch")

    client._http_once.side_effect = broken
    assert client.native_request("control_action", {})["execution"] == "unknown"
    assert sleeps == [3.0] and clock[0] == 5
    client._http_once.assert_called_once()


@pytest.mark.parametrize("command", ["inspect_control", "cancel_gesture"])
def test_non_input_calls_do_not_drain_on_failure(native, command):
    client, _, _, sleeps = native
    client._http_once.side_effect = OSError("offline")
    result = client.native_request(command, {}, timeout=0.02)
    assert not result["success"] and sleeps == []
    client._http_once.assert_called_once()


def test_no_session_or_capability_cannot_trigger_attach_or_input(native):
    client, manager, _, sleeps = native
    client._session.capabilities = ()
    assert client.native_request("touch_gesture", {})["code"] == "UNSUPPORTED"
    client._session = None
    assert client.native_capabilities == frozenset()
    assert client.native_request("control_action", {})["execution"] == "not_started"
    client._http_once.assert_not_called()
    manager.attach.assert_not_called()
    assert sleeps == []


@pytest.mark.parametrize(
    "duration,budget", [(0, 5), (1000, 6), (10000, 15), (999999, 15), (None, 15)]
)
def test_touch_completion_budget_is_bounded(duration, budget):
    assert input_completion_seconds("touch_gesture", {"frames": [{"at_ms": duration}]}) == budget


@pytest.mark.parametrize("backend", ["helper", "uiautomator"])
def test_native_input_never_uses_fallback_dispatch(backend):
    client = object.__new__(FallbackScreenClient)
    client._active_backend = backend
    client._helper = Mock()
    client._helper.native_capabilities = frozenset({"control_action"})
    unknown = {"success": False, "execution": "unknown", "code": "TRANSPORT_ERROR"}
    client._helper.native_request.return_value = unknown
    client._call = Mock(side_effect=AssertionError("fallback is forbidden"))
    result = client.native_request("control_action", {})
    if backend == "helper":
        assert result is unknown and client.native_capabilities == frozenset({"control_action"})
        client._helper.native_request.assert_called_once()
    else:
        assert result["execution"] == "not_started" and client.native_capabilities == frozenset()
        client._helper.native_request.assert_not_called()
    client._call.assert_not_called()