test(regression): cover 11 closed logs, streams, pipeline and enrichment bugs (#14978)

tests/api-testing/tests/pipelines/test_enrichment_table_validation.py

@@ -0,0 +1,136 @@
+"""Enrichment table name validation and VRL lookup  [regressions #2072, #5344].
+
+#2072: the UI refused a whitespace-only table name but the API accepted it,
+so a table could be created that the UI could never address again.
+
+#5344: `get_enrichment_table_record` returned the row but the looked-up value
+never reached the event, which silently produced un-enriched records rather
+than an error.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import time
+import uuid
+
+import pytest
+
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+# Both the file-upload and URL handlers validate the name, so both are exercised.
+BLANK_NAMES = ["%20%20", "   ", "%09"]
+
+ENRICHMENT_CSV = b"code,label\nc1,ALPHA\nc2,BETA\n"
+
+# The table is written asynchronously, so the VRL lookup needs it to settle first.
+TABLE_READY_TIMEOUT = 60
+
+
+def _delete_table(session, base_url, name):
+    session.delete(
+        f"{base_url}api/{ORG_ID}/streams/{name}?type=enrichment_tables", timeout=30
+    )
+
+
+def _wait_for_lookup(session, base_url, vrl, events):
+    """Poll the function-test endpoint until the table backs the lookup."""
+    deadline = time.time() + TABLE_READY_TIMEOUT
+    last = None
+    while time.time() < deadline:
+        last = session.post(
+            f"{base_url}api/{ORG_ID}/functions/test",
+            json={"function": vrl, "events": events},
+            timeout=60,
+        )
+        if last.status_code == 200:
+            results = last.json().get("results", [])
+            if results and all(r.get("event", {}).get("label") for r in results):
+                return last
+        time.sleep(3)
+    return last
+
+
+@pytest.fixture
+def enrichment_table(create_session, base_url):
+    # Deletes are asynchronous, so a per-second name collides with the previous
+    # test's teardown and the create is refused as "being deleted".
+    name = f"et_validation_{uuid.uuid4().hex[:12]}"
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/enrichment_tables/{name}",
+        files={"file": ("lookup.csv", ENRICHMENT_CSV, "text/csv")},
+        timeout=60,
+    )
+    assert resp.status_code == 200, f"table create failed: {resp.text}"
+    yield name
+    _delete_table(create_session, base_url, name)
+
+
+@pytest.mark.parametrize("blank", BLANK_NAMES)
+def test_file_upload_rejects_a_blank_table_name(create_session, base_url, blank):
+    """A whitespace-only name must be refused, not silently accepted (#2072)."""
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/enrichment_tables/{blank}",
+        files={"file": ("lookup.csv", ENRICHMENT_CSV, "text/csv")},
+        timeout=60,
+    )
+    assert resp.status_code == 400, f"blank name {blank!r} was accepted: {resp.text}"
+    assert "empty" in resp.text.lower(), f"error does not name the problem: {resp.text}"
+
+
+def test_url_upload_rejects_a_blank_table_name(create_session, base_url):
+    """The URL handler validates the name on its own path, so it is asserted too."""
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/enrichment_tables/%20%20/url",
+        json={"url": "https://example.com/lookup.csv"},
+        timeout=60,
+    )
+    assert resp.status_code == 400, f"blank name was accepted: {resp.text}"
+    assert "empty" in resp.text.lower(), f"error does not name the problem: {resp.text}"
+
+
+def test_a_valid_table_name_is_still_accepted(create_session, base_url, enrichment_table):
+    """Negative control: the validation must not reject ordinary names."""
+    resp = create_session.get(
+        f"{base_url}api/{ORG_ID}/streams?type=enrichment_tables", timeout=30
+    )
+    assert resp.status_code == 200
+    names = [s["name"] for s in resp.json().get("list", [])]
+    assert enrichment_table in names, f"{enrichment_table} missing from {names}"
+
+
+def test_vrl_lookup_substitutes_the_enrichment_value(
+    create_session, base_url, enrichment_table
+):
+    """The looked-up column must reach the event, not just be found (#5344)."""
+    vrl = (
+        f'rec, err = get_enrichment_table_record("{enrichment_table}", '
+        '{"code": to_string!(.code)})\n.label = rec.label\n.'
+    )
+    events = [{"code": "c1"}, {"code": "c2"}]
+
+    resp = _wait_for_lookup(create_session, base_url, vrl, events)
+
+    assert resp.status_code == 200, f"function test failed: {resp.text}"
+    labels = [r["event"].get("label") for r in resp.json()["results"]]
+    assert labels == ["ALPHA", "BETA"], f"enrichment not applied, got {labels}"
+
+
+def test_vrl_lookup_of_a_missing_key_leaves_the_event_unenriched(
+    create_session, base_url, enrichment_table
+):
+    """A key with no row must not fabricate a value (#5344)."""
+    vrl = (
+        f'rec, err = get_enrichment_table_record("{enrichment_table}", '
+        '{"code": to_string!(.code)})\n.label = rec.label\n.'
+    )
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/functions/test",
+        json={"function": vrl, "events": [{"code": "no-such-code"}]},
+        timeout=60,
+    )
+
+    assert resp.status_code == 200, f"function test failed: {resp.text}"
+    assert not resp.json()["results"][0]["event"].get("label")

