openobserve/tests/api-testing/tests/ingest/test_logs.py

"""Log Ingestion API Tests.

Rewritten in Phase 4 of the api-tests revamp:
- Uses unique stream names per test via `temp_stream_name` fixture (no more
  fixed `newpy_tests` that created hidden ordering dependencies).
- Replaces hardcoded Nov-2023 epoch timestamps in the VRL test with
  current-time `time_window()` (the old timestamps had silently gone past
  retention, returning 0 hits while still asserting status==200).
- Body validation: ingest response checks `status` field; stream list checks
  for the stream's presence; schema response checks for known fields.
- Parametrize the two invalid-stream-name cases.
"""
from __future__ import annotations

import logging

from support.client import OpenObserveClient
from support.factories import time_window

logger = logging.getLogger(__name__)

ORG_ID = "default"

# Athletics records used as test payload — small, deterministic.
SAMPLE_LOG_RECORDS = [
    {
        "Athlete": "newtemp",
        "City": "Athens",
        "Country": "HUN",
        "Discipline": "Swimming",
        "Sport": "Aquatics",
        "Year": 1896,
    },
    {
        "Athlete": "HERSCHMANN",
        "City": "Athens",
        "Country": "CHN",
        "Discipline": "Swimming",
        "Sport": "Aquatics",
        "Year": 1896,
    },
]


# ----- ingest happy path -----


def test_ingest_logs_returns_200_and_records_count(
    client: OpenObserveClient, temp_stream_name: str
):
    """POST /api/{org}/{stream}/_json returns 200 + a body listing what was ingested."""
    resp = client.streams.ingest_json(temp_stream_name, SAMPLE_LOG_RECORDS)
    assert resp.status_code == 200, resp.text

    body = resp.json()
    assert "status" in body, f"ingest response should have 'status' key: {body}"
    # OO returns a per-record status array — verify count matches what we sent
    statuses = body.get("status") or []
    assert len(statuses) >= 1, f"expected at least 1 status entry, got {statuses}"


# ----- ingest error paths -----


def test_ingest_with_empty_stream_name_returns_400(
    client: OpenObserveClient,
):
    """POST to /api/{org}//_json (empty stream segment) → 400.

    Constructed manually because the client's urljoin would collapse the
    double slash; we specifically want to test what OO does when the stream
    name segment is present-but-empty.
    """
    url = f"{client.base_url}api/{ORG_ID}//_json"
    resp = client.session.post(url, json=SAMPLE_LOG_RECORDS)
    assert resp.status_code == 400, \
        f"empty stream name should be 400, got {resp.status_code}: {resp.text}"


def test_ingest_with_invalid_stream_name_returns_404(
    client: OpenObserveClient,
):
    """POST with disallowed characters in stream name → 404."""
    resp = client.post("newpy#$#$#$#$2@2/_json", json=SAMPLE_LOG_RECORDS)
    assert resp.status_code == 404, \
        f"invalid stream name should be 404, got {resp.status_code}: {resp.text}"


# ----- search/VRL -----


def test_vrl_query_against_ingested_stream(
    client: OpenObserveClient, temp_stream_name: str
):
    """A VRL-augmented query against a freshly-ingested stream returns 200 + hits.

    Replaces the legacy test that used hardcoded Nov-2023 epochs (which silently
    returned 0 hits once past retention but still asserted status==200 — the
    test would have masked complete breakage of VRL).
    """
    # Ingest first so there's data to query
    ing = client.streams.ingest_json(temp_stream_name, SAMPLE_LOG_RECORDS)
    assert ing.status_code == 200, ing.text

    # Use a real, current time window
    start, end = time_window(minutes=30)
    payload = {
        "query": {
            "sql": f'select * from "{temp_stream_name}"',
            "start_time": start,
            "end_time": end,
            "from": 0,
            "size": 150,
            # base64 of `.a = 1\n .` — adds a constant field to each result
            "query_fn": "LmEgPTEgCiAu",
        }
    }
    resp = client.post("_search?type=logs", json=payload)
    assert resp.status_code == 200, f"VRL query failed: {resp.status_code} {resp.text}"

    body = resp.json()
    assert "hits" in body, f"VRL response should have 'hits' key: {body}"
    assert isinstance(body["hits"], list), "'hits' should be a list"
    # Note: hits may be empty if ingest hasn't finished indexing — that's a
    # separate concern. We're asserting the query CONTRACT here, not data
    # latency (which is covered by the conftest ingest_data fixture for the
    # session-wide test stream).


# ----- streams metadata -----


def test_list_streams_returns_list_with_known_baseline(
    client: OpenObserveClient,
):
    """GET /api/{org}/streams returns a list including the session-wide stream_pytest_data."""
    streams = client.streams.list()
    assert isinstance(streams, list)
    names = [s.get("name") for s in streams]
    # stream_pytest_data is ingested by the autouse conftest fixture
    assert "stream_pytest_data" in names, \
        f"baseline stream missing from list: {names}"


def test_get_schema_after_ingest(
    client: OpenObserveClient, temp_stream_name: str
):
    """After ingesting to a new stream, GET /streams/{name}/schema returns the field schema."""
    client.streams.ingest_json(temp_stream_name, SAMPLE_LOG_RECORDS)
    schema = client.streams.schema(temp_stream_name)

    assert isinstance(schema, dict), f"schema should be a dict, got {type(schema).__name__}"
    # OO schema responses include a 'schema' or 'fields' key
    assert ("schema" in schema or "fields" in schema), \
        f"schema response should have 'schema' or 'fields' key: {list(schema.keys())}"


# ----- delete -----


def test_delete_nonexistent_stream_returns_404(
    client: OpenObserveClient,
):
    """DELETE /streams/{name} where name doesn't exist → 404."""
    resp = client.streams.delete("definitely-does-not-exist-12345")
    assert resp.status_code == 404, \
        f"deleting nonexistent stream should be 404, got {resp.status_code}: {resp.text}"


# ----- settings error path -----


def test_put_stream_settings_without_body_returns_415(
    client: OpenObserveClient, temp_stream_name: str
):
    """PUT /streams/{name}/settings without Content-Type body returns 415."""
    # Ingest first to ensure the stream exists
    client.streams.ingest_json(temp_stream_name, SAMPLE_LOG_RECORDS)

    resp = client.put(f"streams/{temp_stream_name}/settings")
    assert resp.status_code == 415, \
        f"settings PUT without body should be 415, got {resp.status_code}: {resp.text}"