openobserve/tests/api-testing/support/factories.py

"""Payload factory functions for API requests.

Each factory returns a dict that can be passed to `client.<resource>.create(...)`.
Sensible defaults are merged with caller-supplied overrides via `**kwargs`.

Pattern: small functions, named per the API resource. No classes — these are
just typed dict builders.
"""
from __future__ import annotations

import uuid
from datetime import datetime, timedelta, UTC
from typing import Any


def _ms_epoch(dt: datetime) -> int:
    """OpenObserve time fields are microseconds-since-epoch as int."""
    return int(dt.timestamp() * 1_000_000)


def time_window(*, minutes: int = 15) -> tuple[int, int]:
    """Return (start, end) microsecond-epoch ints for the last `minutes` minutes."""
    now = datetime.now(UTC)
    end = _ms_epoch(now)
    start = _ms_epoch(now - timedelta(minutes=minutes))
    return start, end


def unique_name(prefix: str = "pytest") -> str:
    """Generate a unique, lowercase, alphanumeric resource name.

    Used to prevent cross-test resource collisions. The original suite had
    fixed names like `pytests@gmail.com` that caused cascading failures and
    blocked parallel runs.
    """
    return f"{prefix}_{uuid.uuid4().hex[:8]}"


def unique_email(prefix: str = "pytest") -> str:
    """Generate a unique test email (using .invalid TLD per RFC 2606)."""
    return f"{prefix}_{uuid.uuid4().hex[:8]}@test.invalid"


# ----- payload factories -----


def search_payload(
    sql: str,
    *,
    start_time: int | None = None,
    end_time: int | None = None,
    size: int = 100,
    from_: int = 0,
    quick_mode: bool = False,
    track_total_hits: bool = False,
) -> dict[str, Any]:
    """SQL search request body for POST /api/{org}/_search.

    Note: `track_total_hits` defaults to **False**. When True, OO switches
    to histogram-aggregate mode and returns `{"zo_sql_num": <count>}` instead
    of raw hits, which surprises callers expecting individual records.
    Callers that genuinely want the aggregate count can pass True explicitly.
    """
    if start_time is None or end_time is None:
        s, e = time_window(minutes=15)
        start_time = start_time if start_time is not None else s
        end_time = end_time if end_time is not None else e
    return {
        "query": {
            "sql": sql,
            "start_time": start_time,
            "end_time": end_time,
            "from": from_,
            "size": size,
            "quick_mode": quick_mode,
            "track_total_hits": track_total_hits,
        }
    }


def user_payload(
    *,
    email: str | None = None,
    password: str = "Complexpass#123",
    first_name: str = "Pytest",
    last_name: str = "User",
    role: str = "admin",
    organization: str = "default",
) -> dict[str, Any]:
    """User create/update body."""
    return {
        "email": email or unique_email(),
        "password": password,
        "first_name": first_name,
        "last_name": last_name,
        "role": role,
        "organization": organization,
    }


def dashboard_payload(
    *,
    title: str | None = None,
    description: str = "",
    folder_id: str = "default",
    version: int = 8,
    tabs: list | None = None,
) -> dict[str, Any]:
    """Dashboard create body."""
    return {
        "version": version,
        "title": title or unique_name("dashboard"),
        "description": description,
        "folder_id": folder_id,
        "tabs": tabs if tabs is not None else [],
    }


def axis_item(alias: str, *, label: str | None = None, column: str | None = None) -> dict[str, Any]:
    """One entry for a panel's `x` / `y` / `breakdown` array.

    `label` is a required String in the v8 schema (not optional) — omitting it
    is a 422. For `customQuery` panels the `alias` must match the SELECT alias
    verbatim, or the axis renders blank.
    """
    return {
        "label": label if label is not None else alias,
        "alias": alias,
        "column": column if column is not None else alias,
        "color": None,
    }


def panel_payload(
    *,
    stream: str,
    sql: str,
    x_alias: str,
    y_alias: str,
    panel_id: str | None = None,
    title: str | None = None,
    description: str = "",
    panel_type: str = "bar",
    stream_type: str = "logs",
    layout: dict[str, int] | None = None,
) -> dict[str, Any]:
    """A v8 `Panel` carrying hand-written SQL, for the AddPanel/UpdatePanel APIs.

    `customQuery` is hard-coded True: with False, OO discards `sql` and rebuilds
    the query from the field config, which silently defeats any test asserting on
    the SQL it supplied.

    Note `config` (PanelConfig) is snake_case while `fields`/`queries` are
    camelCase — the v8 schema genuinely mixes both.
    """
    return {
        "id": panel_id or unique_name("panel"),
        "type": panel_type,
        "title": title or unique_name("Panel"),
        "description": description,
        "config": {
            "show_legends": True,
            "legends_position": None,
        },
        "queryType": "sql",
        "queries": [
            {
                "query": sql,
                "vrlFunctionQuery": None,
                "customQuery": True,
                "fields": {
                    "stream": stream,
                    "stream_type": stream_type,
                    "x": [axis_item(x_alias)],
                    "y": [axis_item(y_alias)],
                    "z": [],
                    # A bare [] is rejected: PanelFilter is an untagged enum and
                    # the empty list matches neither variant. It must be a group.
                    "filter": {
                        "filterType": "group",
                        "logicalOperator": "AND",
                        "conditions": [],
                    },
                },
                "config": {"promql_legend": ""},
            }
        ],
        "layout": layout if layout is not None else {"x": 0, "y": 0, "w": 24, "h": 9, "i": 1},
    }


def alert_template_payload(
    *,
    name: str | None = None,
    body: str | None = None,
    type_: str = "http",
    title: str = "",
) -> dict[str, Any]:
    """Alert template create body."""
    return {
        "name": name or unique_name("template"),
        "body": body or '{"text": "alert {alert_name} on {stream_name} active"}',
        "type": type_,
        "title": title,
    }


def webhook_destination_payload(
    *,
    name: str | None = None,
    url: str = "http://localhost:0/sink",  # test default — override per test
    method: str = "post",
    template: str,
    headers: dict | None = None,
    skip_tls_verify: bool = False,
) -> dict[str, Any]:
    """Webhook destination create body."""
    return {
        "name": name or unique_name("dest"),
        "url": url,
        "method": method,
        "template": template,
        "headers": headers or {},
        "skip_tls_verify": skip_tls_verify,
    }


def folder_payload(*, name: str | None = None, description: str = "") -> dict[str, Any]:
    """Folder create body (alerts or dashboards)."""
    return {
        "name": name or unique_name("folder"),
        "description": description,
    }