tests/api-testing/tests/pipelines/test_realtime_pipeline_resave.py

@@ -0,0 +1,165 @@
+"""Re-saving a realtime pipeline must not duplicate events  [regression #13228].
+
+Saving an already-enabled realtime pipeline left the previous entry in the
+ingester's pipeline cache, so every subsequent record was processed once per
+save and the destination stream over-counted until the service was restarted.
+Fixed in PR #13186 by deduping the cache on save.
+
+The oracle is the destination row count after a known ingest, because the
+pipeline list looks identical either way -- the duplication lived in the cache,
+not in the stored definition.
+"""
+from __future__ import annotations
+
+import logging
+import time
+import uuid
+from collections.abc import Generator
+from typing import Any
+
+import pytest
+
+from support.client import OpenObserveClient
+from support.factories import unique_name
+
+logger = logging.getLogger(__name__)
+
+ORG_ID = "default"
+
+SAVE_COUNT = 3
+EVENT_COUNT = 10
+ROUTING_TIMEOUT = 120
+
+
+def _realtime_payload(name: str, source_stream: str, destination_stream: str) -> dict[str, Any]:
+    input_id = str(uuid.uuid4())
+    output_id = str(uuid.uuid4())
+    return {
+        "name": name,
+        "description": "",
+        "source": {"source_type": "realtime"},
+        "nodes": [
+            {
+                "id": input_id,
+                "type": "input",
+                "data": {
+                    "node_type": "stream",
+                    "stream_name": source_stream,
+                    "stream_type": "logs",
+                    "org_id": ORG_ID,
+                },
+                "position": {"x": 100, "y": 100},
+                "io_type": "input",
+            },
+            {
+                "id": output_id,
+                "type": "output",
+                "data": {
+                    "node_type": "stream",
+                    "stream_name": destination_stream,
+                    "stream_type": "logs",
+                    "org_id": ORG_ID,
+                },
+                "position": {"x": 300, "y": 100},
+                "io_type": "output",
+            },
+        ],
+        "edges": [{"id": f"e-{input_id}-{output_id}", "source": input_id, "target": output_id}],
+        "org": ORG_ID,
+    }
+
+
+def _find_pipeline(client: OpenObserveClient, name: str) -> dict[str, Any] | None:
+    pipelines = client.get("pipelines").json().get("list", [])
+    return next((p for p in pipelines if p["name"] == name.lower()), None)
+
+
+def _count_rows(client: OpenObserveClient, stream: str) -> int:
+    now = int(time.time() * 1_000_000)
+    resp = client.post(
+        "_search?type=logs&use_cache=false",
+        json={
+            "query": {
+                "sql": f'SELECT COUNT(*) AS cnt FROM "{stream}"',
+                "start_time": now - 3_600_000_000,
+                "end_time": now + 3_600_000_000,
+                "size": -1,
+            }
+        },
+    )
+    if resp.status_code != 200:
+        return 0
+    hits = resp.json().get("hits") or []
+    return hits[0].get("cnt", 0) if hits else 0
+
+
+def _wait_for_rows(client: OpenObserveClient, stream: str, expected: int) -> int:
+    """Settle on a count, then hold to catch a late duplicate arriving after it."""
+    deadline = time.time() + ROUTING_TIMEOUT
+    while time.time() < deadline:
+        if _count_rows(client, stream) >= expected:
+            break
+        time.sleep(3)
+    time.sleep(10)
+    return _count_rows(client, stream)
+
+
+@pytest.fixture
+def resaved_pipeline(client: OpenObserveClient) -> Generator[tuple[str, str], None, None]:
+    """A realtime pipeline saved SAVE_COUNT times, as the bug required."""
+    name = unique_name("pyt_13228").lower()
+    source = f"{name}_src"
+    destination = f"{name}_dst"
+
+    seed = client.post(f"{source}/_json", json=[{"level": "info", "msg": "seed"}])
+    assert seed.status_code == 200, f"seed ingest failed: {seed.text}"
+
+    resp = client.post("pipelines", json=_realtime_payload(name, source, destination))
+    assert resp.status_code == 200, f"create failed: {resp.text}"
+
+    pipeline = _find_pipeline(client, name)
+    assert pipeline is not None, f"created pipeline {name} not in list"
+
+    # Saves are guarded by `version`, so each one re-reads the stored pipeline.
+    for attempt in range(SAVE_COUNT):
+        current = _find_pipeline(client, name)
+        assert current is not None, f"pipeline {name} vanished before re-save {attempt}"
+
+        payload = _realtime_payload(name, source, destination)
+        payload["pipeline_id"] = current["pipeline_id"]
+        payload["id"] = current["pipeline_id"]
+        payload["version"] = current["version"]
+        payload["enabled"] = current["enabled"]
+
+        update = client.put("pipelines", json=payload)
+        assert update.status_code == 200, f"re-save {attempt} failed: {update.text}"
+        time.sleep(2)
+
+    yield source, destination
+
+    found = _find_pipeline(client, name)
+    if found is not None:
+        client.delete(f"pipelines/{found['pipeline_id']}")
+    for stream in (source, destination):
+        client.delete(f"streams/{stream}?type=logs")
+
+
+def test_resaving_a_realtime_pipeline_routes_each_event_once(
+    client: OpenObserveClient, resaved_pipeline
+):
+    """Each ingested record must land once, however many times the pipeline was saved."""
+    source, destination = resaved_pipeline
+    now = int(time.time() * 1_000_000)
+    records = [
+        {"_timestamp": now + i, "seq": i, "msg": "routed"} for i in range(EVENT_COUNT)
+    ]
+
+    resp = client.post(f"{source}/_json", json=records)
+    assert resp.status_code == 200, f"ingest failed: {resp.text}"
+
+    delivered = _wait_for_rows(client, destination, EVENT_COUNT)
+
+    assert delivered == EVENT_COUNT, (
+        f"expected {EVENT_COUNT} routed records after {SAVE_COUNT} saves, "
+        f"got {delivered} -- a multiple means the pipeline cache kept a stale entry"
+    )

