test: regression coverage for 29 closed Needs-Automation issues (#14883)

This commit is contained in:
Neha P
2026-09-25 12:42:57 +00:00
committed by GitHub
parent c7036d47f7
commit 0cbdb03dc2
29 changed files with 1982 additions and 12 deletions
@@ -0,0 +1,78 @@
"""Concurrent dashboard create in a fresh org must not 500 [regression #14163].
In a brand-new org the `default` dashboards folder does not exist yet. Several
concurrent `POST /dashboards` all passed the folder-existence check and then
raced to create it; the losers' INSERT violated the folder unique index and
surfaced as a raw HTTP 500. Retrying worked, because by then the folder existed.
PR #14186 made the creation idempotent -- `folders::get_or_create` attempts the
insert and only reports failure if the row is still absent afterwards.
This is a check-then-act race, so it only shows up when the requests genuinely
overlap; issuing them sequentially would pass on the unfixed build. The requests
are therefore fired from a thread pool against a freshly created org.
Storage contention is NOT the defect. On SQLite a concurrent burst can answer
"database is locked", which is a 5xx that says nothing about folder creation, so
those are counted and tolerated separately -- the defect's own signature is a
unique-constraint failure, which does not mention a lock. Treating every 5xx as
the bug made this test fail on CI even against a fixed build.
"""
import logging
import os
import uuid
from concurrent.futures import ThreadPoolExecutor
import pytest
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
CONCURRENCY = 6
LOCK_MARKERS = ("database is locked", "is locked", "locked")
def _is_storage_contention(resp) -> bool:
body = (resp.text or "").lower()
return any(m in body for m in LOCK_MARKERS)
@pytest.fixture(scope="module")
def fresh_org(create_session, base_url):
"""A brand-new org, so the default dashboards folder is guaranteed absent.
Module-scoped deliberately: `DELETE /api/organizations/{id}` answers 404 on
this build, so an org cannot be cleaned up and a function-scoped fixture
would leak one per test. One org per module bounds that to a single row.
"""
session = create_session
org = f"e2e_dash_race_{uuid.uuid4().hex[:8]}"
resp = session.post(f"{base_url}api/organizations", json={"name": org})
if resp.status_code not in (200, 201):
pytest.skip(f"could not create a fresh org ({resp.status_code}): {resp.text[:200]}")
yield resp.json().get("identifier", org)
def test_concurrent_dashboard_create_does_not_500(create_session, base_url, fresh_org):
session = create_session
def create(i: int):
return session.post(
f"{base_url}api/{fresh_org}/dashboards",
json={"title": f"race_{i}", "description": "concurrent create race", "panels": []},
)
with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
responses = list(pool.map(create, range(CONCURRENCY)))
codes = [r.status_code for r in responses]
contention = [r for r in responses if r.status_code >= 500 and _is_storage_contention(r)]
defects = [r for r in responses if r.status_code >= 500 and not _is_storage_contention(r)]
logger.info("status codes: %s (storage contention: %d)", codes, len(contention))
assert not defects, (
"concurrent dashboard creates raced default-folder creation and returned "
f"{[r.status_code for r in defects]}; first body: {defects[0].text[:300]}"
)
assert any(c in (200, 201) for c in codes), \
f"no concurrent create succeeded at all, got {codes}"
@@ -0,0 +1,87 @@
"""OTLP log severity must be consistent [regression #14639].
`severity` was stored as `severityText` when the record set it and as the raw
`severityNumber` otherwise, so the same level landed as `"ERROR"` from one
exporter and `17` from another. The column type then depended on which record
arrived first, and the log viewer read a numeric severity as a syslog level.
PR #14791 derives the text from the number and keeps the number alongside.
Both records carry severityNumber 17, differing only in whether severityText is
set, so any divergence between them is the defect.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
SEVERITY_NUMBER = 17 # OTLP ERROR
def test_severity_is_derived_when_only_the_number_is_sent(create_session, base_url):
session = create_session
tag_with = f"withtext_{uuid.uuid4().hex[:6]}"
tag_without = f"numonly_{uuid.uuid4().hex[:6]}"
t_ns = int(time.time() * 1e9) - 180 * 10**9
def record(offset_ns, tag, with_text):
rec = {
"timeUnixNano": str(t_ns + offset_ns),
"severityNumber": SEVERITY_NUMBER,
"body": {"stringValue": f"probe {tag}"},
"attributes": [{"key": "probe_tag", "value": {"stringValue": tag}}],
}
if with_text:
rec["severityText"] = "ERROR"
return rec
payload = {"resourceLogs": [{
"resource": {"attributes": [{"key": "service.name",
"value": {"stringValue": "pytest_severity"}}]},
"scopeLogs": [{"scope": {"name": "pytest"}, "logRecords": [
record(0, tag_with, True),
record(10**6, tag_without, False),
]}],
}]}
r = session.post(f"{base_url}api/{ORG_ID}/v1/logs", json=payload)
assert r.status_code == 200, f"OTLP log ingest failed: {r.status_code} {r.text[:300]}"
now_us = int(time.time() * 1_000_000)
rows = []
for _ in range(30):
q = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {
"sql": 'SELECT probe_tag, severity, severity_number FROM "default" '
f"WHERE probe_tag IN ('{tag_with}','{tag_without}')",
"start_time": now_us - 600_000_000, "end_time": now_us + 60_000_000,
"from": 0, "size": 10}},
)
if q.status_code == 200 and len(q.json().get("hits", [])) >= 2:
rows = q.json()["hits"]
break
time.sleep(2)
assert len(rows) >= 2, f"both log records never became searchable, got {rows}"
by_tag = {r_["probe_tag"]: r_ for r_ in rows}
logger.info("severity by tag: %s",
{k: (v.get("severity"), v.get("severity_number")) for k, v in by_tag.items()})
sev_with = by_tag[tag_with].get("severity")
sev_without = by_tag[tag_without].get("severity")
# The defect stored the raw number here, giving two representations of one level.
assert sev_without == sev_with, \
(f"severity must not depend on whether severityText was sent: "
f"with-text={sev_with!r} number-only={sev_without!r}")
assert str(sev_without).upper() == "ERROR", \
f"severityNumber {SEVERITY_NUMBER} should derive ERROR, got {sev_without!r}"
# The number must survive alongside the derived text.
for tag, row in by_tag.items():
assert row.get("severity_number") == SEVERITY_NUMBER, \
f"{tag} lost its severity_number: {row.get('severity_number')!r}"
@@ -0,0 +1,79 @@
"""Resource attributes must survive span flattening [regression #12448].
When a span attribute used the same key as a resource-derived field, flattening
let the span value overwrite it -- so a span carrying its own `service.name`
replaced the resource's, and the trace was attributed to the wrong service.
PR #12456 keeps the resource value canonical and moves the colliding span
attribute aside.
The control span (no collision) is what makes this meaningful: without it a
build that simply ignored span attributes entirely would also pass.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
RESOURCE_SERVICE = "pytest_resource_svc"
SPAN_SERVICE = "pytest_span_override"
def test_span_attribute_does_not_overwrite_resource_service_name(create_session, base_url):
session = create_session
now_ns = int(time.time() * 1e9) - 120 * 10**9
colliding_id = uuid.uuid4().hex
control_id = uuid.uuid4().hex
def span(trace_id, name, attrs):
return {
"traceId": trace_id, "spanId": uuid.uuid4().hex[:16], "parentSpanId": "",
"flags": 1, "name": name, "kind": 2,
"startTimeUnixNano": str(now_ns), "endTimeUnixNano": str(now_ns + 5 * 10**6),
"attributes": attrs, "status": {"code": 0},
}
payload = {"resourceSpans": [{
"resource": {"attributes": [
{"key": "service.name", "value": {"stringValue": RESOURCE_SERVICE}}]},
"scopeSpans": [{"scope": {"name": "pytest"}, "spans": [
# Collides with the resource-derived service_name.
span(colliding_id, "colliding",
[{"key": "service.name", "value": {"stringValue": SPAN_SERVICE}}]),
span(control_id, "control",
[{"key": "http.route", "value": {"stringValue": "/control"}}]),
]}],
}]}
r = session.post(f"{base_url}api/{ORG_ID}/v1/traces", json=payload)
assert r.status_code == 200, f"trace ingest failed: {r.status_code} {r.text[:300]}"
start_us, end_us = int(now_ns / 1000) - 60_000_000, int(time.time() * 1_000_000)
rows = []
for _ in range(40):
q = session.post(
f"{base_url}api/{ORG_ID}/_search?type=traces",
json={"query": {
"sql": 'SELECT trace_id, service_name, operation_name FROM "default" '
f"WHERE trace_id IN ('{colliding_id}','{control_id}')",
"start_time": start_us, "end_time": end_us, "from": 0, "size": 20}},
)
if q.status_code == 200 and len(q.json().get("hits", [])) >= 2:
rows = q.json()["hits"]
break
time.sleep(2)
assert len(rows) >= 2, f"seeded spans never became searchable, got {rows}"
by_trace = {r_["trace_id"]: r_.get("service_name") for r_ in rows}
logger.info("service_name by trace: %s", by_trace)
assert by_trace.get(control_id) == RESOURCE_SERVICE, \
f"control span should carry the resource service, got {by_trace.get(control_id)}"
# The defect let the span attribute win here.
assert by_trace.get(colliding_id) == RESOURCE_SERVICE, \
(f"a span attribute must not overwrite the resource service_name: "
f"got {by_trace.get(colliding_id)!r}, expected {RESOURCE_SERVICE!r}")
@@ -0,0 +1,141 @@
"""OTLP data-point flags and exponential histograms [#14637, #14634].
#14637 (PR #14727): bit 0 of `flags` is NO_RECORDED_VALUE -- the point marks a
gap, e.g. a Prometheus staleness marker translated by the Collector. It was
turned into a label and the point's value stored as if it were real data, so
gaps became readings.
#14634 (PR #14724): exponential histograms were flattened into classic buckets
incorrectly -- each bucket carried its own count rather than the cumulative
count, `zero_count` was never written so zero-bucket observations were lost, and
there was no `le="inf"` bucket, which makes `histogram_quantile` return NaN.
All three are asserted separately because each fails on its own.
"""
import time
import uuid
import pytest
from support.wait import wait_until
@pytest.fixture(scope="module")
def flagged_metric(create_session, base_url, org_id):
"""One point flagged NO_RECORDED_VALUE, one clean point a second later."""
session = create_session
metric = f"pytest_flags_{uuid.uuid4().hex[:6]}"
t_ns = int(time.time() * 1e9) - 180 * 10**9
payload = {"resourceMetrics": [{
"resource": {"attributes": [{"key": "service.name",
"value": {"stringValue": "pytest_flags"}}]},
"scopeMetrics": [{"scope": {"name": "pytest"}, "metrics": [{
"name": metric,
"gauge": {"dataPoints": [
{"asDouble": 99.0, "timeUnixNano": str(t_ns), "flags": 1},
{"asDouble": 7.0, "timeUnixNano": str(t_ns + 10**9), "flags": 0},
]},
}]}],
}]}
r = session.post(f"{base_url}api/{org_id}/v1/metrics", json=payload)
assert r.status_code == 200, f"OTLP ingest failed: {r.status_code} {r.text[:300]}"
at = int(t_ns / 1e9) + 60
def visible():
q = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
params={"query": metric, "time": str(at)})
return q.status_code == 200 and bool(q.json().get("data", {}).get("result"))
wait_until(visible, timeout=120, interval=2, msg=f"{metric} never became queryable")
yield metric, at
session.delete(f"{base_url}api/{org_id}/streams/{metric}?type=metrics")
@pytest.fixture(scope="module")
def exp_histogram(create_session, base_url, org_id):
"""An exponential histogram with a non-empty zero bucket."""
session = create_session
metric = f"pytest_exphist_{uuid.uuid4().hex[:6]}"
t_ns = int(time.time() * 1e9) - 180 * 10**9
zero_count, positive = 3, ["2", "3", "2"] # total = 3 + 7 = 10
payload = {"resourceMetrics": [{
"resource": {"attributes": [{"key": "service.name",
"value": {"stringValue": "pytest_exphist"}}]},
"scopeMetrics": [{"scope": {"name": "pytest"}, "metrics": [{
"name": metric,
"exponentialHistogram": {"aggregationTemporality": 2, "dataPoints": [{
"startTimeUnixNano": str(t_ns), "timeUnixNano": str(t_ns),
"count": "10", "sum": 55.0, "scale": 0,
"zeroCount": str(zero_count),
"positive": {"offset": 0, "bucketCounts": positive},
}]},
}]}],
}]}
r = session.post(f"{base_url}api/{org_id}/v1/metrics", json=payload)
assert r.status_code == 200, f"OTLP ingest failed: {r.status_code} {r.text[:300]}"
at = int(t_ns / 1e9) + 60
def visible():
q = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
params={"query": f"{metric}_bucket", "time": str(at)})
return q.status_code == 200 and bool(q.json().get("data", {}).get("result"))
wait_until(visible, timeout=120, interval=2, msg=f"{metric}_bucket never appeared")
yield metric, at, zero_count, 10
for suffix in ("", "_count", "_sum", "_bucket"):
session.delete(f"{base_url}api/{org_id}/streams/{metric}{suffix}?type=metrics")
def _buckets(session, base_url, org_id, metric, at):
q = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
params={"query": f"{metric}_bucket", "time": str(at)})
assert q.status_code == 200, f"bucket query failed: {q.status_code} {q.text[:300]}"
out = {}
for s in q.json()["data"]["result"]:
out[s["metric"].get("le")] = float(s["value"][1])
return out
def test_no_recorded_value_point_is_dropped(create_session, base_url, org_id, flagged_metric):
"""The flagged point marks a gap and must not be stored as a reading."""
metric, at = flagged_metric
q = create_session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query_range",
params={"query": metric, "start": str(at - 600),
"end": str(at), "step": "30"})
assert q.status_code == 200, f"range query failed: {q.status_code} {q.text[:300]}"
values = {v[1] for s in q.json()["data"]["result"] for v in s.get("values", [])}
assert "7" in values or 7.0 in {float(v) for v in values}, \
f"the clean point must survive, got {values}"
assert "99" not in values and 99.0 not in {float(v) for v in values}, \
f"a NO_RECORDED_VALUE point was stored as real data: {values}"
def test_exponential_histogram_buckets_are_cumulative(create_session, base_url, org_id, exp_histogram):
metric, at, _zero, total = exp_histogram
buckets = _buckets(create_session, base_url, org_id, metric, at)
finite = sorted(((float(le), v) for le, v in buckets.items() if le not in (None, "inf", "+Inf")),
key=lambda p: p[0])
counts = [v for _le, v in finite]
assert counts == sorted(counts), \
f"bucket counts must be cumulative (non-decreasing), got {finite}"
assert counts[-1] == total, \
f"the widest finite bucket must reach the total {total}, got {counts[-1]}"
def test_exponential_histogram_has_zero_and_inf_buckets(create_session, base_url, org_id, exp_histogram):
metric, at, zero_count, total = exp_histogram
buckets = _buckets(create_session, base_url, org_id, metric, at)
inf = next((v for le, v in buckets.items() if le in ("inf", "+Inf")), None)
assert inf is not None, f"an le=inf bucket must exist, got {sorted(buckets)}"
assert inf == total, f"the inf bucket must carry the total {total}, got {inf}"
# Zero-bucket observations were silently lost before the fix.
assert buckets.get("0") == zero_count, \
f"the zero bucket must carry zeroCount {zero_count}, got {buckets.get('0')}"
@@ -0,0 +1,87 @@
"""OTLP histogram optional stats must be skipped when absent [regression #13019].
OTLP's `HistogramDataPoint` defines `sum`, `min` and `max` as optional. The
ingestion path turned a missing optional into JSON `null` and still emitted the
`_sum` / `_min` / `_max` record, so those metrics existed with no real value.
PR #13082 skips the record entirely when the stat is absent, while `_count` and
the bucket records are always emitted.
Asserting only that `_count` exists would pass on the unfixed build, so the
contract here is the ABSENCE of `_sum` / `_min` / `_max` for a point that
carries none -- proven against a second histogram in the same payload that does
carry them, so a build emitting nothing at all cannot pass either.
"""
import time
import uuid
import pytest
from support.wait import wait_until
@pytest.fixture(scope="module")
def histogram_names(create_session, base_url, org_id):
"""Two histograms: one without sum/min/max, one with, ingested together."""
session = create_session
bare = f"pytest_hist_bare_{uuid.uuid4().hex[:6]}"
full = f"pytest_hist_full_{uuid.uuid4().hex[:6]}"
now_ns = int(time.time() * 1e9) - 120 * 10**9
def point(with_stats):
p = {
"startTimeUnixNano": str(now_ns),
"timeUnixNano": str(now_ns),
"count": "6",
"bucketCounts": ["2", "3", "1"],
"explicitBounds": [1.0, 5.0],
}
if with_stats:
p.update({"sum": 21.0, "min": 0.5, "max": 9.0})
return p
payload = {"resourceMetrics": [{
"resource": {"attributes": [
{"key": "service.name", "value": {"stringValue": "pytest_hist"}}]},
"scopeMetrics": [{"scope": {"name": "pytest"}, "metrics": [
{"name": bare, "unit": "1",
"histogram": {"dataPoints": [point(False)], "aggregationTemporality": 2}},
{"name": full, "unit": "1",
"histogram": {"dataPoints": [point(True)], "aggregationTemporality": 2}},
]}],
}]}
resp = session.post(f"{base_url}api/{org_id}/v1/metrics", json=payload)
assert resp.status_code == 200, f"OTLP metrics ingest failed: {resp.status_code} {resp.text[:400]}"
def names():
r = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/label/__name__/values")
return set(r.json().get("data", [])) if r.status_code == 200 else set()
wait_until(lambda: f"{full}_sum" in names(), timeout=120, interval=2,
msg="the control histogram's _sum never appeared")
yield bare, full, names()
for base in (bare, full):
for suffix in ("", "_count", "_sum", "_min", "_max", "_bucket"):
session.delete(f"{base_url}api/{org_id}/streams/{base}{suffix}?type=metrics")
def test_absent_optional_stats_emit_no_record(create_session, base_url, org_id, histogram_names):
bare, _full, present = histogram_names
for suffix in ("sum", "min", "max"):
assert f"{bare}_{suffix}" not in present, \
f"{bare}_{suffix} must not exist for a point that carries no {suffix}"
def test_count_and_buckets_survive(create_session, base_url, org_id, histogram_names):
bare, _full, present = histogram_names
assert f"{bare}_count" in present, f"{bare}_count must still be emitted"
assert f"{bare}_bucket" in present, f"{bare}_bucket must still be emitted"
def test_present_optional_stats_are_still_emitted(create_session, base_url, org_id, histogram_names):
"""Control: a point that DOES carry the stats keeps its _sum/_min/_max."""
_bare, full, present = histogram_names
for suffix in ("sum", "min", "max"):
assert f"{full}_{suffix}" in present, \
f"{full}_{suffix} must exist for a point that carries {suffix}"
@@ -0,0 +1,127 @@
"""Prometheus label metadata endpoints [regression #14635].
Two separate defects, fixed in two PRs, and the shipped behaviour for the second
is NOT what the issue asked for:
- #14729 made `/labels` and `/label/__name__/values` honour `start`/`end`.
They previously returned an empty list whenever a time range was supplied,
which is exactly what Grafana's metric browser sends, so it showed nothing.
- #14814 made `/label/{name}/values` REJECT a request that does not identify a
single metric, with HTTP 400 and a Prometheus `bad_data` error. The issue
asked for these to work without `match[]` the way Prometheus does; the fix
deliberately chose an explicit error over a cross-stream query, because
querying across all metric streams is unsupported. A successful empty list
was the bug: it hid that nothing had been queried.
So the asserted contract here is the error, not the values. `/labels` and
`/label/__name__/values` keep working without `match[]`.
"""
import time
import uuid
import pytest
from support.wait import wait_until
LABEL_NAME = "job"
JOB_VALUE = "pytest_label_metadata"
@pytest.fixture(scope="module")
def labelled_metric(create_session, base_url, org_id):
"""Ingest one gauge carrying a non-__name__ label, and wait for it to be queryable."""
session = create_session
metric = f"pytest_labels_{uuid.uuid4().hex[:8]}"
now_sec = int(time.time())
payload = [
{
"__name__": metric,
"__type__": "gauge",
LABEL_NAME: JOB_VALUE,
"_timestamp": (now_sec - 60 + i) * 1000,
"value": float(i),
}
for i in range(5)
]
resp = session.post(f"{base_url}api/{org_id}/ingest/metrics/_json", json=payload)
assert resp.status_code == 200, f"metrics ingest failed: {resp.status_code} {resp.text[:400]}"
def visible():
r = session.get(
f"{base_url}api/{org_id}/prometheus/api/v1/label/__name__/values",
)
return r.status_code == 200 and metric in r.json().get("data", [])
wait_until(visible, timeout=90, interval=1,
msg=f"metric {metric} never appeared in /label/__name__/values")
yield metric, now_sec
session.delete(f"{base_url}api/{org_id}/streams/{metric}?type=metrics")
def test_labels_honours_a_time_range(create_session, base_url, org_id, labelled_metric):
"""`/labels` must return names with start/end, not the empty list it used to."""
session = create_session
_, now_sec = labelled_metric
resp = session.get(
f"{base_url}api/{org_id}/prometheus/api/v1/labels",
params={"start": str(now_sec - 3600), "end": str(now_sec + 60)},
)
assert resp.status_code == 200, f"/labels failed: {resp.status_code} {resp.text[:400]}"
names = resp.json().get("data", [])
assert names, "/labels returned nothing for a range that contains ingested data"
assert LABEL_NAME in names, f"expected {LABEL_NAME!r} among label names, got {names}"
def test_name_values_honours_a_time_range(create_session, base_url, org_id, labelled_metric):
"""`/label/__name__/values` must return metric names with start/end."""
session = create_session
metric, now_sec = labelled_metric
resp = session.get(
f"{base_url}api/{org_id}/prometheus/api/v1/label/__name__/values",
params={"start": str(now_sec - 3600), "end": str(now_sec + 60)},
)
assert resp.status_code == 200, \
f"/label/__name__/values failed: {resp.status_code} {resp.text[:400]}"
values = resp.json().get("data", [])
assert metric in values, \
f"{metric} missing from /label/__name__/values over a range containing it, got {values[:20]}"
def test_label_values_without_a_metric_is_rejected(create_session, base_url, org_id, labelled_metric):
"""An unscoped label-values request must error, not answer an empty success."""
session = create_session
_, now_sec = labelled_metric
resp = session.get(
f"{base_url}api/{org_id}/prometheus/api/v1/label/{LABEL_NAME}/values",
params={"start": str(now_sec - 3600), "end": str(now_sec + 60)},
)
assert resp.status_code == 400, \
f"unscoped label values should be rejected with 400, got {resp.status_code}: {resp.text[:400]}"
body = resp.json()
assert body.get("status") == "error", f"expected a Prometheus error envelope, got {body}"
assert "match[]" in body.get("error", ""), \
f"the error should tell the caller match[] is required, got {body.get('error')!r}"
def test_label_values_with_a_metric_returns_values(create_session, base_url, org_id, labelled_metric):
"""Scoped by match[], the same request returns the label's values."""
session = create_session
metric, now_sec = labelled_metric
resp = session.get(
f"{base_url}api/{org_id}/prometheus/api/v1/label/{LABEL_NAME}/values",
params={
"match[]": metric,
"start": str(now_sec - 3600),
"end": str(now_sec + 60),
},
)
assert resp.status_code == 200, \
f"scoped label values failed: {resp.status_code} {resp.text[:400]}"
values = resp.json().get("data", [])
assert JOB_VALUE in values, f"expected {JOB_VALUE!r} among {LABEL_NAME} values, got {values}"
@@ -0,0 +1,100 @@
"""PromQL offset and label_replace regressions [#11322, #11321].
Both fixed in PR #11335: an instant query with `offset` returned nothing, and a
label generated by `label_replace` was not visible to an aggregation grouping by
it.
#13892 (mixed-case `__name__` selectors) is deliberately NOT covered here: it
only reproduces with `ZO_FORMAT_STREAM_NAME_TO_LOWERCASE=false`, which is not the
default, so on a stock server the name is lowercased at ingest and the selector
legitimately finds nothing. `tests/api-testing/tests/regression/test_promql_mixed_case_metric_name.py`
already covers the related #13926 -- though note that folder is excluded from the
integration run, so it never executes.
"""
import time
import uuid
import pytest
from support.wait import wait_until
SAMPLES = 6
SPACING_SEC = 30
@pytest.fixture(scope="module")
def seeded_metric(create_session, base_url, org_id):
"""A gauge sampled every 30s, ~20 minutes in the past, on two instances."""
session = create_session
metric = f"pytest_selector_{uuid.uuid4().hex[:6]}"
t0 = (int(time.time()) // SPACING_SEC) * SPACING_SEC - 20 * 60
payload = [
{
"__name__": metric,
"__type__": "gauge",
"job": "pytest_selectors",
"instance": f"host{i % 2}",
"_timestamp": (t0 + i * SPACING_SEC) * 1000,
"value": float(i + 1),
}
for i in range(SAMPLES)
]
resp = session.post(f"{base_url}api/{org_id}/ingest/metrics/_json", json=payload)
assert resp.status_code == 200, f"metrics ingest failed: {resp.status_code} {resp.text[:400]}"
def visible():
r = session.get(
f"{base_url}api/{org_id}/prometheus/api/v1/query",
params={"query": f'{{__name__="{metric}"}}', "time": str(t0 + (SAMPLES - 1) * SPACING_SEC)},
)
return r.status_code == 200 and bool(r.json().get("data", {}).get("result"))
wait_until(visible, timeout=120, interval=2,
msg=f"metric {metric} never became queryable by __name__ selector")
yield metric, t0
session.delete(f"{base_url}api/{org_id}/streams/{metric}?type=metrics")
def _instant(session, base_url, org_id, query, at):
r = session.get(f"{base_url}api/{org_id}/prometheus/api/v1/query",
params={"query": query, "time": str(at)})
assert r.status_code == 200, f"instant query failed: {r.status_code} {r.text[:300]}"
body = r.json()
assert body.get("status") == "success", body
return body["data"]["result"]
def test_instant_query_with_offset_returns_the_earlier_sample(create_session, base_url, org_id, seeded_metric):
"""`offset` must shift the evaluation back, not return an empty result."""
metric, t0 = seeded_metric
at = t0 + (SAMPLES - 1) * SPACING_SEC
now_res = _instant(create_session, base_url, org_id, f'{{__name__="{metric}"}}', at)
off_res = _instant(create_session, base_url, org_id,
f'{{__name__="{metric}"}} offset {SPACING_SEC}s', at)
assert off_res, f"offset query returned nothing (the defect); plain query returned {len(now_res)}"
now_val = float(now_res[0]["value"][1])
off_val = float(off_res[0]["value"][1])
# Values ascend with time, so the offset sample must be strictly earlier/smaller.
assert off_val < now_val, \
f"offset must evaluate an earlier sample: offset={off_val} current={now_val}"
def test_label_replace_output_is_available_to_aggregation(create_session, base_url, org_id, seeded_metric):
"""A label generated by `label_replace` must be groupable by an aggregation."""
metric, t0 = seeded_metric
at = t0 + (SAMPLES - 1) * SPACING_SEC
query = (
f'sum by (derived) (label_replace({{__name__="{metric}"}}, '
f'"derived", "$1", "instance", "(.*)"))'
)
result = _instant(create_session, base_url, org_id, query, at)
assert result, "aggregation over a label_replace-generated label returned nothing"
derived = {r["metric"].get("derived") for r in result}
assert derived and all(d for d in derived), \
f"the generated label must reach the aggregation output, got {derived}"
@@ -0,0 +1,82 @@
"""User listing, org validation and default ingestion tokens [#9993, #1679, #12407].
#9993 (PR #10004): the root user was omitted from the users list on an OSS
build, so an operator could not see the account they were signed in as.
#1679 (PR #13806): `POST /api/{org_id}/users` accepted an empty or
whitespace-only org_id path segment, creating user records against a
meaningless organization.
#12407 (PR #12412): organizations created before ingestion tokens existed had
none, and no migration backfilled them. The fix creates the default token
lazily when the list endpoint is first called, so simply asking must yield one.
"""
import logging
import os
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
ROOT_EMAIL = os.environ.get("ZO_ROOT_USER_EMAIL", "root@example.com")
def test_root_user_is_listed(create_session, base_url):
"""The signed-in root account must appear in the org's user list."""
session = create_session
resp = session.get(f"{base_url}api/{ORG_ID}/users")
assert resp.status_code == 200, f"user list failed: {resp.status_code} {resp.text[:300]}"
body = resp.json()
rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
emails = {(r.get("email") or "").lower() for r in rows if isinstance(r, dict)}
logger.info("users listed: %s", sorted(emails))
assert ROOT_EMAIL.lower() in emails, \
f"root user {ROOT_EMAIL} must be listed, got {sorted(emails)}"
def test_ingestion_tokens_are_created_on_demand(create_session, base_url):
"""Listing tokens must yield the default one even if the org never had it."""
session = create_session
resp = session.get(f"{base_url}api/{ORG_ID}/ingestion-tokens")
assert resp.status_code == 200, \
f"ingestion-tokens list failed: {resp.status_code} {resp.text[:300]}"
body = resp.json()
rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
logger.info("ingestion tokens returned: %s", len(rows))
# The defect returned an empty list for orgs predating the feature.
assert rows, "listing ingestion tokens must create and return the default token"
def test_blank_org_id_does_not_create_a_user(create_session, base_url):
"""A whitespace-only org_id must not produce a user record.
The status alone is a weak signal: on this build org resolution answers 404
before `validate_org_id` can return its 400, and either layer refusing is
acceptable. What the defect actually produced was a user account against a
meaningless organization, so the assertion is that no such account exists.
"""
session = create_session
email = f"e2e_blankorg_{uuid.uuid4().hex[:8]}@example.com"
payload = {
"email": email,
"password": "Complexpass#123",
"role": "admin",
"first_name": "blank",
"last_name": "org",
}
resp = session.post(f"{base_url}api/%20/users", json=payload)
logger.info("blank org_id user create answered %s", resp.status_code)
assert resp.status_code >= 400, \
f"a whitespace org_id must be rejected, got {resp.status_code}: {resp.text[:300]}"
listing = session.get(f"{base_url}api/{ORG_ID}/users")
assert listing.status_code == 200, f"user list failed: {listing.status_code}"
body = listing.json()
rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
emails = {(r.get("email") or "").lower() for r in rows if isinstance(r, dict)}
assert email.lower() not in emails, \
f"{email} was created despite the blank org_id being rejected"
@@ -0,0 +1,58 @@
"""A malformed function must error, not panic [regression #7708].
Adding an incorrect function surfaced a panic rather than a usable error, so the
caller saw a failure with no indication of what was wrong -- and a panicking
request path is a liveness risk.
Asserting only the status would be weak: the contract is a *usable* error (a
message naming the problem) AND a server still answering afterwards, since a
panic can take the request down while the process restarts behind it.
"""
import logging
import os
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
BAD_VRL = ".x = this is not valid vrl ((("
def _alive(session, base_url):
return session.get(f"{base_url}api/{ORG_ID}/streams", timeout=30).status_code == 200
def test_malformed_function_returns_a_usable_error(create_session, base_url):
session = create_session
assert _alive(session, base_url), "server was already unhealthy before the request"
resp = session.post(
f"{base_url}api/{ORG_ID}/functions/test",
json={"function": BAD_VRL, "events": [{"a": 1}]},
)
logger.info("malformed function answered %s", resp.status_code)
assert 400 <= resp.status_code < 500, \
f"a malformed function should be a client error, got {resp.status_code}: {resp.text[:300]}"
body = resp.text.lower()
assert "error" in body or "syntax" in body, \
f"the response should explain what is wrong, got {resp.text[:300]}"
assert "panic" not in body, f"a panic leaked into the response: {resp.text[:300]}"
# A panicking path can answer and still have taken the worker down.
assert _alive(session, base_url), "server stopped answering after the malformed function"
def test_valid_function_still_works(create_session, base_url):
"""Control: the rejection is about the function, not all requests."""
session = create_session
resp = session.post(
f"{base_url}api/{ORG_ID}/functions/test",
json={"function": '.added = "ok"\n.', "events": [{"a": 1}]},
)
assert resp.status_code == 200, f"a valid function must succeed: {resp.status_code} {resp.text[:300]}"
results = resp.json().get("results", [])
assert results and results[0].get("event", {}).get("added") == "ok", \
f"the valid function did not apply: {resp.text[:300]}"
@@ -0,0 +1,63 @@
"""VRL parse_json must not leak serde_json's private number wrapper [#10919].
With `serde_json` built with `arbitrary_precision`, a high-precision number
deserialises into an object keyed `$serde_json::private::Number` rather than a
scalar. `parse_json` in VRL passed that object straight through, so flattening
produced a field literally named `__serde_json__private__number` instead of the
value. PR #10920 detects the wrapper and converts it to a float.
The assertion is therefore twofold: the parsed value must be numeric, AND no
field carrying the private wrapper's name may appear anywhere in the event --
checking only the value would miss the wrapper leaking alongside it.
"""
import logging
import os
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
# Enough digits that serde_json cannot hold it as a plain f64/i64.
HIGH_PRECISION = "12345678901234567890.123456789"
def _run_vrl(session, base_url, function, events):
resp = session.post(
f"{base_url}api/{ORG_ID}/functions/test",
json={"function": function, "events": events},
)
assert resp.status_code == 200, \
f"function test failed: {resp.status_code} {resp.text[:300]}"
results = resp.json().get("results", [])
assert results, f"no results returned: {resp.text[:300]}"
return results[0].get("event", {})
def test_parse_json_yields_a_number_for_high_precision_input(create_session, base_url):
event = _run_vrl(
create_session, base_url,
".parsed = parse_json!(.payload)\n.",
[{"payload": '{"big": ' + HIGH_PRECISION + '}'}],
)
logger.info("parsed event keys: %s", sorted(event.keys()))
assert "parsed_big" in event, f"the parsed number is missing: {event}"
assert isinstance(event["parsed_big"], (int, float)), \
f"parse_json must yield a number, got {type(event['parsed_big']).__name__}: {event['parsed_big']!r}"
def test_no_serde_json_private_wrapper_leaks_into_fields(create_session, base_url):
"""The wrapper object must not survive as a field name."""
event = _run_vrl(
create_session, base_url,
".parsed = parse_json!(.payload)\n.",
[{"payload": '{"big": ' + HIGH_PRECISION + ', "nested": {"inner": ' + HIGH_PRECISION + '}}'}],
)
leaked = [k for k in event if "serde_json" in k.lower() or "private" in k.lower()]
logger.info("event keys: %s", sorted(event.keys()))
assert not leaked, f"serde_json's private number wrapper leaked as field(s): {leaked}"
assert isinstance(event.get("parsed_nested_inner"), (int, float)), \
f"a nested high-precision number must also parse as a number: {event.get('parsed_nested_inner')!r}"
@@ -0,0 +1,80 @@
"""RUM must record the real client IP, not the proxy's [regression #11377].
Behind a proxy every RUM record was attributed to the proxy (or loopback), so
per-user geo and IP were meaningless. PR #11378 resolves the address through
`RealIp`, which honours the forwarded header, and stores it as `ip`.
The test sends the beacon with an explicit `X-Forwarded-For` and asserts the
stored value is that address rather than a loopback one -- asserting merely that
`ip` is non-empty would pass on the unfixed build, which happily stored the
proxy address.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
FORWARDED_IP = "203.0.113.7" # TEST-NET-3, never a real client
LOOPBACK = {"127.0.0.1", "::1"}
def _rum_token(session, base_url):
r = session.get(f"{base_url}api/{ORG_ID}/rumtoken")
assert r.status_code == 200, f"rum token fetch failed: {r.status_code} {r.text[:200]}"
body = r.json()
token = (body.get("data") or {}).get("rum_token") or body.get("rum_token")
assert token, f"no rum token in response: {body}"
return token
def test_forwarded_client_ip_is_recorded(create_session, base_url):
session = create_session
token = _rum_token(session, base_url)
service = f"pytest_realip_{uuid.uuid4().hex[:8]}"
beacon = {
"type": "view",
"date": int(time.time() * 1000),
"service": service,
"application": {"id": "pytest_realip"},
"session": {"id": uuid.uuid4().hex},
"view": {"id": uuid.uuid4().hex, "url": "http://example.test/"},
}
resp = session.post(
f"{base_url}rum/v1/{ORG_ID}/rum",
params={"oo-api-key": token},
headers={"X-Forwarded-For": FORWARDED_IP},
json=beacon,
)
assert resp.status_code in (200, 202), \
f"rum ingest failed: {resp.status_code} {resp.text[:300]}"
# RUM records carry their own `date`, so query a wide window rather than
# assuming the row lands near "now".
now_us = int(time.time() * 1_000_000)
row = None
for _ in range(30):
q = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {
"sql": f"SELECT ip, service FROM \"_rumdata\" WHERE service = '{service}'",
"start_time": now_us - 86_400_000_000, "end_time": now_us + 60_000_000,
"from": 0, "size": 5}},
)
if q.status_code == 200 and q.json().get("hits"):
row = q.json()["hits"][0]
break
time.sleep(2)
assert row, f"the RUM beacon for {service} never became searchable"
logger.info("stored RUM ip: %s", row.get("ip"))
assert row.get("ip") not in LOOPBACK, \
f"the proxy/loopback address was recorded instead of the client: {row.get('ip')}"
assert row.get("ip") == FORWARDED_IP, \
f"expected the forwarded client IP {FORWARDED_IP}, got {row.get('ip')!r}"
@@ -11,13 +11,12 @@ Test Coverage:
- Delete sourcemaps
Prerequisites:
- OpenObserve ENTERPRISE build running on ZO_BASE_URL
- OpenObserve running on ZO_BASE_URL (either edition; the sourcemaps routes are not enterprise-gated)
- RUM enabled (ZO_RUM_ENABLED=true)
- Static sourcemap fixtures at tests/api-testing/fixtures/sourcemaps/
Note:
- Tests use pre-built sourcemaps (no build step required)
- Tests marked with @pytest.mark.skip for OSS CI (enterprise-only feature)
- Tests must run serially due to shared module-scoped fixture
- Use: pytest test_sourcemap_api.py -v (no -n flag for parallel)
"""
@@ -25,10 +24,11 @@ Note:
import pytest
import logging
# Mark all tests in this module to run serially and skip in OSS (enterprise-only feature)
# Serial: these cases share one sourcemap store. They run on OSS too -- the
# sourcemaps routes are outside the enterprise cfg block in router/mod.rs, and
# the previous blanket skip (which never ran them on ANY build) was mistaken.
pytestmark = [
pytest.mark.order(1),
pytest.mark.skip(reason="Sourcemaps is an enterprise feature and cannot be tested in CI with non-enterprise build")
]
from helpers.sourcemap_helpers import (
load_static_sourcemaps,
@@ -19,7 +19,8 @@ Prerequisites:
- OpenFGA enabled for RBAC enforcement
Note:
- Tests marked with @pytest.mark.skip for OSS CI (enterprise-only feature)
- Skipped at runtime by `_require_custom_roles` when the build has no custom
roles (GET /roles answers 403 on OSS); they run on enterprise
- Tests must run serially due to shared module-scoped fixture
- Uses session-based authentication (matches enterprise RBAC test pattern)
"""
@@ -28,10 +29,8 @@ import pytest
import logging
import requests
# Mark all tests in this module to run serially and skip in OSS (enterprise-only feature)
pytestmark = [
pytest.mark.order(2),
pytest.mark.skip(reason="Sourcemaps is an enterprise feature and cannot be tested in CI with non-enterprise build")
]
logger = logging.getLogger(__name__)
@@ -40,6 +39,32 @@ logger = logging.getLogger(__name__)
test_users = {}
@pytest.fixture(scope="module", autouse=True)
def _require_custom_roles(create_session, base_url):
"""Skip on builds without custom roles, rather than skipping everywhere.
These cases need viewer/editor/admin accounts. An OSS build has no custom
roles, so the users cannot exist and every case errors at login.
This replaces an unconditional `pytest.mark.skip`, which skipped the module
on EVERY build -- enterprise included -- so these cases had never run
anywhere. The capability is probed rather than assumed from the edition, in
the same way `test_anomaly_conditions.py` probes its endpoint.
The probe is deliberately READ-ONLY. Creating and deleting a probe user
would mutate the shared `default` org from an autouse fixture, in parallel
(xdist) with the RBAC suite that tests user permissions -- `GET /roles`
answers 403 "Not Supported" on OSS and lists roles on enterprise, which is
the same signal without touching anything.
"""
probe = create_session.get(f"{base_url}api/default/roles")
if probe.status_code != 200:
pytest.skip(
f"custom roles are not available on this build "
f"(GET /roles -> {probe.status_code})"
)
@pytest.fixture(scope="module", autouse=True)
def setup_rbac_users(create_session, base_url, org_id):
"""
@@ -61,7 +86,10 @@ def setup_rbac_users(create_session, base_url, org_id):
for role in roles:
email = f"{role}@sourcemap-test.local"
password = "TestPass123!"
# The enterprise password policy requires at least 15 characters; the
# previous 12-character value was rejected with HTTP 400, so every case
# then failed at login.
password = "TestPass123!Secure"
logger.info(f"Creating user: {email} with role: {role}")
@@ -85,8 +113,18 @@ def setup_rbac_users(create_session, base_url, org_id):
# Create authenticated session via login endpoint
user_session = create_user_session(email, password, url)
if not user_session:
# Fail loudly rather than skip. The module gate already
# established that this build supports custom roles, so failing
# to provision one here is a real defect -- skipping would hide
# it, which is how the 15-character password policy went
# unnoticed. The create response is included so the cause is
# visible without re-running.
logger.error(f"Failed to create session for {email}")
raise Exception(f"Login failed for {email}")
raise AssertionError(
f"could not provision the {role} user ({email}) on a build that "
f"reports custom-role support; create returned "
f"{response.status_code}: {response.text[:200]}"
)
# Store user info
test_users[role] = {
@@ -0,0 +1,99 @@
"""A sourcemap ZIP with no valid pairs must error [regression #12951].
Uploading a ZIP that contained no usable source-map files reported success, so
an operator believed their maps were in place while error stack traces stayed
unminified. PR #12962 makes the upload fail instead.
The valid-ZIP control is what stops a build that rejects every upload from
passing: the error must be specific to there being nothing usable inside.
"""
import io
import logging
import os
import uuid
import zipfile
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
def _zip(entries: dict[str, str]) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
for name, body in entries.items():
zf.writestr(name, body)
return buf.getvalue()
def _upload(session, base_url, data: bytes, filename: str, service: str):
"""Upload under an explicit service so the group can be deleted afterwards."""
return session.post(
f"{base_url}api/{ORG_ID}/sourcemaps",
files={"file": (filename, data, "application/zip")},
data={"service": service, "env": "pytest", "version": "1.0.0"},
)
def _delete_group(session, base_url, service: str):
return session.delete(
f"{base_url}api/{ORG_ID}/sourcemaps",
params={"service": service, "env": "pytest", "version": "1.0.0"},
)
def _listed_count(session, base_url) -> int:
r = session.get(f"{base_url}api/{ORG_ID}/sourcemaps")
if r.status_code != 200:
return -1
body = r.json()
rows = body if isinstance(body, list) else body.get("data", body.get("list", []))
return len(rows) if isinstance(rows, list) else -1
def test_zip_without_any_sourcemap_is_rejected(create_session, base_url):
session = create_session
data = _zip({"README.txt": "no maps here", "notes/other.md": "# nothing"})
service = f"pytest_nomaps_{uuid.uuid4().hex[:6]}"
before = _listed_count(session, base_url)
resp = _upload(session, base_url, data, "nomaps.zip", service)
logger.info("no-sourcemap ZIP answered %s: %s", resp.status_code, resp.text[:160])
# Reporting success for a ZIP with nothing usable inside was the defect.
assert resp.status_code >= 400, \
f"a ZIP with no sourcemap must be rejected, got {resp.status_code}: {resp.text[:300]}"
assert "sourcemap" in resp.text.lower(), \
f"the error should say what was missing, got {resp.text[:300]}"
# A status alone would not catch a build that errored yet still stored something.
after = _listed_count(session, base_url)
assert after == before, \
f"a rejected upload must store nothing: sourcemap count went {before} -> {after}"
def test_zip_with_a_valid_sourcemap_is_accepted(create_session, base_url):
"""Control: the rejection must be about the contents, not all uploads."""
session = create_session
smap = (
'{"version":3,"file":"app.min.js","sources":["app.js"],'
'"names":[],"mappings":"AAAA","sourcesContent":["console.log(1)"]}'
)
data = _zip({"app.min.js.map": smap, "app.min.js": "console.log(1)\n"})
service = f"pytest_withmap_{uuid.uuid4().hex[:6]}"
before = _listed_count(session, base_url)
try:
resp = _upload(session, base_url, data, "withmap.zip", service)
logger.info("valid ZIP answered %s: %s", resp.status_code, resp.text[:160])
assert resp.status_code < 400, \
f"a ZIP containing a real sourcemap must be accepted, got {resp.status_code}: {resp.text[:300]}"
# Accepting and then storing nothing would be the same failure wearing a 2xx.
after = _listed_count(session, base_url)
assert after > before, \
f"an accepted upload must store the map: sourcemap count went {before} -> {after}"
finally:
# This test is the only one here that stores anything, so it reaps its own group.
_delete_group(session, base_url, service)
@@ -0,0 +1,72 @@
"""The result cache must key on query size [regression #10382].
`query.size` was missing from the cache hash, so a second search for the same
SQL over the same range with a DIFFERENT size was served the first size's cached
rows. PR #10384 includes size in the hash.
Order matters: the small query must run first so its result is what a
size-blind cache would wrongly replay for the larger one.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
def _drop(session, base_url, stream, stream_type="logs"):
"""Follows the suite convention: the test that seeds a stream drops it."""
session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
ROW_COUNT = 40
SMALL, LARGE = 5, 25
def _seed(session, base_url, stream):
now = int(time.time() * 1_000_000)
rows = [{"_timestamp": now - i * 1_000_000, "job": "pytest_cache_size", "n": i}
for i in range(ROW_COUNT)]
r = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=rows)
assert r.status_code == 200, f"ingest failed: {r.status_code} {r.text[:300]}"
start, end = now - 3_600_000_000, now + 60_000_000
for _ in range(30):
q = session.post(f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {"sql": f'SELECT count(*) AS n FROM "{stream}"',
"start_time": start, "end_time": end,
"from": 0, "size": 1}})
if q.status_code == 200 and (q.json().get("hits") or [{}])[0].get("n", 0) >= ROW_COUNT:
return start, end
time.sleep(1)
raise AssertionError(f"stream {stream} never became searchable")
def test_cached_result_is_not_replayed_for_a_different_size(create_session, base_url):
session = create_session
stream = f"e2e_cachesize_{uuid.uuid4().hex[:8]}"
try:
start, end = _seed(session, base_url, stream)
sql = f'SELECT * FROM "{stream}" ORDER BY _timestamp DESC'
def run(size):
r = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs&use_cache=true",
json={"query": {"sql": sql, "start_time": start, "end_time": end,
"from": 0, "size": size}},
)
assert r.status_code == 200, f"search failed: {r.status_code} {r.text[:300]}"
return len(r.json().get("hits", []))
small = run(SMALL)
large = run(LARGE)
logger.info("size %s -> %s rows; size %s -> %s rows", SMALL, small, LARGE, large)
assert small == SMALL, f"first query must return {SMALL} rows, got {small}"
# A size-blind cache replays the first result here, returning SMALL again.
assert large == LARGE, \
f"second query must return {LARGE} rows, got {large} (cache ignored query size)"
finally:
_drop(session, base_url, stream)
@@ -0,0 +1,106 @@
"""Many OR-ed match_all() conditions must not abort the process [regression #12398].
A single search that ORs together several `match_all()` terms aborted the whole
node with `fatal runtime error: stack overflow` -- one query could take the
server down, and it then restarted. Fixed in PR #14469 by hardening the leader
and follower against deeply nested full-text disjunctions.
The contract the issue states is "the query runs, or fails gracefully with an
error -- it must not abort". So the assertion is deliberately NOT that the query
succeeds: a well-formed error is an acceptable outcome. What must hold is that
the server is still answering afterwards, which is the half a status-code check
alone would miss -- a crashed node restarts, so a later request can succeed
while the query still killed the process.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
def _drop(session, base_url, stream, stream_type="logs"):
"""Follows the suite convention: the test that seeds a stream drops it."""
session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
ROW_COUNT = 60
TERMS = ["error", "Error", "exception", "Exception",
"Traceback", "FATAL", "panic", "CRITICAL"]
def _seed(session, base_url, stream):
now = int(time.time() * 1_000_000)
rows = [
{
"_timestamp": now - i * 1_000_000,
"container_name": f"c{i % 4}",
"project": f"p{i % 3}",
"message": f"{TERMS[i % len(TERMS)]} something happened {i}",
}
for i in range(ROW_COUNT)
]
resp = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=rows)
assert resp.status_code == 200, f"ingest failed: {resp.status_code} {resp.text[:400]}"
for _ in range(30):
r = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {
"sql": f'SELECT count(*) AS n FROM "{stream}"',
"start_time": now - 3_600_000_000, "end_time": now + 60_000_000,
"from": 0, "size": 1,
}},
)
if r.status_code == 200 and (r.json().get("hits") or [{}])[0].get("n", 0) >= ROW_COUNT:
return now
time.sleep(1)
raise AssertionError(f"stream {stream} never became searchable")
def _server_is_alive(session, base_url) -> bool:
r = session.get(f"{base_url}api/{ORG_ID}/streams", timeout=30)
return r.status_code == 200
def test_ored_match_all_terms_do_not_abort_the_server(create_session, base_url):
session = create_session
stream = f"e2e_matchall_or_{uuid.uuid4().hex[:8]}"
try:
now = _seed(session, base_url, stream)
assert _server_is_alive(session, base_url), "server was already unhealthy before the query"
disjunction = " OR ".join(f"match_all('{t}')" for t in TERMS)
sql = (
f'SELECT project AS proj, container_name, count(*) AS n FROM "{stream}" '
f"WHERE ({disjunction}) GROUP BY proj, container_name ORDER BY n DESC"
)
resp = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {
"sql": sql,
"start_time": now - 3_600_000_000, "end_time": now + 60_000_000,
"from": 0, "size": 100,
}},
)
logger.info("disjunction query answered %s", resp.status_code)
# A graceful error is acceptable; a dead connection is not.
assert resp.status_code < 500 or resp.text, \
f"query returned {resp.status_code} with no body, which suggests the node died"
# The real contract: the process is still serving.
assert _server_is_alive(session, base_url), \
"server stopped answering after the OR-ed match_all query, i.e. the query aborted it"
if resp.status_code == 200:
hits = resp.json().get("hits", [])
assert isinstance(hits, list), f"unexpected result shape: {resp.text[:300]}"
logger.info("query succeeded with %d grouped rows", len(hits))
finally:
_drop(session, base_url, stream)
@@ -0,0 +1,67 @@
"""The result cache must not return duplicate records [regression #4274].
Overlapping cached ranges replayed rows that the fresh half of the query also
returned, so a search spanning a cached window came back with duplicates. PR
#4293 dedupes on timestamp.
The first query must be cached and the second must OVERLAP it -- a
non-overlapping range exercises nothing, and running only one query cannot
produce the duplication at all.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
def _drop(session, base_url, stream, stream_type="logs"):
"""Follows the suite convention: the test that seeds a stream drops it."""
session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
ROW_COUNT = 40
def test_overlapping_cached_ranges_do_not_duplicate_rows(create_session, base_url):
session = create_session
stream = f"e2e_cachedup_{uuid.uuid4().hex[:8]}"
try:
now = int(time.time() * 1_000_000)
rows = [{"_timestamp": now - i * 1_000_000, "job": "pytest_cache_dup", "n": i}
for i in range(ROW_COUNT)]
r = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=rows)
assert r.status_code == 200, f"ingest failed: {r.status_code} {r.text[:300]}"
oldest = now - (ROW_COUNT - 1) * 1_000_000
sql = f'SELECT _timestamp, n FROM "{stream}" ORDER BY _timestamp DESC'
def run(start, end):
resp = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs&use_cache=true",
json={"query": {"sql": sql, "start_time": start, "end_time": end,
"from": 0, "size": 200}},
)
assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:300]}"
return resp.json().get("hits", [])
for _ in range(30):
if len(run(oldest - 60_000_000, now + 60_000_000)) >= ROW_COUNT:
break
time.sleep(1)
# Warm the cache over the inner half, then query a window that overlaps it.
inner = run(oldest + 10 * 1_000_000, now - 5 * 1_000_000)
outer = run(oldest - 60_000_000, now + 60_000_000)
logger.info("inner=%s outer=%s rows", len(inner), len(outer))
stamps = [h["_timestamp"] for h in outer]
dupes = {t for t in stamps if stamps.count(t) > 1}
assert not dupes, f"{len(dupes)} timestamps were returned more than once: {sorted(dupes)[:5]}"
assert len(stamps) == len(set(stamps)) == ROW_COUNT, \
f"expected {ROW_COUNT} unique rows, got {len(stamps)} ({len(set(stamps))} unique)"
finally:
_drop(session, base_url, stream)
@@ -0,0 +1,125 @@
"""SQL LIMIT precedence and awkward stream names [regressions #8372, #10378].
#8372 (PR #14473): an explicit SQL `LIMIT` must win over the request's `size`.
`_search` already honoured it because `AddSortAndLimit` leaves an existing Limit
node alone; the streaming path did not. The contract asserted here is the one
both paths must agree on -- LIMIT decides, size does not override it.
#10378 (PR #10394): a stream whose name contains `:` produced a ParserError
because the name was interpolated into SQL unquoted. Quoting it fixes the parse,
so the assertion is that such a stream is queryable at all.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
def _drop(session, base_url, stream, stream_type="logs"):
"""Follows the suite convention: the test that seeds a stream drops it."""
session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
ROW_COUNT = 30
SQL_LIMIT = 7
REQUEST_SIZE = 100
def _window():
now = int(time.time() * 1_000_000)
return now - 3_600_000_000, now + 60_000_000
def _seed(session, base_url, stream, rows=ROW_COUNT):
now = int(time.time() * 1_000_000)
payload = [
{"_timestamp": now - i * 1_000_000, "job": "pytest_sql_limit", "n": i,
"message": f"row {i}"}
for i in range(rows)
]
resp = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=payload)
assert resp.status_code == 200, f"ingest failed: {resp.status_code} {resp.text[:400]}"
start, end = _window()
for _ in range(30):
r = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {"sql": f'SELECT count(*) AS n FROM "{stream}"',
"start_time": start, "end_time": end, "from": 0, "size": 1}},
)
if r.status_code == 200 and (r.json().get("hits") or [{}])[0].get("n", 0) >= rows:
return
time.sleep(1)
raise AssertionError(f"stream {stream} never became searchable")
def test_explicit_sql_limit_wins_over_request_size(create_session, base_url):
"""`LIMIT 7` with `size: 100` must return 7 rows, not 100."""
session = create_session
stream = f"e2e_sqllimit_{uuid.uuid4().hex[:8]}"
try:
_seed(session, base_url, stream)
start, end = _window()
resp = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {
"sql": f'SELECT * FROM "{stream}" ORDER BY _timestamp DESC LIMIT {SQL_LIMIT}',
"start_time": start, "end_time": end, "from": 0, "size": REQUEST_SIZE,
}},
)
assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:400]}"
hits = resp.json().get("hits", [])
logger.info("LIMIT %s with size %s returned %s rows", SQL_LIMIT, REQUEST_SIZE, len(hits))
# There are more rows than the LIMIT, so a build that ignores it returns more.
assert len(hits) == SQL_LIMIT, \
f"explicit SQL LIMIT {SQL_LIMIT} must decide the row count, got {len(hits)}"
finally:
_drop(session, base_url, stream)
def test_request_size_still_applies_without_an_explicit_limit(create_session, base_url):
"""A negative control: with no SQL LIMIT, `size` is what caps the result."""
session = create_session
stream = f"e2e_sqllimit_{uuid.uuid4().hex[:8]}"
try:
_seed(session, base_url, stream)
start, end = _window()
size = 5
resp = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {"sql": f'SELECT * FROM "{stream}"',
"start_time": start, "end_time": end, "from": 0, "size": size}},
)
assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:400]}"
hits = resp.json().get("hits", [])
logger.info("no SQL LIMIT, size %s returned %s rows", size, len(hits))
assert len(hits) == size, f"size must cap the result when the SQL has no LIMIT, got {len(hits)}"
finally:
_drop(session, base_url, stream)
def test_stream_name_containing_a_colon_is_queryable(create_session, base_url):
"""A `:` in the stream name must be quoted into the SQL, not raise a ParserError."""
session = create_session
stream = f"e2e:colon:{uuid.uuid4().hex[:6]}"
try:
_seed(session, base_url, stream, rows=5)
start, end = _window()
resp = session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {"sql": f'SELECT * FROM "{stream}"',
"start_time": start, "end_time": end, "from": 0, "size": 10}},
)
logger.info("colon-named stream answered %s", resp.status_code)
assert resp.status_code == 200, \
f"a colon in the stream name must not break parsing: {resp.status_code} {resp.text[:400]}"
assert "ParserError" not in resp.text, f"ParserError surfaced: {resp.text[:300]}"
assert len(resp.json().get("hits", [])) == 5, "all seeded rows must come back"
finally:
_drop(session, base_url, stream)
@@ -0,0 +1,153 @@
"""Traces list ordering and duration filtering [regressions #10986, #11339, #11505].
#10986 (PR #10993): the traces list did not come back ordered by `_timestamp`
descending, so the newest trace was not first.
#11339 (PR #11334): partition ordering ignored the requested sort direction --
partitions stay as-is for `zo_sql_timestamp DESC` and must be reversed for ASC.
Asserting both directions is the point: a build that always returns one fixed
order satisfies a single-direction test.
#11505 (PR #11514): filtering the list by duration produced blank rows, because
the per-trace enrichment query was unbounded. The contract is that a filtered
row still carries its trace identity, not that some number of rows comes back.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
TRACE_COUNT = 6
STREAM = "default"
def _span(trace_id, span_id, start_ns, dur_ns, name):
return {
"traceId": trace_id, "spanId": span_id, "parentSpanId": "", "flags": 1,
"name": name, "kind": 2,
"startTimeUnixNano": str(start_ns), "endTimeUnixNano": str(start_ns + dur_ns),
"attributes": [{"key": "http.route", "value": {"stringValue": "/probe"}}],
"status": {"code": 0},
}
def _seed_traces(session, base_url):
"""Six traces a second apart, with deliberately different durations."""
base_ns = int(time.time() * 1e9) - 300 * 10**9
ids = []
spans = []
for i in range(TRACE_COUNT):
tid = uuid.uuid4().hex
ids.append(tid)
spans.append(_span(tid, uuid.uuid4().hex[:16], base_ns + i * 10**9,
(i + 1) * 5 * 10**6, f"op_{i}"))
payload = {"resourceSpans": [{
"resource": {"attributes": [{"key": "service.name",
"value": {"stringValue": "pytest_traces_order"}}]},
"scopeSpans": [{"scope": {"name": "pytest"}, "spans": spans}],
}]}
r = session.post(f"{base_url}api/{ORG_ID}/v1/traces", json=payload)
assert r.status_code == 200, f"trace ingest failed: {r.status_code} {r.text[:300]}"
return ids, base_ns
def _seed_window(base_ns):
"""Bound the query to the seeded span only.
The traces stream is shared and earlier runs leave data behind. With a wide
window `sort_order=asc` returns the oldest traces in the whole stream and
`size` truncates before this module's trace_id filter can apply, so the two
directions would compare different populations.
"""
start_us = int(base_ns / 1000) - 1_000_000
end_us = int(base_ns / 1000) + (TRACE_COUNT + 2) * 1_000_000
return start_us, end_us
def _latest(session, base_url, start_us, end_us, **extra):
params = {"start_time": start_us, "end_time": end_us, "from": 0, "size": 20}
params.update(extra)
r = session.get(f"{base_url}api/{ORG_ID}/{STREAM}/traces/latest", params=params)
assert r.status_code == 200, f"latest traces failed: {r.status_code} {r.text[:300]}"
return r.json()
def _wait_for_traces(session, base_url, ids, start_us, end_us):
for _ in range(40):
body = _latest(session, base_url, start_us, end_us)
got = {h.get("trace_id") for h in body.get("hits", [])}
if len([i for i in ids if i in got]) >= TRACE_COUNT:
return body
time.sleep(2)
raise AssertionError("seeded traces never became visible in the latest-traces list")
def test_latest_traces_are_ordered_newest_first(create_session, base_url):
session = create_session
ids, base_ns = _seed_traces(session, base_url)
start_us, end_us = _seed_window(base_ns)
body = _wait_for_traces(session, base_url, ids, start_us, end_us)
# Restrict to this run's traces; the stream is shared and older ones linger.
mine = [h for h in body["hits"] if h.get("trace_id") in set(ids)]
assert len(mine) == TRACE_COUNT, f"expected {TRACE_COUNT} seeded traces, got {len(mine)}"
stamps = [h["start_time"] for h in mine]
logger.info("latest-traces start_times: %s", stamps)
# Seeded one second apart, so a stable sort cannot mask a wrong direction.
assert stamps == sorted(stamps, reverse=True), \
f"traces must be newest-first, got {stamps}"
def test_duration_filtered_rows_are_not_blank(create_session, base_url):
session = create_session
ids, base_ns = _seed_traces(session, base_url)
start_us, end_us = _seed_window(base_ns)
_wait_for_traces(session, base_url, ids, start_us, end_us)
# Durations are (i+1)*5ms, so this threshold must exclude the shortest traces
# rather than matching everything -- a filter that is a no-op proves nothing.
threshold_us = 15_000
body = _latest(session, base_url, start_us, end_us, filter=f"duration > {threshold_us}")
mine = [h for h in body.get("hits", []) if h.get("trace_id") in set(ids)]
logger.info("duration-filtered rows: %s of %s seeded", len(mine), TRACE_COUNT)
assert mine, "a duration filter matching some traces must still return rows"
assert len(mine) < TRACE_COUNT, \
f"the filter must exclude the shorter traces, got all {len(mine)}"
# The defect rendered rows with no trace identity, so populated rows are the contract.
for h in mine:
assert h.get("trace_id"), f"filtered row came back blank: {h}"
assert h.get("duration", 0) > threshold_us, \
f"row below the duration threshold leaked through: {h.get('duration')}"
def test_sort_order_asc_reverses_the_list(create_session, base_url):
"""`sort_order=asc` must reverse the order, not return the same fixed list."""
session = create_session
ids, base_ns = _seed_traces(session, base_url)
start_us, end_us = _seed_window(base_ns)
_wait_for_traces(session, base_url, ids, start_us, end_us)
seeded = set(ids)
desc = [h["start_time"] for h in
_latest(session, base_url, start_us, end_us, sort_order="desc")["hits"]
if h.get("trace_id") in seeded]
asc = [h["start_time"] for h in
_latest(session, base_url, start_us, end_us, sort_order="asc")["hits"]
if h.get("trace_id") in seeded]
logger.info("desc=%s asc=%s", desc, asc)
assert desc == sorted(desc, reverse=True), f"desc must be newest-first, got {desc}"
assert asc == sorted(asc), f"asc must be oldest-first, got {asc}"
# The defect ignored the requested direction, so the two must actually differ.
assert asc == list(reversed(desc)), \
f"asc must be the reverse of desc; asc={asc} desc={desc}"
@@ -0,0 +1,75 @@
"""Boolean to numeric schema evolution [regression #10475].
A field first seen as a boolean and later as a number was rejected, because
`is_widening_conversion` did not permit boolean widening. PR #10478 expanded the
allowed conversions so the schema evolves instead of failing the write.
The second ingest is the contract. Asserting only that the first succeeds would
pass on any build, so the test checks the later records are accepted AND that
both generations are queryable afterwards -- a build that accepted the write but
dropped the rows would otherwise look fixed.
"""
import logging
import os
import time
import uuid
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
ORG_ID = os.environ.get("TEST_ORG_ID", "default")
def _drop(session, base_url, stream, stream_type="logs"):
"""Follows the suite convention: the test that seeds a stream drops it."""
session.delete(f"{base_url}api/{ORG_ID}/streams/{stream}?type={stream_type}")
def _window():
now = int(time.time() * 1_000_000)
return now - 3_600_000_000, now + 60_000_000
def _search(session, base_url, sql):
start, end = _window()
return session.post(
f"{base_url}api/{ORG_ID}/_search?type=logs",
json={"query": {"sql": sql, "start_time": start, "end_time": end,
"from": 0, "size": 100}},
)
def test_boolean_field_widens_to_numeric(create_session, base_url):
session = create_session
stream = f"e2e_evolve_{uuid.uuid4().hex[:8]}"
try:
now = int(time.time() * 1_000_000)
bool_rows = [{"_timestamp": now - 5_000_000 + i, "job": "pytest_evolve",
"flag": True, "tag": "boolgen"} for i in range(3)]
r1 = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=bool_rows)
assert r1.status_code == 200, f"boolean ingest failed: {r1.status_code} {r1.text[:300]}"
num_rows = [{"_timestamp": now + i, "job": "pytest_evolve",
"flag": 7, "tag": "numgen"} for i in range(3)]
r2 = session.post(f"{base_url}api/{ORG_ID}/{stream}/_json", json=num_rows)
assert r2.status_code == 200, f"numeric ingest failed: {r2.status_code} {r2.text[:300]}"
body = r2.json()
failed = sum(s.get("failed", 0) for s in body.get("status", []))
logger.info("numeric generation ingest status: %s", body.get("status"))
assert failed == 0, f"numeric records were rejected after a boolean field: {body}"
for _ in range(30):
resp = _search(session, base_url, f'SELECT tag, flag FROM "{stream}"')
if resp.status_code == 200 and len(resp.json().get("hits", [])) >= 6:
break
time.sleep(1)
assert resp.status_code == 200, f"search failed: {resp.status_code} {resp.text[:300]}"
tags = [h.get("tag") for h in resp.json()["hits"]]
logger.info("rows visible after evolution: %s", sorted(tags))
assert tags.count("boolgen") == 3, f"boolean-generation rows lost: {tags}"
assert tags.count("numgen") == 3, f"numeric-generation rows lost: {tags}"
finally:
_drop(session, base_url, stream)
@@ -21,7 +21,8 @@
"logs-multistream-matchall-pagination.spec.js",
"logs-14303-streaming-render.spec.js",
"logs-multistream-share-url.spec.js",
"logs-histogram-severity.spec.js"
"logs-histogram-severity.spec.js",
"logs-field-value-rendering.spec.js"
],
"disabled": []
},
@@ -102,7 +103,8 @@
"browser": "chrome",
"run_files": [
"ui-regression.spec.js",
"landing-regression.spec.js"
"landing-regression.spec.js",
"internal-route-guards.spec.js"
],
"disabled": []
},
@@ -0,0 +1,45 @@
import { expect } from '@playwright/test';
/** Internal pages that must not be reachable in a shipped build. */
export class RouteGuardPage {
constructor(page) {
this.page = page;
// The removed demo rendered its own heading; the URL alone is not the tell.
this.emptyStateDemoHeading = page.getByText('Empty states', { exact: false });
this.notFoundHeading = page.getByText('Page not found', { exact: false });
}
/** Navigate to an app route under /web, carrying the org identifier. */
async gotoAppRoute(routePath, orgIdentifier) {
const org = orgIdentifier || process.env.ORGNAME || 'default';
const base = process.env.ZO_BASE_URL.replace(/\/$/, '');
await this.page.goto(`${base}/web/${routePath}?org_identifier=${org}`);
await this.page.waitForLoadState('networkidle', { timeout: 15000 }).catch(() => {});
}
/** build_type as the server reports it, so an edition-specific guard can be skipped elsewhere. */
async getBuildType() {
const base = process.env.ZO_BASE_URL.replace(/\/$/, '');
return await this.page.evaluate(async (url) => {
try {
const r = await fetch(`${url}/config`, { credentials: 'include' });
return (await r.json()).build_type || 'unknown';
} catch {
return 'unknown';
}
}, base);
}
async expectEmptyStateDemoNotRendered() {
await expect(this.emptyStateDemoHeading).toHaveCount(0, { timeout: 10000 });
}
/** A removed route falls through to the 404 view rather than changing the URL. */
async expectNotFoundRendered() {
await expect(this.notFoundHeading.first()).toBeVisible({ timeout: 15000 });
}
currentUrl() {
return this.page.url();
}
}
@@ -511,6 +511,8 @@ export class LogsPage {
// The "=" icon is revealed on hover over its sidebar field row.
this.fieldListItemPrefix = 'logs-field-list-item-';
this.logDetailRowPrefix = 'log-detail-row-';
this.fieldValuesList = '[data-test="field-values-panel-values-list"]';
this.fieldValueLabelPrefix = (field) => `[data-test^="logs-search-subfield-add-${field}-"]`;
this.fieldEqualsButton = (field) => `[data-test="log-search-index-list-filter-${field}-field-btn"]`;
this.logDetailRow = (field) => `[data-test="log-detail-row-${field}"]`;
this.logDetailFieldMenuTrigger = '[data-test="log-details-include-exclude-field-btn"]';
@@ -2045,6 +2047,28 @@ export class LogsPage {
.toContainText(`1 to ${size}`, { timeout: 30000 });
}
/**
* Wait until the banner reports the expected total.
*
* Ingest acknowledges before every row is searchable, so a query fired
* immediately after seeding sees an arbitrary prefix of the rows -- the same
* seed produced 12, then 6, then 3 across consecutive runs. Re-running the
* query while polling lets the index catch up.
*/
async waitForResultTotal(expected, attempts = 12) {
for (let i = 0; i < attempts; i++) {
const title = await this.getResultTitleText();
const total = Number((title.match(/out of\s+([\d.]+)/) || [])[1]);
if (total === expected) return title;
await this.page.waitForTimeout(2000);
await this.clickRefreshButton();
await this.expectResultsGridSettledWithRows();
}
throw new Error(
`banner never reported ${expected} events; last was "${(await this.getResultTitleText()).replace(/\n/g, ' | ')}"`,
);
}
/** The "1 to N out of M events" banner, used to compare the claim against rendered rows. */
async getResultTitleText() {
return (await this.page.locator(this.paginationRowCountTitle).first().innerText()).trim();
@@ -2059,6 +2083,25 @@ export class LogsPage {
return await this.page.locator(this.highlightedMatch).count();
}
/**
* Rendered text and wrapping of a sidebar field's first value row.
*
* The wrapping lives on an inner value div carrying `title`, not on the
* labelled element itself, so reading the label's own computed style
* reports the wrapper's `normal` and misses the fix entirely.
*/
async getFirstFieldValueRendering(field) {
const label = this.page.locator(this.fieldValueLabelPrefix(field)).first();
await label.waitFor({ state: 'visible', timeout: 20000 });
const value = label.locator('div[title]').first();
await value.waitFor({ state: 'visible', timeout: 10000 });
return await value.evaluate((el) => ({
whiteSpace: getComputedStyle(el).whiteSpace,
text: el.innerText,
newlineCount: (el.innerText.match(/\n/g) || []).length,
}));
}
/** Sidebar fields that offer an "=" filter, in render order; the set varies by dataset. */
async getFilterableSidebarFields(count = 2) {
const fields = await this.page
+2
View File
@@ -51,6 +51,7 @@ import { IngestionPage } from "./generalPages/ingestionPage.js";
import { CloudLoginPage } from "./cloudPages/cloudLoginPage.js";
import { isCloudEnvironment } from "./cloudPages/cloud-env.js";
import { IngestionConfigPage } from "./generalPages/ingestionConfigPage.js";
import { RouteGuardPage } from "./generalPages/routeGuardPage.js";
import { SplunkHecPage } from "./generalPages/splunkHecPage.js";
// ===== GENERAL TESTS ADDITIONAL PAGE OBJECTS =====
@@ -201,6 +202,7 @@ class PageManager {
this.pipelinesFormValidation = new PipelinesFormValidationPage(page);
this.loginPage = isCloudEnvironment() ? new CloudLoginPage(page) : new LoginPage(page);
this.ingestionPage = new IngestionPage(page);
this.routeGuardPage = new RouteGuardPage(page);
this.ingestionConfigPage = new IngestionConfigPage(page);
this.splunkHecPage = new SplunkHecPage(page);
@@ -0,0 +1,25 @@
const { test, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
const testLogger = require('../../utils/test-logger.js');
const PageManager = require('../../../pages/page-manager.js');
test.describe("Internal Route Guards Regression", () => {
test.describe.configure({ mode: 'parallel' });
let pm;
test.beforeEach(async ({ page }, testInfo) => {
testLogger.testStart(testInfo.title, testInfo.file);
await navigateToBase(page);
pm = new PageManager(page);
});
test("the internal empty-state demo page is not shipped @bug-14375 @P2 @regression @routeGuards", async () => {
await pm.routeGuardPage.gotoAppRoute('empty-state-demo');
// The route and its component were removed outright. The router keeps the
// requested URL and falls through to the 404 view, so the demo's own heading
// and the 404 view are the contract -- not the address bar.
await pm.routeGuardPage.expectEmptyStateDemoNotRendered();
await pm.routeGuardPage.expectNotFoundRendered();
testLogger.info(`empty-state-demo fell through to the 404 view`);
});
});
@@ -1970,6 +1970,45 @@ test.describe("Logs Regression Bug Fixes", () => {
// Bug #14228: Builder saved view does not apply when already on the Build tab
// https://github.com/openobserve/openobserve/issues/14228
// ==========================================================================
// ==========================================================================
// Bug #13990: Builder saved view restores the wrong stream
// https://github.com/openobserve/openobserve/issues/13990
// ==========================================================================
test('Builder saved view restores the stream it was saved with @bug-13990 @P1 @regression @logsRegression @savedViews @queryBuilder', {
tag: ['@bug-13990', '@P1', '@regression', '@logsRegression', '@savedViews', '@queryBuilder']
}, async ({ page }) => {
testLogger.info('Test: Builder saved view restores its own stream (Bug #13990)');
const savedViewName = `streamslog_build_13990_${Date.now()}`;
const otherStream = 'e2e_13990_other_' + Math.random().toString(36).slice(2, 7);
// A second stream to move away to; without it the assertion would hold
// trivially, since the page would never have left the saved stream.
const orgId = getOrgIdentifier() || 'default';
await sendRequest(page, getIngestionUrl(orgId, otherStream), [{
level: 'info', job: 'test_13990', log: 'second stream for saved view restore', e2e: '1',
}], getHeaders());
fieldCacheStreamsToCleanup.push(otherStream);
await page.goto(`${logData.logsUrl}?org_identifier=${orgId}`);
await page.waitForLoadState('networkidle', { timeout: 15000 }).catch(() => {});
await pm.logsPage.selectStream('e2e_automate');
await pm.logsPage.expectLogsSearchIndexListContainsText('e2e_automate');
await createBuilderSavedView(pm, page, savedViewName, 'line');
await pm.logsPage.selectStream(otherStream);
await pm.logsPage.expectLogsSearchIndexListContainsText(otherStream);
testLogger.info(`Moved to ${otherStream} before applying the saved view`);
await applySavedViewByName(pm, page, savedViewName);
// The defect restored an unrelated stream, so the stream is the contract here
// rather than the chart type that #14228 covers.
await pm.logsPage.expectLogsSearchIndexListContainsText('e2e_automate');
testLogger.info('Saved view restored its original stream');
});
test("should apply a builder saved view while already on the build tab", {
tag: ['@bug-14228', '@P1', '@regression', '@logsRegression', '@savedViews', '@queryBuilder']
}, async ({ page }) => {
@@ -0,0 +1,45 @@
const { test, expect, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
const testLogger = require('../../utils/test-logger.js');
const PageManager = require('../../../pages/page-manager.js');
const { sendRequest, getHeaders, getIngestionUrl } = require('../../utils/data-ingestion.js');
const { getOrgIdentifier } = require('../../utils/cloud-auth.js');
const MULTILINE_FIELD = 'stacktrace';
const MULTILINE_VALUE = 'Traceback (most recent call last):\n File "a.py", line 1\n raise ValueError';
test.describe("Logs Field Value Rendering Regression", () => {
test.describe.configure({ mode: 'parallel' });
let pm;
let stream;
test.beforeEach(async ({ page }, testInfo) => {
testLogger.testStart(testInfo.title, testInfo.file);
await navigateToBase(page);
pm = new PageManager(page);
// A field whose value genuinely contains newlines; the shared fixture has none.
stream = 'e2e_13354_' + Math.random().toString(36).slice(2, 7);
const orgId = getOrgIdentifier() || 'default';
await sendRequest(page, getIngestionUrl(orgId, stream), [{
level: 'error', job: 'test_13354', [MULTILINE_FIELD]: MULTILINE_VALUE,
}], getHeaders());
await pm.logsPage.selectStream(stream);
await pm.logsPage.clickRefreshButton();
});
test("multiline field values keep their newlines in the values panel @bug-13354 @P2 @regression @logsRegression", async () => {
await pm.logsPage.clickFieldExpandButton(MULTILINE_FIELD);
const rendering = await pm.logsPage.getFirstFieldValueRendering(MULTILINE_FIELD);
testLogger.info(`white-space=${rendering.whiteSpace} newlines=${rendering.newlineCount}`);
// The label collapsed newlines because it did not carry pre-wrap, so the
// wrapping mode is the contract -- the text alone can look right while the
// browser still renders it on one line.
expect(rendering.whiteSpace, 'the value label must preserve newlines')
.toMatch(/pre-wrap|pre-line|pre/);
expect(rendering.newlineCount, 'the rendered value must still contain its newlines')
.toBeGreaterThan(0);
});
});
@@ -1,10 +1,13 @@
const { test, expect, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
const testLogger = require('../../utils/test-logger.js');
const PageManager = require('../../../pages/page-manager.js');
const { ingestTestData } = require('../../utils/data-ingestion.js');
const { ingestTestData, sendRequest, getHeaders, getIngestionUrl } = require('../../utils/data-ingestion.js');
const { getOrgIdentifier } = require('../../utils/cloud-auth.js');
const logData = require('../../../fixtures/log.json');
const STREAM = 'e2e_automate';
const PAGE_SIZE = 100;
const DENSE_ROWS = 12;
test.describe("Logs Result Display Regression", () => {
test.describe.configure({ mode: 'parallel' });
@@ -52,4 +55,49 @@ test.describe("Logs Result Display Regression", () => {
expect(rendered, 'rendered rows must match the page size the banner claims')
.toBe(PAGE_SIZE);
});
test("the event total matches the range, not the histogram bucket @bug-13896 @P1 @regression @logsRegression", async ({ page }) => {
// The counter was summed from histogram buckets, so it only diverges when the
// selected range is NARROWER than the histogram interval -- hence one dense
// second of its own stream rather than the shared fixture.
const denseStream = 'e2e_13896_' + Math.random().toString(36).slice(2, 7);
const orgId = getOrgIdentifier() || 'default';
const secondStartMs = Date.now() - 5 * 60 * 1000;
const rows = Array.from({ length: DENSE_ROWS }, (_, i) => ({
_timestamp: (secondStartMs + i * 20) * 1000,
level: 'info',
job: 'test_13896',
log: `dense row ${i}`,
}));
await sendRequest(page, getIngestionUrl(orgId, denseStream), rows, getHeaders());
await pm.logsPage.waitForStreamAvailable(denseStream);
await page.goto(
`${logData.logsUrl}?org_identifier=${orgId}&stream=${denseStream}&stream_type=logs` +
`&from=${secondStartMs}&to=${secondStartMs + 1000}`,
);
await pm.logsPage.clickRefreshButton();
await pm.logsPage.expectResultsGridSettledWithRows();
// Ingest acks before every row is searchable, so settle on the full count
// first -- otherwise the banner reports whatever prefix happens to be
// indexed and the comparison below is against the wrong number.
const title = await pm.logsPage.waitForResultTotal(DENSE_ROWS);
const rendered = await pm.logsPage.getLogRowCount();
testLogger.info(`Banner: ${title.replace(/\n/g, ' | ')} | rendered rows: ${rendered}`);
const claimedTotal = Number((title.match(/out of\s+([\d.]+)/) || [])[1]);
expect(Number.isFinite(claimedTotal), `could not read a total from: ${title}`).toBeTruthy();
// The defect overstated the total by an order of magnitude, counting the whole
// histogram bucket while the log query correctly filtered to the range.
// Asserted as EQUAL, not <=: a `<=` here also passes when the window is wrong
// and only part of the seeded second is matched, which is a weaker test than
// it looks. from/to are milliseconds (see logsUtils), while ingest takes
// microseconds -- both denote the same instant.
expect(claimedTotal, 'the banner total must equal the rows seeded in the range')
.toBe(DENSE_ROWS);
expect(rendered, 'all rows in the one-second window must render').toBe(claimedTotal);
});
});
@@ -90,6 +90,7 @@ test.describe("Pre-Test Cleanup", () => {
// them. Harmless on CI's throwaway server, but they pile up on a shared dev env.
await pm.apiCleanup.cleanupStreams([/^e2e_14238_/], [], { streamType: 'metrics' });
// Clean up all reports owned by automation user
await pm.apiCleanup.cleanupReports();
@@ -297,6 +298,9 @@ test.describe("Pre-Test Cleanup", () => {
await pm.apiCleanup.cleanupStreams(
[
/^e2e_slo_14269_/, // slo-error-messages.spec.js (#14269) seeded stream
/^e2e_13354_/, // logs-field-value-rendering.spec.js (#13354)
/^e2e_13896_/, // logs-result-display.spec.js (#13896)
/^e2e_13990_other_/, // logs-bugs.spec.js (#13990)
/^e2e_sev_str_/, // logs-histogram-severity.spec.js (#11353)
/^e2e_sev_num_/, // logs-histogram-severity.spec.js (#11441)
/^e2e_ms7332_/, // logs-multistream-share-url.spec.js (#7332)