tracer-cloud--opensre/tests/integrations/temporal/test_client.py

import base64
import json
from typing import Any

from integrations.temporal.client import TemporalClient, TemporalConfig


def _encode_status_payload(status: str) -> str:
    """Encode a status name the way Temporal's HTTP API does: the value is a
    JSON-encoded string, base64-encoded (e.g. "Failed" -> "IkZhaWxlZCI=")."""
    return base64.b64encode(json.dumps(status).encode()).decode()


class _FakeResponse:
    def __init__(self, payload: Any, status_code: int = 200) -> None:
        self._payload = payload
        self.status_code = status_code
        self.text = str(payload)[:200]

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            import httpx

            raise httpx.HTTPStatusError(
                f"HTTP {self.status_code}",
                request=None,  # type: ignore[arg-type]
                response=self,  # type: ignore[arg-type]
            )

    def json(self) -> Any:
        return self._payload


def _client() -> TemporalClient:
    return TemporalClient(TemporalConfig(base_url="http://localhost:7233"))


def error_payload() -> dict[str, Any]:
    return {"code": 5, "message": "Namespace not-a-real-namespace is not found.", "details": []}


def test_temporal_client_is_configured():
    assert _client().is_configured is True


def test_temporal_client_is_not_configured():
    assert TemporalClient(TemporalConfig(base_url="")).is_configured is False
    assert (
        TemporalClient(TemporalConfig(base_url="http://localhost:7233", namespace="")).is_configured
        is False
    )


def test_list_workflow_executions_success(monkeypatch):
    fake_payload = {
        "executions": [
            {
                "execution": {"workflowId": "wf-1", "runId": "run-1"},
                "type": {"name": "MyWorkflowType"},
                "startTime": "2024-01-01T00:00:00Z",
                "closeTime": "2024-01-01T00:05:00Z",
                "status": "WORKFLOW_EXECUTION_STATUS_FAILED",
                "taskQueue": "my-queue",
                "historyLength": "150",
                "historySizeBytes": "8192",
            }
        ],
        "nextPageToken": "",
    }
    temporal = _client()

    captured = []

    def fake_get(url, **kwargs):
        captured.append({"url": url, "params": kwargs.get("params")})
        return _FakeResponse(fake_payload)

    monkeypatch.setattr(temporal._client, "get", fake_get)

    response = temporal.list_workflow_executions()
    assert response["success"] is True

    executions = response["executions"]

    # Verify the right endpoint was hit
    assert captured[0]["url"] == "/api/v1/namespaces/default/workflows"
    assert captured[0]["params"]["pageSize"] == 10

    # verify the response
    assert response["total"] == 1
    assert response["next_page_token"] == ""
    assert executions[0]["type"]["name"] == "MyWorkflowType"
    assert executions[0]["status"] == "WORKFLOW_EXECUTION_STATUS_FAILED"


def test_list_workflow_executions_omits_next_page_token(monkeypatch):
    """The live HTTP API omits nextPageToken when there are no more pages.
    The client must not KeyError — it should default to an empty token."""
    fake_payload = {
        "executions": [
            {
                "execution": {"workflowId": "wf-1", "runId": "run-1"},
                "type": {"name": "MyWorkflowType"},
                "status": "WORKFLOW_EXECUTION_STATUS_FAILED",
            }
        ],
        # nextPageToken intentionally absent — matches a real single-page response.
    }
    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", lambda _url, **_k: _FakeResponse(fake_payload))

    response = temporal.list_workflow_executions()
    assert response["success"] is True
    assert response["total"] == 1
    assert response["next_page_token"] == ""