tests/api-testing/tests/search/test_streaming_order_and_values.py

@@ -0,0 +1,157 @@
+"""Streaming result order and field values  [regressions #7790, #7308].
+
+#7790 (PR #7787): under http2 streaming the new histogram flow returned hits
+out of order, so the chart plotted against a sequence the rows did not have.
+The order is asserted against an ingested sequence column, not just against
+`_timestamp`, so a stable-but-wrong order still fails.
+
+#7308: `_values` answered 400 for some fields reachable from the logs sidebar,
+which left the field list with no values to offer.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import time
+import uuid
+
+import pytest
+
+from support.sse import read_sse_response
+
+logger = logging.getLogger(__name__)
+
+ORG_ID = os.environ.get("TEST_ORG_ID", "default")
+
+ROW_COUNT = 25
+INGEST_SETTLE_TIMEOUT = 120
+
+
+def _now_us() -> int:
+    return int(time.time() * 1_000_000)
+
+
+@pytest.fixture(scope="module")
+def ordered_stream(create_session, base_url):
+    """A stream whose rows carry a known sequence, so order is checkable."""
+    stream = f"order_regression_{uuid.uuid4().hex[:12]}"
+    now = _now_us()
+    records = [
+        {"_timestamp": now - i * 1_000_000, "seq": i, "body_code": f"c{i % 3}"}
+        for i in range(ROW_COUNT)
+    ]
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/{stream}/_json", json=records, timeout=60
+    )
+    assert resp.status_code == 200, f"ingest failed: {resp.text}"
+
+    deadline = time.time() + INGEST_SETTLE_TIMEOUT
+    while time.time() < deadline:
+        r = create_session.post(
+            f"{base_url}api/{ORG_ID}/_search?type=logs&use_cache=false",
+            json={
+                "query": {
+                    "sql": f'SELECT COUNT(*) AS cnt FROM "{stream}"',
+                    "start_time": now - 7_200_000_000,
+                    "end_time": now + 3_600_000_000,
+                    "size": -1,
+                }
+            },
+            timeout=60,
+        )
+        hits = (r.json() or {}).get("hits") or [] if r.status_code == 200 else []
+        if hits and hits[0].get("cnt", 0) >= ROW_COUNT:
+            break
+        time.sleep(2)
+    else:
+        pytest.fail(f"{stream} never reached {ROW_COUNT} rows")
+
+    yield stream, now
+    create_session.delete(
+        f"{base_url}api/{ORG_ID}/streams/{stream}?type=logs", timeout=30
+    )
+
+
+def test_streaming_search_returns_hits_in_the_requested_order(
+    create_session, base_url, ordered_stream
+):
+    """`ORDER BY _timestamp DESC` over the streaming path must arrive sorted (#7790)."""
+    stream, now = ordered_stream
+    payload = {
+        "query": {
+            "sql": f'SELECT _timestamp, seq FROM "{stream}" ORDER BY _timestamp DESC',
+            "start_time": now - 7_200_000_000,
+            "end_time": now + 3_600_000_000,
+            "from": 0,
+            "size": ROW_COUNT,
+        }
+    }
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/_search_stream?type=logs&use_cache=false",
+        json=payload,
+        headers={"Accept": "text/event-stream"},
+        stream=True,
+        timeout=120,
+    )
+    assert resp.status_code == 200, f"streaming search failed: {resp.text[:300]}"
+
+    hits = read_sse_response(resp)["results"]["hits"]
+    assert len(hits) == ROW_COUNT, f"expected {ROW_COUNT} hits, got {len(hits)}"
+
+    timestamps = [h["_timestamp"] for h in hits]
+    assert timestamps == sorted(timestamps, reverse=True), "hits are not ordered"
+
+    # The ingested sequence descends with _timestamp, so it must arrive ascending.
+    assert [h["seq"] for h in hits] == list(range(ROW_COUNT))
+
+
+def test_streaming_search_honours_ascending_order(
+    create_session, base_url, ordered_stream
+):
+    """The mirror direction, so a hardcoded DESC cannot pass this file."""
+    stream, now = ordered_stream
+    payload = {
+        "query": {
+            "sql": f'SELECT _timestamp, seq FROM "{stream}" ORDER BY _timestamp ASC',
+            "start_time": now - 7_200_000_000,
+            "end_time": now + 3_600_000_000,
+            "from": 0,
+            "size": ROW_COUNT,
+        }
+    }
+    resp = create_session.post(
+        f"{base_url}api/{ORG_ID}/_search_stream?type=logs&use_cache=false",
+        json=payload,
+        headers={"Accept": "text/event-stream"},
+        stream=True,
+        timeout=120,
+    )
+    assert resp.status_code == 200, f"streaming search failed: {resp.text[:300]}"
+
+    hits = read_sse_response(resp)["results"]["hits"]
+    assert [h["seq"] for h in hits] == list(range(ROW_COUNT - 1, -1, -1))
+
+
+def test_values_returns_field_values_without_a_bad_request(
+    create_session, base_url, ordered_stream
+):
+    """A sidebar field must answer with its values, not 400 (#7308)."""
+    stream, now = ordered_stream
+    resp = create_session.get(
+        f"{base_url}api/{ORG_ID}/{stream}/_values",
+        params={
+            "fields": "body_code",
+            "size": 10,
+            "start_time": now - 7_200_000_000,
+            "end_time": now + 3_600_000_000,
+            "type": "logs",
+        },
+        timeout=60,
+    )
+
+    assert resp.status_code == 200, f"_values failed: {resp.text[:300]}"
+    hits = resp.json().get("hits", [])
+    assert hits, f"no field values returned: {resp.text[:300]}"
+
+    values = {v["zo_sql_key"] for v in hits[0]["values"]}
+    assert values == {"c0", "c1", "c2"}, f"unexpected values: {values}"

