diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py new file mode 100644 index 0000000..3e6198b --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -0,0 +1 @@ +"""Private evaluation management adapters.""" diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/api.py b/packages/client/src/launchdarkly_ai_server/evaluations/api.py new file mode 100644 index 0000000..a751bd5 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/api.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +import json +import random +import time +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import UTC, datetime +from email.utils import parsedate_to_datetime +from typing import Any, Protocol + +DEFAULT_BASE_URI = "https://app.launchdarkly.com" + + +class EvaluationsError(Exception): + """Base error for the evaluations harness.""" + + +class LDApiError(EvaluationsError): + """A non-2xx response from the LaunchDarkly API.""" + + def __init__(self, status: int, method: str, path: str, body: str) -> None: + super().__init__( + f"LaunchDarkly API {method} {path} failed with {status}: {body}" + ) + self.status = status + self.method = method + self.path = path + self.body = body + + +@dataclass +class HttpResponse: + status: int + body: str + headers: dict[str, str] = field(default_factory=dict) + + +class Transport(Protocol): + """Seam the API client sends requests through; replaced in tests.""" + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: ... + + +def urllib_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + request = urllib.request.Request(url, data=body, headers=headers, method=method) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return HttpResponse( + status=response.status, + body=response.read().decode("utf-8"), + headers={k.lower(): v for k, v in response.headers.items()}, + ) + except urllib.error.HTTPError as error: + return HttpResponse( + status=error.code, + body=error.read().decode("utf-8"), + headers={k.lower(): v for k, v in error.headers.items()}, + ) + + +class LDApiClient: + """Minimal retrying client for the LaunchDarkly public management API.""" + + def __init__( + self, + api_token: str, + base_uri: str = DEFAULT_BASE_URI, + transport: Transport = urllib_transport, + timeout: float = 30.0, + max_retries: int = 3, + sleep: Callable[[float], None] = time.sleep, + random_value: Callable[[], float] = random.random, + ) -> None: + self.api_token = api_token + self.base_uri = base_uri.rstrip("/") + self._transport = transport + self._timeout = timeout + self._max_retries = max(0, max_retries) + self._sleep = sleep + self._random_value = random_value + + def url_for(self, path: str, params: dict[str, Any] | None = None) -> str: + url = f"{self.base_uri}/api/v2/{path.lstrip('/')}" + if params: + query = {k: str(v) for k, v in params.items() if v is not None} + if query: + url = f"{url}?{urllib.parse.urlencode(query)}" + return url + + def _retry_delay(self, attempt: int, response: HttpResponse | None = None) -> float: + if response is not None: + retry_after = response.headers.get("retry-after") or response.headers.get( + "Retry-After" + ) + if retry_after: + try: + return max(0.0, float(retry_after)) + except ValueError: + try: + when: datetime = parsedate_to_datetime(retry_after) + now = datetime.now(UTC) + return max(0.0, (when - now).total_seconds()) + except (TypeError, ValueError, OverflowError): + pass + exponential = float(min(30.0, 0.5 * (2**attempt))) + jitter = float(self._random_value()) * min(1.0, exponential) + return exponential + jitter + + def request( + self, + method: str, + path: str, + body: Any = None, + params: dict[str, Any] | None = None, + ) -> Any: + headers = { + "Authorization": self.api_token, + "Accept": "application/json", + "User-Agent": "launchdarkly-ai-evaluations-python", + } + payload: bytes | None = None + if body is not None: + headers["Content-Type"] = "application/json" + payload = json.dumps(body).encode("utf-8") + + retryable_method = method.upper() in {"GET", "HEAD", "PUT", "DELETE", "OPTIONS"} + response: HttpResponse | None = None + for attempt in range(self._max_retries + 1): + try: + response = self._transport( + method, self.url_for(path, params), headers, payload, self._timeout + ) + except (TimeoutError, urllib.error.URLError) as error: + if not retryable_method or attempt >= self._max_retries: + suffix = " after retries" if retryable_method else "" + raise EvaluationsError( + f"LaunchDarkly API {method} {path} failed{suffix}: {error}" + ) from error + self._sleep(self._retry_delay(attempt)) + continue + + retryable = response.status == 429 or response.status >= 500 + if retryable_method and retryable and attempt < self._max_retries: + self._sleep(self._retry_delay(attempt, response)) + continue + break + + if response is None: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} returned no response" + ) + if response.status < 200 or response.status >= 300: + raise LDApiError(response.status, method, path, response.body) + if not response.body: + return None + try: + return json.loads(response.body) + except json.JSONDecodeError as error: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} returned invalid JSON" + ) from error + + def get(self, path: str, params: dict[str, Any] | None = None) -> Any: + return self.request("GET", path, params=params) + + def post(self, path: str, body: Any = None) -> Any: + return self.request("POST", path, body=body) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py new file mode 100644 index 0000000..146cf79 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -0,0 +1,300 @@ +from __future__ import annotations + +import asyncio +import time +import urllib.parse +from collections.abc import Callable, Mapping +from typing import Any + +from ..types import NativeTool +from .api import EvaluationsError, LDApiClient, LDApiError +from .types import ( + DatasetRef, + DatasetRow, + EvaluationRef, + EvaluationRunRef, + GenerationConfig, + ResolvedTool, + RunSummary, +) + +DATASET_PAGE_SIZE = 200 + +ToolImplementation = Callable[..., Any] | NativeTool + + +def _segment(value: str) -> str: + return urllib.parse.quote(value, safe="") + + +def _mapping(value: Any, *, description: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise EvaluationsError( + f"LaunchDarkly returned an invalid {description} response" + ) + return value + + +def _required_string(data: Mapping[str, Any], key: str, description: str) -> str: + value = data.get(key) + if not isinstance(value, str) or not value: + raise EvaluationsError( + f"LaunchDarkly {description} response is missing string field {key!r}" + ) + return value + + +class EvaluationsRunner: + """Private API operations and orchestration used by EvaluationsModule.run().""" + + def __init__(self, api: LDApiClient) -> None: + self._api = api + + def _resolve_tools( + self, + project_key: str, + tools: Mapping[str, ToolImplementation], + ) -> dict[str, ResolvedTool]: + resolved: dict[str, ResolvedTool] = {} + for key, implementation in tools.items(): + if not callable(implementation) and not isinstance( + implementation, NativeTool + ): + raise EvaluationsError( + f"Tool {key!r} must be callable or a NativeTool instance" + ) + path = f"projects/{_segment(project_key)}/ai-tools/{_segment(key)}" + try: + raw = _mapping(self._api.get(path), description=f"tool {key!r}") + except LDApiError as error: + if error.status == 404: + raise EvaluationsError( + f"LaunchDarkly AI tool {key!r} was not found in project {project_key!r}" + ) from error + raise + version = raw.get("version") + if not isinstance(version, int): + raise EvaluationsError( + f"LaunchDarkly AI tool {key!r} has no integer version" + ) + schema = raw.get("schema") + if not isinstance(schema, Mapping): + schema = {} + resolved[key] = ResolvedTool( + key=key, + version=version, + description=str(raw.get("description") or ""), + schema=dict(schema), + ) + return resolved + + def _fetch_dataset(self, project_key: str, dataset_key: str) -> DatasetRef: + path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}" + try: + raw = _mapping(self._api.get(path), description=f"dataset {dataset_key!r}") + except LDApiError as error: + if error.status == 404: + raise EvaluationsError( + f"LaunchDarkly dataset {dataset_key!r} was not found in project {project_key!r}" + ) from error + raise + dataset_id = _required_string(raw, "id", "dataset") + response_key = raw.get("key", raw.get("name", dataset_key)) + return DatasetRef(id=dataset_id, key=str(response_key)) + + def _fetch_dataset_rows_page( + self, + project_key: str, + dataset_key: str, + *, + offset: int, + ) -> Mapping[str, Any]: + path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}/rows" + return _mapping( + self._api.get( + path, + params={ + "mode": "all", + "limit": DATASET_PAGE_SIZE, + "offset": offset, + }, + ), + description=f"rows for dataset {dataset_key!r}", + ) + + def _get_dataset_rows(self, project_key: str, dataset_key: str) -> list[DatasetRow]: + rows: list[DatasetRow] = [] + offset = 0 + total: int | None = None + while total is None or len(rows) < total: + page = self._fetch_dataset_rows_page( + project_key, dataset_key, offset=offset + ) + items = page.get("items") + page_total = page.get("totalCount") + if not isinstance(items, list) or not isinstance(page_total, int): + raise EvaluationsError( + f"LaunchDarkly returned invalid rows for dataset {dataset_key!r}" + ) + total = page_total + if not items: + break + for item_value in items: + item = _mapping(item_value, description="dataset row") + row_index = item.get("rowIndex") + if not isinstance(row_index, int): + raise EvaluationsError( + "A dataset row is missing its integer rowIndex" + ) + variables_value = item.get("variables") + variables = ( + dict(variables_value) + if isinstance(variables_value, Mapping) + else {} + ) + input_value = item.get("input") + expected_value = item.get("expectedOutput") + row_input = input_value if isinstance(input_value, str) else None + expected_output = ( + expected_value if isinstance(expected_value, str) else None + ) + metadata_value = item.get("metadata") + rows.append( + DatasetRow( + row_index=row_index, + input=row_input, + expected_output=expected_output, + variables=variables, + metadata=( + dict(metadata_value) + if isinstance(metadata_value, Mapping) + else None + ), + ) + ) + offset += len(items) + if not rows: + raise EvaluationsError(f"Dataset {dataset_key!r} is empty") + if total is not None and len(rows) != total: + raise EvaluationsError( + f"Dataset {dataset_key!r} returned {len(rows)} of {total} rows" + ) + return rows + + def _create_evaluation( + self, + project_key: str, + key: str, + generation: GenerationConfig, + tools: Mapping[str, ResolvedTool], + ) -> EvaluationRef: + body: dict[str, Any] = { + "name": key, + "generationProvider": generation["provider"], + "generationModel": generation["model"], + } + if "parameters" in generation: + body["parameters"] = generation["parameters"] + if "instructions" in generation: + body["messages"] = [ + {"role": "system", "content": generation["instructions"]} + ] + elif "messages" in generation: + body["messages"] = generation["messages"] + else: + body["messages"] = [] + if "prompt_snippets" in generation: + body["promptSnippets"] = generation["prompt_snippets"] + if "output_format" in generation: + body["outputFormat"] = generation["output_format"] + if tools: + body["tools"] = [ + {"key": tool.key, "version": tool.version} for tool in tools.values() + ] + + path = f"projects/{_segment(project_key)}/evaluations" + raw = _mapping(self._api.post(path, body=body), description="evaluation") + evaluation_id = _required_string(raw, "id", "evaluation") + response_key = raw.get("name", raw.get("label", key)) + version = raw.get("version") + return EvaluationRef( + id=evaluation_id, + key=str(response_key), + version=version if isinstance(version, int) else None, + ) + + def _create_evaluation_run( + self, + project_key: str, + evaluation_id: str, + dataset_id: str, + ) -> EvaluationRunRef: + path = ( + f"projects/{_segment(project_key)}/evaluations/" + f"{_segment(evaluation_id)}/runs" + ) + raw = _mapping( + self._api.post( + path, + body={ + "source": "api", + "datasetId": dataset_id, + }, + ), + description="evaluation run", + ) + return self._run_ref(raw) + + def _run_ref(self, raw: Mapping[str, Any]) -> EvaluationRunRef: + return EvaluationRunRef( + id=_required_string(raw, "id", "evaluation run"), + evaluation_id=_required_string(raw, "evaluationId", "evaluation run"), + state=_required_string(raw, "state", "evaluation run"), + status_reason=( + str(raw["statusReason"]) + if raw.get("statusReason") is not None + else None + ), + ) + + async def _poll_run( + self, + project_key: str, + evaluation_id: str, + run_id: str, + timeout: float, + ) -> EvaluationRunRef: + path = ( + f"projects/{_segment(project_key)}/evaluations/{_segment(evaluation_id)}" + f"/runs/{_segment(run_id)}" + ) + deadline = time.monotonic() + timeout + delay = 0.25 + while True: + run = self._run_ref( + _mapping(self._api.get(path), description="evaluation run") + ) + if run.state == "COMPLETE": + return run + if run.state in {"CANCELLED", "TEMPORARY_ERROR", "PERMANENT_ERROR"}: + reason = f": {run.status_reason}" if run.status_reason else "" + raise EvaluationsError( + f"Evaluation run {run_id!r} failed in state {run.state}{reason}" + ) + if time.monotonic() >= deadline: + raise EvaluationsError( + f"Evaluation run {run_id!r} is still in progress after {timeout} seconds" + ) + await asyncio.sleep(delay) + delay = min(5.0, delay * 2) + + def _get_summary( + self, project_key: str, evaluation_id: str, run_id: str + ) -> RunSummary: + path = ( + f"projects/{_segment(project_key)}/evaluations/{_segment(evaluation_id)}" + f"/runs/{_segment(run_id)}/summary" + ) + return RunSummary.from_wire( + _mapping(self._api.get(path), description="evaluation run summary") + ) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py new file mode 100644 index 0000000..4033a3f --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -0,0 +1,120 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any, TypedDict + + +@dataclass +class Usage: + """Token counts for one generation, using the ingest wire field names.""" + + input_tokens: int + output_tokens: int + + def to_wire(self) -> dict[str, int]: + return { + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + } + + @classmethod + def from_wire(cls, data: Mapping[str, Any]) -> Usage: + return cls( + input_tokens=int(data.get("input_tokens") or 0), + output_tokens=int(data.get("output_tokens") or 0), + ) + + +class GenerationConfig(TypedDict, total=False): + """Generation settings stored on the evaluation and passed to its handler.""" + + provider: str + model: str + parameters: dict[str, Any] + instructions: str + messages: list[dict[str, Any]] + prompt_snippets: dict[str, str] + output_format: dict[str, Any] + + +@dataclass +class DatasetRef: + """Identifiers returned when resolving a dataset by key.""" + + id: str + key: str + + +@dataclass +class DatasetRow: + """A rendered dataset row ready for handler invocation and ingest.""" + + row_index: int + input: str | None = None + expected_output: str | None = None + variables: dict[str, Any] = field(default_factory=dict) + metadata: dict[str, Any] | None = None + + +@dataclass +class ResolvedTool: + """The schema and pinned version returned by the LaunchDarkly tool API.""" + + key: str + version: int + description: str = "" + schema: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class EvaluationRef: + """Identifiers returned after creating an evaluation.""" + + id: str + key: str + version: int | None = None + + +@dataclass +class EvaluationRunRef: + """Identifiers and state returned by the evaluation-run API.""" + + id: str + evaluation_id: str + state: str + status_reason: str | None = None + + +@dataclass +class RunSummary: + """Row counts for a finished evaluation run.""" + + total_rows: int = 0 + passed_rows: int = 0 + failed_rows: int = 0 + error_rows: int = 0 + pending_rows: int = 0 + + @classmethod + def from_wire(cls, data: Mapping[str, Any] | None) -> RunSummary: + data = data or {} + counts_value = data.get("statusCounts") + counts = counts_value if isinstance(counts_value, Mapping) else data + return cls( + total_rows=int(counts.get("total", counts.get("total_rows", 0)) or 0), + passed_rows=int(counts.get("passed", counts.get("passed_rows", 0)) or 0), + failed_rows=int(counts.get("failed", counts.get("failed_rows", 0)) or 0), + error_rows=int(counts.get("error", counts.get("error_rows", 0)) or 0), + pending_rows=int(counts.get("pending", counts.get("pending_rows", 0)) or 0), + ) + + +@dataclass +class EvalRunResult: + """The result of an evaluation run, derived from its row summary.""" + + passed: bool + url: str + run_id: str + summary: RunSummary diff --git a/packages/client/tests/test_evaluation_management.py b/packages/client/tests/test_evaluation_management.py new file mode 100644 index 0000000..a6ba25a --- /dev/null +++ b/packages/client/tests/test_evaluation_management.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from launchdarkly_ai_server.evaluations.api import EvaluationsError +from launchdarkly_ai_server.evaluations.runner import EvaluationsRunner + + +class RecordingApi: + def __init__(self) -> None: + self.posts: list[tuple[str, dict[str, Any]]] = [] + self.gets: list[tuple[str, dict[str, Any] | None]] = [] + self.responses: list[dict[str, Any]] = [] + + def post(self, path: str, *, body: dict[str, Any]) -> dict[str, Any]: + self.posts.append((path, body)) + return self.responses.pop(0) + + def get(self, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: + self.gets.append((path, params)) + return self.responses.pop(0) + + +def test_evaluation_create_sends_structured_output_format() -> None: + api = RecordingApi() + api.responses = [{"id": "evaluation-id", "name": "evaluation-key"}] + runner = EvaluationsRunner(api) # type: ignore[arg-type] + output_format = { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], + } + + runner._create_evaluation( + "project", + "evaluation-key", + { + "provider": "OpenAI", + "model": "gpt-4o", + "output_format": output_format, + }, + {}, + ) + + assert api.posts == [ + ( + "projects/project/evaluations", + { + "name": "evaluation-key", + "generationProvider": "OpenAI", + "generationModel": "gpt-4o", + "messages": [], + "outputFormat": output_format, + }, + ) + ] + + +def test_run_create_uses_landed_api_source_contract() -> None: + api = RecordingApi() + api.responses = [ + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"} + ] + runner = EvaluationsRunner(api) # type: ignore[arg-type] + + run = runner._create_evaluation_run("project", "evaluation-id", "dataset-id") + + assert run.id == "run-id" + assert api.posts == [ + ( + "projects/project/evaluations/evaluation-id/runs", + {"source": "api", "datasetId": "dataset-id"}, + ) + ] + + +def test_flat_public_summary_includes_pending_rows() -> None: + api = RecordingApi() + api.responses = [{"total": 4, "passed": 1, "failed": 1, "error": 1, "pending": 1}] + runner = EvaluationsRunner(api) # type: ignore[arg-type] + + summary = runner._get_summary("project", "evaluation-id", "run-id") + + assert summary.total_rows == 4 + assert summary.pending_rows == 1 + assert api.gets[0][0].endswith("/runs/run-id/summary") + + +@pytest.mark.asyncio +async def test_poll_uses_lifecycle_state_until_complete( + monkeypatch: pytest.MonkeyPatch, +) -> None: + api = RecordingApi() + api.responses = [ + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "COMPLETE"}, + ] + runner = EvaluationsRunner(api) # type: ignore[arg-type] + + async def no_sleep(delay: float) -> None: + assert delay == 0.25 + + monkeypatch.setattr("asyncio.sleep", no_sleep) + run = await runner._poll_run("project", "evaluation-id", "run-id", 1) + + assert run.state == "COMPLETE" + assert len(api.gets) == 2 + + +@pytest.mark.asyncio +async def test_poll_preserves_terminal_error_reason() -> None: + api = RecordingApi() + api.responses = [ + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "PERMANENT_ERROR", + "statusReason": "invalid dataset", + } + ] + runner = EvaluationsRunner(api) # type: ignore[arg-type] + + with pytest.raises(EvaluationsError, match="invalid dataset"): + await runner._poll_run("project", "evaluation-id", "run-id", 1) diff --git a/packages/client/tests/test_evaluations_management_api.py b/packages/client/tests/test_evaluations_management_api.py new file mode 100644 index 0000000..5f26b1b --- /dev/null +++ b/packages/client/tests/test_evaluations_management_api.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from launchdarkly_ai_server.evaluations.api import ( + DEFAULT_BASE_URI, + EvaluationsError, + HttpResponse, + LDApiClient, + LDApiError, +) +from launchdarkly_ai_server.evaluations.types import EvalRunResult, RunSummary, Usage + + +class RecordingTransport: + """Mocked LD API — records requests and replays canned responses.""" + + def __init__(self, responses: list[HttpResponse] | None = None) -> None: + self.requests: list[dict[str, Any]] = [] + self.responses = responses or [HttpResponse(status=200, body="{}")] + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + self.requests.append( + { + "method": method, + "url": url, + "headers": headers, + "body": json.loads(body) if body else None, + "timeout": timeout, + } + ) + index = min(len(self.requests) - 1, len(self.responses) - 1) + return self.responses[index] + + +def failing_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + raise AssertionError("no network I/O expected") + + +def test_requests_carry_token_auth_and_json_body() -> None: + transport = RecordingTransport([HttpResponse(status=201, body='{"key": "run-1"}')]) + client = LDApiClient(api_token="api-token", transport=transport) + + result = client.post("projects/proj/evaluations", body={"key": "support-qa"}) + + assert result == {"key": "run-1"} + request = transport.requests[0] + assert request["method"] == "POST" + assert request["url"] == f"{DEFAULT_BASE_URI}/api/v2/projects/proj/evaluations" + assert request["headers"]["Authorization"] == "api-token" + assert request["headers"]["Content-Type"] == "application/json" + assert request["body"] == {"key": "support-qa"} + + +def test_get_encodes_query_params_and_omits_none() -> None: + transport = RecordingTransport([HttpResponse(status=200, body='{"items": []}')]) + client = LDApiClient( + api_token="api-token", base_uri="https://ld.example.com", transport=transport + ) + + client.get("projects/proj/datasets/golden", params={"limit": 50, "offset": None}) + + request = transport.requests[0] + assert ( + request["url"] + == "https://ld.example.com/api/v2/projects/proj/datasets/golden?limit=50" + ) + assert "Content-Type" not in request["headers"] + + +def test_rate_limit_retries_and_honors_retry_after() -> None: + transport = RecordingTransport( + [ + HttpResponse( + status=429, + body='{"message": "slow down"}', + headers={"retry-after": "2"}, + ), + HttpResponse(status=200, body='{"items": []}'), + ] + ) + sleeps: list[float] = [] + client = LDApiClient( + api_token="api-token", + transport=transport, + max_retries=1, + sleep=sleeps.append, + random_value=lambda: 0.0, + ) + + assert client.get("projects/proj/datasets") == {"items": []} + assert len(transport.requests) == 2 + assert sleeps == [2.0] + + +def test_get_retries_network_error() -> None: + attempts = 0 + + def transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + nonlocal attempts + del method, url, headers, body, timeout + attempts += 1 + if attempts == 1: + raise TimeoutError("timed out") + return HttpResponse(status=200, body='{"items": []}') + + sleeps: list[float] = [] + client = LDApiClient( + api_token="api-token", + transport=transport, + max_retries=1, + sleep=sleeps.append, + random_value=lambda: 0.0, + ) + + assert client.get("projects/proj/datasets") == {"items": []} + assert attempts == 2 + assert sleeps == [0.5] + + +@pytest.mark.parametrize("status", [429, 500]) +def test_post_does_not_retry_transient_status(status: int) -> None: + transport = RecordingTransport( + [HttpResponse(status=status, body='{"message": "transient"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport, max_retries=3) + + with pytest.raises(LDApiError) as excinfo: + client.post("projects/proj/evaluations", body={"name": "evaluation"}) + + assert excinfo.value.status == status + assert len(transport.requests) == 1 + + +def test_post_does_not_retry_network_error() -> None: + attempts = 0 + + def transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + nonlocal attempts + del method, url, headers, body, timeout + attempts += 1 + raise TimeoutError("timed out after creation") + + client = LDApiClient(api_token="api-token", transport=transport, max_retries=3) + + with pytest.raises(EvaluationsError, match="timed out after creation"): + client.post("projects/proj/evaluations", body={"name": "evaluation"}) + + assert attempts == 1 + + +def test_forbidden_response_is_not_retried() -> None: + transport = RecordingTransport( + [HttpResponse(status=403, body='{"message": "forbidden"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport, max_retries=3) + + with pytest.raises(LDApiError) as excinfo: + client.get("projects/proj/evaluations") + + assert excinfo.value.status == 403 + assert len(transport.requests) == 1 + + +def test_error_response_raises_ld_api_error() -> None: + transport = RecordingTransport( + [HttpResponse(status=404, body='{"message": "nope"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport) + + with pytest.raises(LDApiError) as excinfo: + client.get("projects/proj/ai-tools/missing") + + assert excinfo.value.status == 404 + assert excinfo.value.path == "projects/proj/ai-tools/missing" + + +def test_empty_response_body_is_none() -> None: + transport = RecordingTransport([HttpResponse(status=204, body="")]) + client = LDApiClient(api_token="api-token", transport=transport) + + assert client.post("projects/proj/evaluations/support-qa/runs") is None + + +def test_usage_matches_ingest_wire_shape() -> None: + usage = Usage(input_tokens=812, output_tokens=96) + + assert usage.to_wire() == {"input_tokens": 812, "output_tokens": 96} + assert Usage.from_wire({"input_tokens": 1, "output_tokens": 2}) == Usage(1, 2) + assert Usage.from_wire({}) == Usage(0, 0) + + +def test_run_summary_and_result() -> None: + summary = RunSummary.from_wire( + { + "total_rows": 500, + "passed_rows": 497, + "failed_rows": 1, + "error_rows": 1, + "pending_rows": 1, + } + ) + result = EvalRunResult( + passed=False, + url="https://app.launchdarkly.com/run", + run_id="run-1", + summary=summary, + ) + + assert summary.total_rows == 500 + assert summary.error_rows == 1 + assert summary.pending_rows == 1 + assert RunSummary.from_wire({"pending": 2}).pending_rows == 2 + assert RunSummary.from_wire(None) == RunSummary() + assert result.passed is False + assert result.run_id == "run-1"