def test_list_workflow_executions_failure(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        return _FakeResponse(error_payload(), 404)

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.list_workflow_executions()

    assert response["success"] is False


def test_list_workflow_executions_exception(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        raise Exception("unexpected exception")

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.list_workflow_executions()

    assert response["success"] is False
    assert response["error"] == "unexpected exception"


def test_get_workflow_history_success(monkeypatch):
    fake_payload = {
        "history": {
            "events": [
                {
                    "eventId": "1",
                    "eventTime": "2024-01-15T10:00:00Z",
                    "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED",
                    "taskId": "1048576",
                    "workerMayIgnore": False,
                },
                {
                    "eventId": "2",
                    "eventTime": "2024-01-15T10:00:01Z",
                    "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED",
                    "taskId": "1048577",
                    "workerMayIgnore": False,
                },
            ]
        },
        "nextPageToken": "",
        "archived": False,
    }

    captured = []

    def fake_get(url, **kwargs) -> _FakeResponse:
        captured.append({"url": url, "params": kwargs.get("params")})
        return _FakeResponse(fake_payload)

    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", fake_get)

    response = temporal.get_workflow_history("wf-1", "run-1")
    assert response["success"] is True

    # Verify the right endpoint was hit
    assert captured[0]["url"] == "/api/v1/namespaces/default/workflows/wf-1/history"
    assert captured[0]["params"]["pageSize"] == 10
    assert captured[0]["params"]["execution.runId"] == "run-1"

    # verify the response
    assert response["total"] == 2
    assert response["next_page_token"] == ""
    assert response["archived"] is False


def test_get_workflow_history_omits_archived_and_token(monkeypatch):
    """A single-page, non-archived history omits both nextPageToken and
    archived on the wire. The client must default them instead of KeyError-ing."""
    fake_payload = {
        "history": {
            "events": [
                {"eventId": "1", "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED"},
            ]
        },
        # nextPageToken and archived intentionally absent.
    }
    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", lambda _url, **_k: _FakeResponse(fake_payload))

    response = temporal.get_workflow_history("wf-1", "run-1")
    assert response["success"] is True
    assert response["total"] == 1
    assert response["next_page_token"] == ""
    assert response["archived"] is False


def test_get_workflow_history_encodes_workflow_id_with_slashes(monkeypatch):
    """Workflow IDs may contain '/' (e.g. 'order-service/order-123'). The client
    must percent-encode it into a single path segment, or the Temporal frontend
    parses the extra segment as a different route and returns 404."""
    captured = []

    def fake_get(url, **_kwargs):
        captured.append(url)
        return _FakeResponse({"history": {"events": []}})

    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", fake_get)

    temporal.get_workflow_history("order-service/order-123", "run-1")
    assert captured[0] == "/api/v1/namespaces/default/workflows/order-service%2Forder-123/history"


def test_get_workflow_history_failure(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        return _FakeResponse(error_payload(), 404)

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.get_workflow_history("wf-1", "run-1")

    assert response["success"] is False


def test_get_workflow_history_exception(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        raise Exception("unexpected exception")

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.get_workflow_history("wf-1", "run-1")

    assert response["success"] is False
    assert response["error"] == "unexpected exception"


def test_describe_task_queue_success(monkeypatch):
    fake_payload = {
        "pollers": [
            {
                "lastAccessTime": "2024-01-15T10:05:00Z",
                "identity": "worker-1@host-abc",
                "ratePerSecond": 100.0,
            },
            {
                "lastAccessTime": "2024-01-15T10:04:55Z",
                "identity": "worker-2@host-def",
                "ratePerSecond": 100.0,
            },
        ],
        "stats": {
            "approximateBacklogCount": "42",
            "approximateBacklogAge": "30.5s",
            "tasksAddRate": 5.2,
            "tasksDispatchRate": 4.8,
        },
    }
    captured = []

    def fake_get(url, **kwargs) -> _FakeResponse:
        captured.append({"url": url, "params": kwargs.get("params")})
        return _FakeResponse(fake_payload)

    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", fake_get)

    response = temporal.describe_task_queue("my-queue")
    assert response["success"] is True
    assert response["stats"]["approximateBacklogCount"] == "42"
    assert response["total"] == 2

    # Verify the right endpoint was hit
    assert captured[0]["url"] == "/api/v1/namespaces/default/task-queues/my-queue"
    assert captured[0]["params"]["reportStats"] is True
    assert captured[0]["params"]["taskQueueType"] == "TASK_QUEUE_TYPE_WORKFLOW"


def test_describe_task_queue_omits_pollers(monkeypatch):
    """When no worker is polling, the live API omits the pollers array entirely.
    The client must default to [] (total 0) rather than KeyError."""
    fake_payload = {
        "stats": {"approximateBacklogAge": "0s"},
        # pollers intentionally absent — no active workers right now.
    }
    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", lambda _url, **_k: _FakeResponse(fake_payload))

    response = temporal.describe_task_queue("idle-queue")
    assert response["success"] is True
    assert response["pollers"] == []
    assert response["total"] == 0


def test_describe_task_queue_encodes_name_with_slashes(monkeypatch):
    """Task queue names are free-form strings and may contain '/'. Encode them
    into a single path segment so the request hits the right route."""
    captured = []

    def fake_get(url, **_kwargs):
        captured.append(url)
        return _FakeResponse({"stats": {}})

    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", fake_get)

    temporal.describe_task_queue("payments/charge-queue")
    assert captured[0] == "/api/v1/namespaces/default/task-queues/payments%2Fcharge-queue"


def test_describe_task_queue_failure(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        return _FakeResponse(error_payload(), 404)

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.describe_task_queue("my-queue")

    assert response["success"] is False


def test_describe_task_queue_exception(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        raise Exception("unexpected exception")

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.describe_task_queue("my-queue")

    assert response["success"] is False
    assert response["error"] == "unexpected exception"


def test_get_namespace_info_success(monkeypatch):
    ns_payload = {
        "namespaceInfo": {
            "name": "default",
            "state": "NAMESPACE_STATE_REGISTERED",
            "description": "Default namespace",
            "ownerEmail": "team@example.com",
            "id": "ns-id-123",
        },
        "config": {},
        "isGlobalNamespace": False,
    }
    # The real HTTP API base64-encodes each status name as a Temporal Payload.
    count_payload = {
        "count": "58",
        "groups": [
            {"groupValues": [{"data": _encode_status_payload("Running")}], "count": "45"},
            {"groupValues": [{"data": _encode_status_payload("Failed")}], "count": "8"},
            {"groupValues": [{"data": _encode_status_payload("TimedOut")}], "count": "5"},
        ],
    }

    captured = []

    def fake_get(url, **kwargs):
        captured.append({"url": url, "params": kwargs.get("params")})
        if "workflow-count" in url:
            return _FakeResponse(count_payload)
        return _FakeResponse(ns_payload)

    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", fake_get)

    response = temporal.get_namespace_info()
    assert response["success"] is True

    # Verify correct endpoints were hit
    assert captured[0]["url"] == "/api/v1/namespaces/default"
    assert captured[1]["url"] == "/api/v1/namespaces/default/workflow-count"
    assert captured[1]["params"]["query"] == "GROUP BY ExecutionStatus"

    # Verify response shape: groups are decoded from base64 and flattened to
    # [{"status", "count"}] — the LLM never sees raw Payload encoding.
    assert response["name"] == "default"
    assert response["state"] == "NAMESPACE_STATE_REGISTERED"
    assert response["workflow_count"] == "58"
    assert response["groups"] == [
        {"status": "Running", "count": "45"},
        {"status": "Failed", "count": "8"},
        {"status": "TimedOut", "count": "5"},
    ]


def test_get_namespace_info_failure(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        return _FakeResponse(error_payload(), 404)

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.get_namespace_info()

    assert response["success"] is False


def test_get_namespace_info_exception(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        raise Exception("connection refused")

    monkeypatch.setattr(temporal._client, "get", fake_get)
    response = temporal.get_namespace_info()

    assert response["success"] is False
    assert response["error"] == "connection refused"


def test_probe_access_success(monkeypatch):
    payload = {
        "namespaceInfo": {
            "name": "default",
            "state": "NAMESPACE_STATE_REGISTERED",
            "description": "Default namespace",
            "ownerEmail": "team@example.com",
            "id": "ns-id-123",
        },
        "config": {},
        "isGlobalNamespace": False,
    }

    def fake_get(_url, **_kwargs):
        return _FakeResponse(payload)

    temporal = _client()
    monkeypatch.setattr(temporal._client, "get", fake_get)

    result = temporal.probe_access()

    assert result.ok is True


def test_probe_access_failure(monkeypatch):
    temporal = _client()

    def fake_get(_url, **_kwargs):
        return _FakeResponse(error_payload(), 404)

    monkeypatch.setattr(temporal._client, "get", fake_get)
    result = temporal.probe_access()

    assert result.ok is False


def test_probe_access_on_unconfigured_client(monkeypatch):
    temporal = TemporalClient(TemporalConfig(base_url="", namespace=""))
    result = temporal.probe_access()
    assert result.ok is False