tests/ui-testing/ci-matrix/ci_matrix_regression.json

@@ -9,7 +9,9 @@
       "logs-result-display.spec.js",
       "logs-nonsql-limit.spec.js",
       "logs-stream-type-scope.spec.js",
-      "logs-14376-int64-precision.spec.js"
+      "logs-14376-int64-precision.spec.js",
+      "logs-stream-selector.spec.js",
+      "logs-table-stream-switch.spec.js"
     ],
     "disabled": []
   },

tests/ui-testing/pages/generalPages/enrichmentPage.js

@@ -1,6 +1,9 @@
 const { expect } = require('@playwright/test');
 const testLogger = require('../../playwright-tests/utils/test-logger.js');
 const { isCloudEnvironment } = require('../cloudPages/cloud-env.js');
+
+// The dropdown trigger intermittently drops the opening click, so it is re-tried.
+const HELP_MENU_OPEN_ATTEMPTS = 3;
 import { openNavFlyoutChild } from '../commonActions.js';
 
 class EnrichmentPage {
@@ -246,7 +249,23 @@ class EnrichmentPage {
      */
     async clickHelpMenuItem() {
         await this.helpMenuItem.waitFor({ state: 'visible', timeout: 10000 });
-        await this.helpMenuItem.click();
+
+        // The trigger intermittently swallows the first click and stays closed, which
+        // then reads as the menu's items being missing rather than never rendered.
+        // `data-state` is the dropdown's own signal, so it separates the two.
+        for (let attempt = 1; attempt <= HELP_MENU_OPEN_ATTEMPTS; attempt++) {
+            await this.helpMenuItem.click();
+            try {
+                await expect(this.helpMenuItem)
+                    .toHaveAttribute('data-state', 'open', { timeout: 3000 });
+                return;
+            } catch (e) {
+                testLogger.warn(`clickHelpMenuItem: menu still closed after attempt ${attempt}`);
+            }
+        }
+        throw new Error(
+            `help menu did not open after ${HELP_MENU_OPEN_ATTEMPTS} attempts`
+        );
     }
 
     /**

tests/ui-testing/pages/logsPages/logsPage.js

@@ -13016,4 +13016,141 @@ export class LogsPage {
         return values;
     }
 
+    /**
+     * Poll the search API until a seeded window is fully searchable.
+     *
+     * Ingest acknowledges before every row is indexed, so a page loaded straight
+     * after seeding queries an arbitrary prefix. Settling over the API first costs
+     * one request per poll, where settling through the UI re-runs the whole render.
+     */
+    async waitForSearchableRowCount(stream, expected, startTimeMicros, endTimeMicros, timeout = 90000) {
+        const orgId = getOrgIdentifier();
+        const url = `${process.env.ZO_BASE_URL}/api/${orgId}/_search?type=logs&use_cache=false`;
+        let last = null;
+        await expect.poll(async () => {
+            const resp = await this.page.request.post(url, {
+                headers: getAuthHeaders(),
+                data: {
+                    query: {
+                        sql: `SELECT COUNT(*) AS cnt FROM "${stream}"`,
+                        start_time: startTimeMicros,
+                        end_time: endTimeMicros,
+                        size: 1,
+                    },
+                },
+            });
+            if (!resp.ok()) return -1;
+            const body = await resp.json().catch(() => ({}));
+            // Coerced: a 64-bit count can serialise as a string, which no strict compare matches.
+            last = Number(body?.hits?.[0]?.cnt ?? 0);
+            return last;
+        }, { timeout, intervals: [1000, 2000, 3000] }).toBe(expected);
+        return last;
+    }
+
+    getWrapContentButton() {
+        return this.page.locator('[data-test="logs-search-result-wrap-table-content-btn"]');
+    }
+
+    /** Wrap is a persisted preference, so localStorage is the state -- not the button's styling.
+     *  The key is absent until the user first toggles, which reads as off. */
+    async isWrapContentOn() {
+        return await this.page.evaluate(() => {
+            try {
+                return localStorage.getItem('wrapContent') === 'true';
+            } catch (e) {
+                return false;
+            }
+        });
+    }
+
+    async setWrapContentOff() {
+        await this.getWrapContentButton().waitFor({ state: 'visible', timeout: 15000 });
+        if (await this.isWrapContentOn()) {
+            await this.getWrapContentButton().click();
+        }
+        await expect
+            .poll(async () => await this.isWrapContentOn(), { timeout: 10000 })
+            .toBe(false);
+    }
+
+    async expectWrapContentOn(expected) {
+        await expect
+            .poll(async () => await this.isWrapContentOn(), { timeout: 15000 })
+            .toBe(expected);
+    }
+
+    async expectHistogramChartVisible() {
+        await expect(
+            this.page.locator('[data-test="logs-search-result-bar-chart"]')
+        ).toBeVisible({ timeout: 30000 });
+    }
+
+    /**
+     * Land on the stream explorer. The explore button on the streams page produces
+     * exactly this URL, and going straight to it avoids picking a row out of a list
+     * whose filter matches substrings -- `e2e_automate` also lists `e2e_automate_w0`
+     * and friends, and which of those a click lands on differs per environment.
+     */
+    async exploreStreamFromStreamsPage(streamName) {
+        const orgId = getOrgIdentifier();
+        const url = `${process.env.ZO_BASE_URL}/web/logs`
+            + `?stream_type=logs&stream=${streamName}&period=15m&refresh=0`
+            + `&type=stream_explorer&org_identifier=${orgId}&show_histogram=true`;
+        await this.page.goto(url, { waitUntil: 'domcontentloaded', timeout: 30000 });
+        await this.page.waitForURL(/type=stream_explorer/, { timeout: 30000 });
+    }
+
+    getStreamPickerTrigger() {
+        return this.page.locator(this.indexDropDownTrigger);
+    }
+
+    getStreamPickerPopover() {
+        return this.page.locator(this.indexDropDownPopover);
+    }
+
+    getStreamPickerOption(streamName) {
+        return this.page.locator(
+            `[data-test="log-search-index-list-select-stream-option"][data-test-value="${streamName}"]`
+        );
+    }
+
+    /** Open the picker, type a filter, and click the option that survives it. */
+    async pickStreamByTypedFilter(filterText, streamName) {
+        await this.getStreamPickerTrigger().click();
+        await this.getStreamPickerPopover().waitFor({ state: 'visible', timeout: 15000 });
+        await this.page.locator(this.indexDropDownSearch).fill(filterText);
+        const option = this.getStreamPickerOption(streamName);
+        await option.waitFor({ state: 'visible', timeout: 15000 });
+        await option.click();
+    }
+
+    async expectStreamPickerClosed() {
+        await expect(this.getStreamPickerPopover()).toBeHidden({ timeout: 15000 });
+    }
+
+    /** The trigger renders the selected name alone -- a typed filter must not survive in it. */
+    async expectSelectedStreamLabel(streamName) {
+        const trigger = this.getStreamPickerTrigger();
+        await expect(trigger).toHaveAttribute('data-test-selected-label', streamName, { timeout: 15000 });
+        await expect(trigger).toHaveText(streamName, { timeout: 15000 });
+    }
+
+    async openStreamPicker() {
+        await this.getStreamPickerTrigger().click();
+        await this.getStreamPickerPopover().waitFor({ state: 'visible', timeout: 15000 });
+    }
+
+    /** Every offered stream must render its name; a blank row is the #10598 symptom. */
+    async getStreamOptionLabels() {
+        const options = this.page.locator('[data-test="log-search-index-list-select-stream-option"]');
+        await options.first().waitFor({ state: 'visible', timeout: 15000 });
+        return await options.evaluateAll((nodes) =>
+            nodes.map((n) => ({
+                value: n.getAttribute('data-test-value'),
+                text: (n.textContent || '').trim(),
+            }))
+        );
+    }
+
 }

tests/ui-testing/pages/pipelinesPages/pipelinesPage.js

@@ -4430,4 +4430,13 @@ export class PipelinesPage {
         return await this.pipelineNodeOutputStreamNode.count();
     }
 
+    /**
+     * Saving the source node adds the mirroring destination asynchronously, so a
+     * plain count races the render -- it won on a fast machine and lost under CI load.
+     */
+    async expectOutputStreamNodePresent() {
+        await expect(this.pipelineNodeOutputStreamNode.first())
+            .toBeVisible({ timeout: 30000 });
+    }
+
 }

tests/ui-testing/playwright-tests/RegressionSet/Alerts/alerts-14796-destination-usage-ent.spec.js

@@ -105,6 +105,53 @@ test.describe('Enterprise consumers of the destination delete guard testcases',
     await pm.alertDestinationsPage.searchDestinations(name);
   }
 
+  /**
+   * A pipeline-module destination plus a realtime pipeline whose `remote_stream`
+   * node names it. Pipeline destinations carry no template and never appear on the
+   * alert destinations list, so this is the only way to reach that arm.
+   */
+  async function seedPipelineOnDestination(page, prefix) {
+    const { v1, org } = urls();
+    const destinationName = await seedDestination(page, prefix, 'pipeline');
+
+    const pipelineName = uniq(`${prefix}_pipeline`);
+    const stream = 'alerts_p0_stream';
+    const pipeRes = await api(page, 'post', `${v1}/pipelines`, {
+      name: pipelineName, description: '', org, enabled: true,
+      source: { source_type: 'realtime', org_id: org, stream_name: stream, stream_type: 'logs' },
+      nodes: [
+        {
+          id: 'n1', type: 'input', io_type: 'input', position: { x: 0, y: 0 },
+          data: { node_type: 'stream', org_id: org, stream_name: stream, stream_type: 'logs' },
+        },
+        {
+          id: 'n2', type: 'output', io_type: 'output', position: { x: 200, y: 200 },
+          data: { node_type: 'remote_stream', org_id: org, destination_name: destinationName },
+        },
+      ],
+      edges: [{ id: 'e1', source: 'n1', target: 'n2' }],
+    });
+    expect(pipeRes.status(), await pipeRes.text()).toBe(200);
+    createdConsumers.push(`${v1}/pipelines/${(await pipeRes.json()).id}`);
+    return { destinationName, pipelineName };
+  }
+
+  test('should refuse deleting a pipeline destination a pipeline still routes to', {
+    tag: ['@alert-destination-pipeline-consumer', '@enterprise', '@alerts', '@P0'],
+  }, async ({ page }) => {
+    testLogger.info('Seeding a pipeline destination and a pipeline that routes to it');
+    const { destinationName, pipelineName } = await seedPipelineOnDestination(page, 'e2e_14796_pipe');
+
+    // Pipeline destinations live on their own Settings tab, which is enterprise-only.
+    await pm.pipelinesPage.openPipelineDestinationsAt(destinationName);
+
+    testLogger.info('Asserting the delete is refused and names the pipeline');
+    await pm.alertDestinationsPage.attemptDeleteDestination(destinationName);
+    await pm.alertDestinationsPage.expectInUseErrorToastContaining(['pipeline', pipelineName]);
+    await pm.alertDestinationsPage.expectDestinationRowStillVisible(destinationName);
+    testLogger.info('Test completed');
+  });
+
   test('should count an anomaly detection config and refuse its destination delete', {
     tag: ['@alert-destination-anomaly-consumer', '@enterprise', '@alerts', '@P0'],
   }, async ({ page }) => {

tests/ui-testing/playwright-tests/RegressionSet/Alerts/alerts-14796-destination-usage.spec.js

@@ -25,6 +25,10 @@
  * produced the original report. Synthetics is off in CI (neither workflow sets
  * `ZO_SYNTHETICS_ENABLED`) and its locations have no env-based seeding, so a check
  * cannot be created. Enabling it is a workflow change, not a spec change.
+ *
+ * The pipeline consumer lives in `alerts-14796-destination-usage-ent.spec.js`: it is
+ * only reachable from Settings -> Pipeline Destinations, a tab rendered when the build
+ * is enterprise, and the regression suite builds opensource.
  */
 
 const { test, expect, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
@@ -46,7 +50,6 @@ test.describe('Alert destination delete guard across consumers testcases', {
   let createdAlerts = [];
   let createdDestinations = [];
   let createdTemplates = [];
-  let createdPipelines = [];
 
   test.beforeEach(async ({ page }, testInfo) => {
     testLogger.testStart(testInfo.title, testInfo.file);
@@ -54,7 +57,6 @@ test.describe('Alert destination delete guard across consumers testcases', {
     createdAlerts = [];
     createdDestinations = [];
     createdTemplates = [];
-    createdPipelines = [];
     await seedAlertFixtures(page);
     await navigateToBase(page);
   });
@@ -66,9 +68,6 @@ test.describe('Alert destination delete guard across consumers testcases', {
     for (const id of [...createdAlerts].reverse()) {
       await deleteAlertInFolder(page, id, 'default');
     }
-    for (const id of createdPipelines) {
-      await api(page, 'delete', `${v1}/pipelines/${id}`).catch(() => {});
-    }
     for (const name of createdDestinations) {
       await api(page, 'delete', `${v1}/alerts/destinations/${name}`).catch(() => {});
     }
@@ -123,42 +122,6 @@ test.describe('Alert destination delete guard across consumers testcases', {
     return { id, name };
   }
 
-  /**
-   * A pipeline-module destination (`?module=pipeline`) plus a realtime pipeline whose
-   * `remote_stream` node names it. Pipeline destinations carry no template and never
-   * appear on the alert destinations list, so this is the only way to reach that arm.
-   */
-  async function seedPipelineOnDestination(page, prefix) {
-    const { v1, org } = urls();
-    const destinationName = uniq(`${prefix}_dest`);
-    const destRes = await api(page, 'post', `${v1}/alerts/destinations?module=pipeline`, {
-      name: destinationName, url: 'http://127.0.0.1:1/never-called', method: 'post', type: 'http',
-    });
-    expect(destRes.status(), await destRes.text()).toBe(200);
-    createdDestinations.push(destinationName);
-
-    const pipelineName = uniq(`${prefix}_pipeline`);
-    const stream = 'alerts_p0_stream';
-    const pipeRes = await api(page, 'post', `${v1}/pipelines`, {
-      name: pipelineName, description: '', org, enabled: true,
-      source: { source_type: 'realtime', org_id: org, stream_name: stream, stream_type: 'logs' },
-      nodes: [
-        {
-          id: 'n1', type: 'input', io_type: 'input', position: { x: 0, y: 0 },
-          data: { node_type: 'stream', org_id: org, stream_name: stream, stream_type: 'logs' },
-        },
-        {
-          id: 'n2', type: 'output', io_type: 'output', position: { x: 200, y: 200 },
-          data: { node_type: 'remote_stream', org_id: org, destination_name: destinationName },
-        },
-      ],
-      edges: [{ id: 'e1', source: 'n1', target: 'n2' }],
-    });
-    expect(pipeRes.status(), await pipeRes.text()).toBe(200);
-    createdPipelines.push((await pipeRes.json()).id);
-    return { destinationName, pipelineName };
-  }
-
   async function openDestinationsScopedTo(name) {
     await pm.alertDestinationsPage.navigateToDestinations();
     await pm.alertDestinationsPage.waitForDestinationListReady();
@@ -221,22 +184,6 @@ test.describe('Alert destination delete guard across consumers testcases', {
     testLogger.info('Test completed');
   });
 
-  test('should refuse deleting a pipeline destination a pipeline still routes to', {
-    tag: ['@alert-destination-pipeline-consumer', '@all', '@alerts', '@P0'],
-  }, async ({ page }) => {
-    testLogger.info('Seeding a pipeline destination and a pipeline that routes to it');
-    const { destinationName, pipelineName } = await seedPipelineOnDestination(page, 'e2e_14796_pipe');
-
-    // Pipeline destinations live on their own Settings tab, not the alerts list.
-    await pm.pipelinesPage.openPipelineDestinationsAt(destinationName);
-
-    testLogger.info('Asserting the delete is refused and names the pipeline');
-    await pm.alertDestinationsPage.attemptDeleteDestination(destinationName);
-    await pm.alertDestinationsPage.expectInUseErrorToastContaining(['pipeline', pipelineName]);
-    await pm.alertDestinationsPage.expectDestinationRowStillVisible(destinationName);
-    testLogger.info('Test completed');
-  });
-
   test('should still delete a destination no consumer references', {
     tag: ['@alert-destination-unused-delete', '@all', '@alerts', '@P0'],
   }, async ({ page }) => {

tests/ui-testing/playwright-tests/RegressionSet/General/ui-regression.spec.js

@@ -69,7 +69,6 @@ test.describe("UI Regression Bugs", () => {
 
     // Click Help menu - using POM method
     await pm.enrichmentPage.clickHelpMenuItem();
-    await page.waitForTimeout(1000);
 
     testLogger.info('Help menu opened');
 

tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-result-display.spec.js

@@ -73,6 +73,17 @@ test.describe("Logs Result Display Regression", () => {
     await sendRequest(page, getIngestionUrl(orgId, denseStream), rows, getHeaders());
 
     await pm.logsPage.waitForStreamAvailable(denseStream);
+
+    // Ingest acks before every row is searchable. Settling over the API before the
+    // page loads means the banner is read once the window is complete, rather than
+    // re-running the query in the UI and hoping the index catches up first.
+    await pm.logsPage.waitForSearchableRowCount(
+      denseStream,
+      DENSE_ROWS,
+      secondStartMs * 1000,
+      (secondStartMs + 1000) * 1000,
+    );
+
     await page.goto(
       `${logData.logsUrl}?org_identifier=${orgId}&stream=${denseStream}&stream_type=logs` +
       `&from=${secondStartMs}&to=${secondStartMs + 1000}`,
@@ -80,9 +91,6 @@ test.describe("Logs Result Display Regression", () => {
     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}`);

tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-stream-selector.spec.js

@@ -0,0 +1,43 @@
+const { test, expect, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
+const testLogger = require('../../utils/test-logger.js');
+const PageManager = require('../../../pages/page-manager.js');
+
+const STREAM = 'e2e_automate';
+// A prefix of the stream name, so the filter and the selection are different strings.
+const TYPED_PREFIX = 'e2e_a';
+
+test.describe("Logs stream picker", () => {
+  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 typed filter does not survive into the selected label @bug-9540 @bug-8456 @P0 @regression @logsRegression", {
+    tag: ['@bug-9540', '@bug-8456', '@P0', '@regression', '@logsRegression'],
+  }, async () => {
+    await pm.logsPage.navigateToLogs();
+    await pm.logsPage.pickStreamByTypedFilter(TYPED_PREFIX, STREAM);
+
+    await pm.logsPage.expectStreamPickerClosed();
+    await pm.logsPage.expectSelectedStreamLabel(STREAM);
+  });
+
+  test("every offered stream renders its name after a query has run @bug-10598 @P2 @regression @logsRegression", {
+    tag: ['@bug-10598', '@P2', '@regression', '@logsRegression'],
+  }, async () => {
+    await pm.logsPage.selectStream(STREAM);
+    await pm.logsPage.selectRunQuery();
+
+    await pm.logsPage.openStreamPicker();
+    const options = await pm.logsPage.getStreamOptionLabels();
+
+    expect(options.length).toBeGreaterThan(0);
+    const blank = options.filter((o) => o.text.length === 0);
+    expect(blank, `streams rendered with no label: ${JSON.stringify(blank)}`).toEqual([]);
+    expect(options[0].text).toBe(options[0].value);
+  });
+});

tests/ui-testing/playwright-tests/RegressionSet/Logs/logs-table-stream-switch.spec.js

@@ -0,0 +1,39 @@
+const { test, navigateToBase } = require('../../utils/enhanced-baseFixtures.js');
+const testLogger = require('../../utils/test-logger.js');
+const PageManager = require('../../../pages/page-manager.js');
+
+const STREAM = 'e2e_automate';
+
+test.describe("Logs table across a stream switch", () => {
+  let pm;
+
+  test.beforeEach(async ({ page }, testInfo) => {
+    testLogger.testStart(testInfo.title, testInfo.file);
+    await navigateToBase(page);
+    pm = new PageManager(page);
+  });
+
+  test("navigating from a stream explorer back to Logs leaves wrap and the histogram alone", {
+    tag: ['@bug-7333', '@bug-7460', '@P2', '@regression', '@logsRegression'],
+  }, async () => {
+    await pm.logsPage.selectStream(STREAM);
+    await pm.logsPage.selectRunQuery();
+
+    // Wrap is a persisted preference, so the invariant is that navigation does
+    // not flip it -- not that it arrives off.
+    await pm.logsPage.setWrapContentOff();
+
+    await pm.logsPage.exploreStreamFromStreamsPage(STREAM);
+
+    // The menu hop out of the explorer is what runs the watcher the fix changed;
+    // stopping at the explorer never reaches it.
+    await pm.logsPage.navigateToLogsFromSidebar();
+
+    // Asserting wrap before the page has rendered reads back the value the test
+    // set itself, so the results have to land first.
+    await pm.logsPage.expectHistogramChartVisible();
+    await pm.logsPage.expectLogsTableVisible();
+
+    await pm.logsPage.expectWrapContentOn(false);
+  });
+});

tests/ui-testing/playwright-tests/RegressionSet/Pipelines/pipeline-default-destination-guard.spec.js

@@ -1,4 +1,4 @@
-const { test, expect } = require('../../utils/enhanced-baseFixtures.js');
+const { test } = require('../../utils/enhanced-baseFixtures.js');
 const testLogger = require('../../utils/test-logger.js');
 const PageManager = require('../../../pages/page-manager.js');
 const logsdata = require("../../../../test-data/logs_data.json");
@@ -27,14 +27,14 @@ test.describe("Pipeline default destination node guard", () => {
     tag: ['@bug-8597', '@P2', '@regression', '@pipelinesRegression']
   }, async () => {
     // Saving the source node implicitly adds a destination mirroring it; that is the guarded node.
-    expect(await pm.pipelinesPage.countOutputStreamNodes()).toBeGreaterThan(0);
+    await pm.pipelinesPage.expectOutputStreamNodePresent();
 
     await pm.pipelinesPage.openOutputStreamNodeDeleteDialog();
     await pm.pipelinesPage.expectDefaultDestinationWarningVisible();
     await pm.pipelinesPage.cancelConfirmDialog();
 
     // Cancelling must leave the node in place, or the warning is decoration.
-    expect(await pm.pipelinesPage.countOutputStreamNodes()).toBeGreaterThan(0);
+    await pm.pipelinesPage.expectOutputStreamNodePresent();
 
     testLogger.info('✓ PASSED: implicit destination node delete is guarded (Bug #8597)');
   });