Files
openobserve/web/src/composables/useStreams.spec.ts
T

1200 lines
42 KiB
TypeScript

// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
import { describe, expect, it, vi, beforeEach, afterEach } from "vitest";
import { CancelledError } from "@tanstack/vue-query";
import useStreams from "@/composables/useStreams";
import StreamService from "@/services/stream";
import { queryClient } from "@/composables/query/queryClient";
import { toast } from "@/lib/feedback/Toast/useToast";
import i18nInstance from "@/locales";
const t = (i18nInstance.global as any).t;
// Mock Stream Service
vi.mock("@/services/stream", async (importOriginal) => {
const { overlayServiceMock } = await import("@/test/unit/helpers/mockService");
return overlayServiceMock(await importOriginal(), {
default: {
nameList: vi.fn(),
schema: vi.fn(),
},
});
});
// Mock Toast
const mockToastDismiss = vi.fn();
vi.mock("@/lib/feedback/Toast/useToast", () => ({
toast: vi.fn(),
}));
// Mock utilities
vi.mock("@/utils/zincutils", () => ({
deepCopy: vi.fn((obj) => JSON.parse(JSON.stringify(obj))),
}));
// Mock Store
const createMockStore = () => ({
state: {
selectedOrganization: {
identifier: "test-org",
},
streams: {
logs: null,
metrics: null,
traces: null,
enrichment_tables: null,
index: null,
metadata: null,
streamsIndexMapping: {
logs: {},
metrics: {},
traces: {},
enrichment_tables: {},
index: {},
metadata: {},
},
areStreamsFetched: false,
},
organizationData: {
isDataIngested: false,
},
},
dispatch: vi.fn(),
commit: vi.fn(),
});
let mockStore: ReturnType<typeof createMockStore>;
vi.mock("vuex", () => ({
useStore: () => mockStore,
}));
describe("useStreams Composable", () => {
let streamsInstance: ReturnType<typeof useStreams>;
beforeEach(() => {
vi.clearAllMocks();
vi.mocked(toast).mockReturnValue(mockToastDismiss);
mockStore = createMockStore();
// Set up default mocks
const mockStreamService = vi.mocked(StreamService);
mockStreamService.nameList.mockResolvedValue({
data: {
list: [{ name: "test-stream", stream_type: "logs" }],
},
});
mockStreamService.schema.mockResolvedValue({
data: {
name: "test-stream",
stream_type: "logs",
schema: [
{ name: "field1", type: "string" },
{ name: "_o2_id", type: "string" },
{ name: "_original", type: "string" },
{ name: "_all_values", type: "string" },
],
},
});
streamsInstance = useStreams(t);
});
afterEach(() => {
vi.clearAllTimers();
});
// Test 1: Composable Initialization
describe("Composable Initialization", () => {
it("should initialize useStreams composable with all methods", () => {
expect(streamsInstance).toBeDefined();
const expectedMethods = [
"getStreams",
"getStream",
"setStreams",
"getMultiStreams",
"resetStreams",
"removeStream",
"addStream",
"getUpdatedSettings",
"resetStreamType",
"getPaginatedStreams",
"isStreamExists",
"isStreamFetched",
"addNewStreams",
"updateStreamsInStore",
"updateStreamIndexMappingInStore",
"updateStreamsFetchedInStore",
"getAllStreamsPayload",
"removeSchemaFields",
"getStreamPayload",
"compareArrays",
"deepEqual",
"comparePatternAssociations",
];
expectedMethods.forEach((method) => {
expect(streamsInstance).toHaveProperty(method);
expect(typeof streamsInstance[method]).toBe("function");
});
});
});
// Test 2-5: Store Update Functions
describe("Store Update Functions", () => {
it("should call store dispatch for updateStreamsInStore", () => {
const streamType = "logs";
const streams = { list: [] };
streamsInstance.updateStreamsInStore(streamType, streams);
expect(mockStore.dispatch).toHaveBeenCalledWith("streams/setStreams", {
streamType,
streams,
});
});
it("should call store commit for updateStreamIndexMappingInStore", () => {
const indexMapping = { logs: { test: 0 } };
streamsInstance.updateStreamIndexMappingInStore(indexMapping);
expect(mockStore.commit).toHaveBeenCalledWith(
"streams/updateStreamIndexMapping",
indexMapping,
);
});
it("should call store commit for updateStreamsFetchedInStore", () => {
const areStreamsFetched = true;
streamsInstance.updateStreamsFetchedInStore(areStreamsFetched);
expect(mockStore.commit).toHaveBeenCalledWith(
"streams/updateStreamsFetched",
areStreamsFetched,
);
});
it("should return stream payload structure from getStreamPayload", () => {
const payload = streamsInstance.getStreamPayload();
expect(payload).toEqual({
name: "",
list: [],
schema: false,
});
});
});
// Test 6-10: Stream Fetching Functions
describe("Stream Fetching Functions", () => {
it("should get streams for specific stream type", async () => {
mockStore.state.streams.logs = null; // Not fetched yet
const result = await streamsInstance.getStreams("logs", false, false);
expect(StreamService.nameList).toHaveBeenCalledWith("test-org", "logs", false);
expect(result).toHaveProperty("name", "logs");
expect(result).toHaveProperty("list");
});
it("caches a profiles stream list like any other type", async () => {
vi.mocked(StreamService.nameList).mockResolvedValueOnce({
data: { list: [{ name: "prof1", stream_type: "profiles" }] },
} as any);
const result = await streamsInstance.getStreams("profiles", false, false);
expect(result).toHaveProperty("list", [{ name: "prof1", stream_type: "profiles" }]);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({ streamType: "profiles" }),
);
});
it("should get all streams when streamName is 'all'", async () => {
mockStore.state.streams.areStreamsFetched = false;
const result = await streamsInstance.getStreams("all", false, false);
expect(result).toHaveProperty("name", "all");
});
/// The QUERY decides freshness, not the store copy: a populated store must
/// not suppress the read, or a refetch could never reach the sidebar.
it("serves a fresh query entry without a request, even with a stale store copy", async () => {
queryClient.clear();
queryClient.setQueryData(
["org", "test-org", "streams", "nameList", "logs"],
[{ name: "from-query" }],
);
mockStore.state.streams.logs = { list: [{ name: "cached" }] };
const result: any = await streamsInstance.getStreams("logs", false, false);
expect(StreamService.nameList).not.toHaveBeenCalled();
expect(result.list).toEqual([{ name: "from-query" }]);
});
it("does not write a list into the store once the org has switched mid-fetch", async () => {
queryClient.clear();
let release: (v: any) => void = () => {};
vi.mocked(StreamService.nameList).mockReturnValueOnce(
new Promise((r) => (release = r)) as any,
);
const pending = streamsInstance.getStreams("logs", false, false);
await vi.waitFor(() => expect(StreamService.nameList).toHaveBeenCalled());
mockStore.state.selectedOrganization.identifier = "other-org";
release({ data: { list: [{ name: "usage", stream_type: "logs" }] } });
await pending;
expect(mockStore.dispatch).not.toHaveBeenCalledWith("streams/setStreams", expect.anything());
});
it("does not write the all-types lists into the store once the org has switched", async () => {
queryClient.clear();
let release: (v: any) => void = () => {};
vi.mocked(StreamService.nameList).mockReturnValue(new Promise((r) => (release = r)) as any);
const pending = streamsInstance.getStreams("all", false, false);
await vi.waitFor(() => expect(StreamService.nameList).toHaveBeenCalled());
mockStore.state.selectedOrganization.identifier = "other-org";
release({ data: { list: [{ name: "usage", stream_type: "logs" }] } });
await pending;
expect(mockStore.dispatch).not.toHaveBeenCalledWith("streams/setStreams", expect.anything());
expect(mockStore.commit).not.toHaveBeenCalledWith(
"streams/updateStreamsFetched",
expect.anything(),
);
});
it("rejects with the CancelledError itself when the fetch is cancelled", async () => {
queryClient.clear();
mockStore.state.streams.logs = null;
vi.mocked(StreamService.nameList).mockReturnValueOnce(new Promise(() => {}) as any);
const pending = streamsInstance.getStreams("logs", false, false);
await vi.waitFor(() => expect(StreamService.nameList).toHaveBeenCalled());
await queryClient.cancelQueries();
await expect(pending).rejects.toBeInstanceOf(CancelledError);
});
it("should force fetch streams when force parameter is true", async () => {
mockStore.state.streams.logs = { list: [{ name: "cached" }] };
await streamsInstance.getStreams("logs", false, false, true);
expect(StreamService.nameList).toHaveBeenCalled();
});
it("a forced read of one type reaches the server and leaves the other types' cached lists alone", async () => {
queryClient.clear();
const key = (type: string) => ["org", "test-org", "streams", "nameList", type];
queryClient.setQueryData(key("logs"), [{ name: "stale" }]);
queryClient.setQueryData(key("metrics"), [{ name: "stale" }]);
mockStore.state.streams.logs = { list: [{ name: "stale" }] };
vi.mocked(StreamService.nameList).mockResolvedValueOnce({
data: { list: [{ name: "test-stream", stream_type: "logs" }] },
} as any);
await streamsInstance.getStreams("logs", false, false, true);
expect(StreamService.nameList).toHaveBeenCalledTimes(1);
expect(queryClient.getQueryData(key("logs"))).toEqual([
{ name: "test-stream", stream_type: "logs" },
]);
expect(queryClient.getQueryState(key("metrics"))?.isInvalidated).toBe(false);
});
it("removeStream still invalidates when Vuex holds no copy of that type", async () => {
delete (mockStore.state.streams.streamsIndexMapping as any).traces;
const spy = vi.spyOn(queryClient, "invalidateQueries");
try {
streamsInstance.removeStream("gone", "traces");
await new Promise((resolve) => setTimeout(resolve, 0));
expect(spy).toHaveBeenCalledWith({ queryKey: ["org", "test-org", "streams"] });
} finally {
spy.mockRestore();
}
});
it("should handle getPaginatedStreams with all parameters", async () => {
const result = await streamsInstance.getPaginatedStreams(
"logs",
false,
true,
0,
50,
"test",
"name",
true,
);
expect(StreamService.nameList).toHaveBeenCalledWith(
"test-org",
"logs",
false,
0,
50,
"test",
"name",
true,
);
expect(result).toHaveProperty("name", "logs");
expect(result).toHaveProperty("total");
});
});
// Test 11-15: Individual Stream Functions
describe("Stream error messages", () => {
beforeEach(() => {
queryClient.clear();
mockStore.state.streams.logs = null;
});
it("uses the server message when the failure carries no message of its own", async () => {
vi.mocked(StreamService.nameList).mockRejectedValueOnce({
response: { status: 404, data: { message: "Stream _sessionreplay not found" } },
});
await expect(streamsInstance.getStreams("logs", false, false)).rejects.toThrow(
"Stream _sessionreplay not found",
);
});
it.each([undefined, null, {}])(
"falls back to a readable message instead of stringifying %s",
async (failure) => {
vi.mocked(StreamService.nameList).mockRejectedValueOnce(failure);
await expect(streamsInstance.getStreams("logs", false, false)).rejects.toThrow(
/^Stream request failed$/,
);
},
);
it("never rejects with an empty message for a bare string failure", async () => {
vi.mocked(StreamService.nameList).mockRejectedValueOnce("boom");
await expect(streamsInstance.getStreams("logs", false, false)).rejects.toThrow("boom");
});
});
describe("Individual Stream Functions", () => {
it("should return null for getStream with empty parameters", async () => {
const result = await streamsInstance.getStream("", "logs", false);
expect(result).toBeNull();
});
it("should get stream with schema when requested", async () => {
// Set up stream in cache
mockStore.state.streams.logs = {
list: [{ name: "test-stream", schema: [] }],
};
mockStore.state.streams.streamsIndexMapping.logs["test-stream"] = 0;
await streamsInstance.getStream("test-stream", "logs", true);
expect(StreamService.schema).toHaveBeenCalledWith("test-org", "test-stream", "logs");
});
it("should reject with 'Stream Not Found' for non-existent stream", async () => {
mockStore.state.streams.logs = { list: [] };
await expect(streamsInstance.getStream("non-existent", "logs", false)).rejects.toThrow(
"Stream 'non-existent' not found for type 'logs'",
);
});
it("should handle getMultiStreams with valid streams", async () => {
const streams = [
{ streamName: "stream1", streamType: "logs", schema: false },
{ streamName: "stream2", streamType: "metrics", schema: true },
];
mockStore.state.streams.logs = { list: [{ name: "stream1" }] };
mockStore.state.streams.metrics = { list: [{ name: "stream2" }] };
mockStore.state.streams.streamsIndexMapping.logs["stream1"] = 0;
mockStore.state.streams.streamsIndexMapping.metrics["stream2"] = 0;
const results = await streamsInstance.getMultiStreams(streams);
expect(results).toHaveLength(2);
});
it("should return null for getMultiStreams with empty parameters", async () => {
const streams = [
{ streamName: "", streamType: "logs", schema: false },
{ streamName: "test", streamType: "", schema: false },
];
const results = await streamsInstance.getMultiStreams(streams);
expect(results).toEqual([null, null]);
});
});
describe("Schema prefetch", () => {
const listStream = (name: string, schema: any[] = []) => {
mockStore.state.streams.logs = { list: [{ name, schema }] } as any;
mockStore.state.streams.streamsIndexMapping.logs = { [name]: 0 } as any;
};
it("lets getStream use a schema request started before the stream list arrived", async () => {
streamsInstance.prefetchStreamSchema("prefetch-a", "logs");
expect(StreamService.schema).toHaveBeenCalledTimes(1);
expect(StreamService.schema).toHaveBeenCalledWith("test-org", "prefetch-a", "logs", {
silentForbidden: true,
});
listStream("prefetch-a");
const stream = await streamsInstance.getStream("prefetch-a", "logs", true);
expect(StreamService.schema).toHaveBeenCalledTimes(1);
expect(stream).toBeTruthy();
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({ streamType: "logs" }),
);
});
it("hands a prefetched schema to one read only", async () => {
streamsInstance.prefetchStreamSchema("prefetch-b", "logs");
listStream("prefetch-b");
await streamsInstance.getStream("prefetch-b", "logs", true);
await streamsInstance.getStream("prefetch-b", "logs", true);
expect(StreamService.schema).toHaveBeenCalledTimes(2);
});
it("asks the server again when the read is forced", async () => {
streamsInstance.prefetchStreamSchema("prefetch-c", "logs");
listStream("prefetch-c");
await streamsInstance.getStream("prefetch-c", "logs", true, true);
expect(StreamService.schema).toHaveBeenCalledTimes(2);
});
it("ignores a prefetch older than its time to live", async () => {
const now = Date.now();
const clock = vi.spyOn(Date, "now").mockReturnValue(now);
streamsInstance.prefetchStreamSchema("prefetch-d", "logs");
clock.mockReturnValue(now + 31_000);
listStream("prefetch-d");
await streamsInstance.getStream("prefetch-d", "logs", true);
clock.mockRestore();
expect(StreamService.schema).toHaveBeenCalledTimes(2);
});
it("starts nothing when the schema is already loaded", () => {
listStream("prefetch-e", [{ name: "field1", type: "string" }]);
streamsInstance.prefetchStreamSchema("prefetch-e", "logs");
expect(StreamService.schema).not.toHaveBeenCalled();
});
it("starts one request for repeated prefetches of the same stream", () => {
streamsInstance.prefetchStreamSchema("prefetch-f", "logs");
streamsInstance.prefetchStreamSchema("prefetch-f", "logs");
expect(StreamService.schema).toHaveBeenCalledTimes(1);
});
it("drops a failed prefetch so the next read asks the server again", async () => {
vi.mocked(StreamService.schema).mockRejectedValueOnce(new Error("offline"));
streamsInstance.prefetchStreamSchema("prefetch-g", "logs");
await new Promise((resolve) => setTimeout(resolve, 0));
listStream("prefetch-g");
await expect(streamsInstance.getStream("prefetch-g", "logs", true)).resolves.toBeTruthy();
expect(StreamService.schema).toHaveBeenCalledTimes(2);
});
it("ignores a prefetch without a stream name or type", () => {
streamsInstance.prefetchStreamSchema("", "logs");
streamsInstance.prefetchStreamSchema("prefetch-h", "");
expect(StreamService.schema).not.toHaveBeenCalled();
});
});
// Test 16-20: Schema and Data Manipulation
describe("Schema and Data Manipulation", () => {
it("should remove schema fields (_o2_id, _original, _all_values)", () => {
const streamData = {
name: "test",
schema: [
{ name: "field1", type: "string" },
{ name: "_o2_id", type: "string" },
{ name: "_original", type: "string" },
{ name: "_all_values", type: "string" },
{ name: "field2", type: "number" },
],
};
const result = streamsInstance.removeSchemaFields(streamData);
expect(result.schema).toHaveLength(2);
expect(result.schema.map((f: any) => f.name)).toEqual(["field1", "field2"]);
});
it("records the removed columns in removedSchemaFields and leaves schema as before", () => {
const streamData = {
name: "test",
schema: [
{ name: "field1", type: "Utf8" },
{ name: "_o2_id", type: "Int64" },
{ name: "_original", type: "Utf8" },
{ name: "_all_values", type: "LargeUtf8" },
{ name: "field2", type: "Int64" },
],
};
const result = streamsInstance.removeSchemaFields(streamData);
expect(result.schema).toEqual([
{ name: "field1", type: "Utf8" },
{ name: "field2", type: "Int64" },
]);
expect(result.removedSchemaFields).toEqual([
{ name: "_o2_id", type: "Int64" },
{ name: "_original", type: "Utf8" },
{ name: "_all_values", type: "LargeUtf8" },
]);
});
it("records an empty removedSchemaFields when nothing is removed", () => {
const result = streamsInstance.removeSchemaFields({
name: "test",
schema: [{ name: "field1", type: "Utf8" }],
});
expect(result.schema).toEqual([{ name: "field1", type: "Utf8" }]);
expect(result.removedSchemaFields).toEqual([]);
});
it("stores removedSchemaFields on a fetched stream", async () => {
mockStore.state.streams.logs = { list: [{ name: "test-stream", schema: [] }] } as any;
mockStore.state.streams.streamsIndexMapping.logs["test-stream"] = 0;
await streamsInstance.getStream("test-stream", "logs", true);
const setStreams = mockStore.dispatch.mock.calls.filter(
(call: any[]) => call[0] === "streams/setStreams",
);
const stream = setStreams.at(-1)?.[1].streams.list[0];
expect(stream.schema.map((f: any) => f.name)).toEqual(["field1"]);
expect(stream.removedSchemaFields.map((f: any) => f.name)).toEqual([
"_o2_id",
"_original",
"_all_values",
]);
});
it("should handle removeSchemaFields with no schema", () => {
const streamData = { name: "test" };
const result = streamsInstance.removeSchemaFields(streamData);
expect(result).toEqual({ name: "test" });
});
it("should check if stream is fetched correctly", () => {
// Test with specific stream type
mockStore.state.streams.logs = { list: [{ name: "test" }] };
expect(streamsInstance.isStreamFetched("logs")).toBe(true);
mockStore.state.streams.metrics = null;
expect(streamsInstance.isStreamFetched("metrics")).toBe(false);
// Test with 'all'
mockStore.state.streams.areAllStreamsFetched = true;
expect(streamsInstance.isStreamFetched("all")).toBe(true);
});
it("should check if specific stream exists", () => {
mockStore.state.streams.streamsIndexMapping.logs["existing-stream"] = 0;
expect(streamsInstance.isStreamExists("existing-stream", "logs")).toBe(true);
expect(streamsInstance.isStreamExists("non-existing", "logs")).toBe(false);
});
it("should handle isStreamExists with missing stream type", () => {
mockStore.state.streams.streamsIndexMapping = {};
expect(streamsInstance.isStreamExists("any-stream", "logs")).toBe(false);
});
});
// Test 21-25: Stream Management Functions
describe("Stream Management Functions", () => {
it("should set streams and update store correctly", () => {
const streamList = [
{ name: "stream1", stream_type: "logs" },
{ name: "stream2", stream_type: "logs" },
];
streamsInstance.setStreams("logs", streamList);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({
streamType: "logs",
streams: expect.objectContaining({
name: "logs",
list: streamList,
schema: false,
}),
}),
);
});
it("should handle setStreams with 'all' streamName", () => {
const streamList = [
{ name: "stream1", stream_type: "logs" },
{ name: "stream2", stream_type: "metrics" },
];
streamsInstance.setStreams("all", streamList);
// Should dispatch multiple times for different stream types
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({
streamType: "logs",
}),
);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({
streamType: "metrics",
}),
);
});
it("should add stream to existing cache", async () => {
const newStream = { name: "new-stream", stream_type: "logs" };
mockStore.state.streams.logs = { list: [] };
mockStore.state.streams.streamsIndexMapping.logs = {};
await streamsInstance.addStream(newStream);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({
streamType: "logs",
}),
);
});
it("should not add duplicate stream", async () => {
const existingStream = { name: "existing", stream_type: "logs" };
mockStore.state.streams.logs = { list: [] };
mockStore.state.streams.streamsIndexMapping.logs = { existing: 0 };
await streamsInstance.addStream(existingStream);
// Should not call dispatch since stream already exists
expect(mockStore.dispatch).not.toHaveBeenCalled();
});
it("should remove stream from cache correctly", () => {
mockStore.state.streams.logs = {
list: [{ name: "stream1" }, { name: "stream2" }, { name: "stream3" }],
};
mockStore.state.streams.streamsIndexMapping.logs = {
stream1: 0,
stream2: 1,
stream3: 2,
};
streamsInstance.removeStream("stream2", "logs");
expect(mockStore.dispatch).toHaveBeenCalled();
});
});
// Test 26-30: Utility Functions
describe("Utility Functions", () => {
it("should compare arrays correctly", () => {
const previousArray = {
0: { field: "field1", value: "old" },
1: { field: "field2", value: "same" },
};
const currentArray = [
{ field: "field2", value: "same" },
{ field: "field3", value: "new" },
];
const result = streamsInstance.compareArrays(previousArray, currentArray);
expect(result.add).toHaveLength(1);
expect(result.add[0].field).toBe("field3");
expect(result.remove).toHaveLength(1);
expect(result.remove[0].field).toBe("field1");
});
it("should perform deep equality check correctly", () => {
const obj1 = { a: 1, b: { c: 2 } };
const obj2 = { a: 1, b: { c: 2 } };
const obj3 = { a: 1, b: { c: 3 } };
expect(streamsInstance.deepEqual(obj1, obj2)).toBe(true);
expect(streamsInstance.deepEqual(obj1, obj3)).toBe(false);
expect(streamsInstance.deepEqual(null, null)).toBe(true);
expect(streamsInstance.deepEqual(null, obj1)).toBe(false);
});
it("should handle deepEqual with different key lengths", () => {
const obj1 = { a: 1, b: 2 };
const obj2 = { a: 1, b: 2, c: 3 };
expect(streamsInstance.deepEqual(obj1, obj2)).toBe(false);
});
it("should compare pattern associations correctly", () => {
const prev = [
{ field: "field1", pattern_id: "p1", policy: "policy1", apply_at: "ingest" },
{ field: "field2", pattern_id: "p2", policy: "policy2", apply_at: null },
];
const curr = [
{ field: "field1", pattern_id: "p1", policy: "policy1", apply_at: "ingest" },
{ field: "field3", pattern_id: "p3", policy: "policy3", apply_at: "query" },
];
const result = streamsInstance.comparePatternAssociations(prev, curr);
expect(result.add).toHaveLength(1);
expect(result.add[0].field).toBe("field3");
expect(result.remove).toHaveLength(1);
expect(result.remove[0].field).toBe("field2");
});
it("should handle pattern associations with null apply_at", () => {
const prev = [{ field: "field1", pattern_id: "p1", policy: "policy1", apply_at: null }];
const curr = [{ field: "field1", pattern_id: "p1", policy: "policy1", apply_at: undefined }];
const result = streamsInstance.comparePatternAssociations(prev, curr);
expect(result.add).toHaveLength(0);
expect(result.remove).toHaveLength(0);
});
});
// Test 31-35: Settings and Configuration
describe("Settings and Configuration Functions", () => {
it("should get updated settings for various attributes", () => {
const previousSettings = {
fields: ["field1", "field2"],
partition_keys: [{ field: "pk1", disabled: false }],
index_fields: ["index1"],
full_text_search_keys: ["fts1"],
bloom_filter_fields: ["bloom1"],
defined_schema_fields: ["schema1"],
extended_retention_days: [{ days: 30 }],
pattern_associations: [
{ field: "f1", pattern_id: "p1", policy: "pol1", apply_at: "ingest" },
],
};
const currentSettings = {
fields: ["field1", "field3"],
partition_keys: [{ field: "pk2", disabled: false }],
index_fields: ["index1", "index2"],
full_text_search_keys: ["fts2"],
bloom_filter_fields: ["bloom2"],
defined_schema_fields: ["schema2"],
extended_retention_days: [{ days: 60 }],
pattern_associations: [
{ field: "f2", pattern_id: "p2", policy: "pol2", apply_at: "query" },
],
};
const result = streamsInstance.getUpdatedSettings(previousSettings, currentSettings);
expect(result.fields.add).toEqual(["field3"]);
expect(result.fields.remove).toEqual(["field2"]);
expect(result.index_fields.add).toEqual(["index2"]);
});
it("should handle empty settings in getUpdatedSettings", () => {
const result = streamsInstance.getUpdatedSettings({}, {});
expect(result).toHaveProperty("fields");
expect(result).toHaveProperty("partition_keys");
expect(result.fields.add).toEqual([]);
expect(result.fields.remove).toEqual([]);
});
it("should reset specific stream type", () => {
streamsInstance.resetStreamType("logs");
expect(mockStore.dispatch).toHaveBeenCalledWith("streams/setStreams", {
streamType: "logs",
streams: {},
});
});
it("should handle resetStreamType with empty string", () => {
streamsInstance.resetStreamType("");
// Should not call dispatch
expect(mockStore.dispatch).not.toHaveBeenCalled();
});
it("should handle resetStreamType error gracefully", () => {
// Mock store dispatch to throw error
mockStore.dispatch = vi.fn().mockImplementation(() => {
throw new Error("Test error");
});
streamsInstance.resetStreamType("logs");
// Restore dispatch for subsequent tests
mockStore.dispatch = vi.fn();
});
});
// Test 36-40: Advanced Stream Operations
describe("Advanced Stream Operations", () => {
it("should reset all streams completely", () => {
streamsInstance.resetStreams();
// Should reset all stream types
const streamTypes = ["logs", "metrics", "traces", "enrichment_tables", "index", "metadata"];
streamTypes.forEach((type) => {
expect(mockStore.dispatch).toHaveBeenCalledWith("streams/setStreams", {
streamType: type,
streams: null,
});
});
expect(mockStore.commit).toHaveBeenCalledWith("streams/updateStreamIndexMapping", {});
expect(mockStore.commit).toHaveBeenCalledWith("streams/updateStreamsFetched", false);
expect(mockStore.dispatch).toHaveBeenCalledWith("setIsDataIngested", false);
});
it("should get all streams payload correctly", () => {
mockStore.state.streams.logs = { list: [{ name: "log1" }] };
mockStore.state.streams.metrics = { list: [{ name: "metric1" }] };
const result = streamsInstance.getAllStreamsPayload();
expect(result.name).toBe("all");
expect(result.schema).toBe(false);
expect(result.list).toEqual([{ name: "log1" }, { name: "metric1" }]);
});
it("includes cached profiles streams in the all-streams payload", () => {
mockStore.state.streams.logs = { list: [{ name: "log1" }] };
(mockStore.state.streams as any).profiles = { list: [{ name: "prof1" }] };
expect(streamsInstance.getAllStreamsPayload().list).toEqual([
{ name: "log1" },
{ name: "prof1" },
]);
});
it("should add new streams only if they don't exist", () => {
const existingStreams = [{ name: "existing", stream_type: "logs" }];
const newStreams = [
{ name: "existing", stream_type: "logs" },
{ name: "new-stream", stream_type: "logs" },
];
mockStore.state.streams.logs = { list: existingStreams };
mockStore.state.streams.streamsIndexMapping.logs = { existing: 0 };
streamsInstance.addNewStreams("logs", newStreams);
// Should only add the new stream
expect(mockStore.dispatch).toHaveBeenCalled();
});
it("should not add streams if all exist", () => {
const existingStreams = [{ name: "existing", stream_type: "logs" }];
const duplicateStreams = [{ name: "existing", stream_type: "logs" }];
// Set up the cache properly
mockStore.state.streams.logs = { list: existingStreams };
mockStore.state.streams.streamsIndexMapping.logs = { existing: 0 };
// Clear previous calls
vi.clearAllMocks();
streamsInstance.addNewStreams("logs", duplicateStreams);
// The function will still call setStreams with existing streams + no new streams
// So we verify it's called with the same streams (no new additions)
const setStreamsCalls = mockStore.dispatch.mock.calls.filter(
(call) => call[0] === "streams/setStreams",
);
expect(setStreamsCalls).toHaveLength(1);
expect(setStreamsCalls[0][1].streams.list).toEqual(existingStreams);
});
it("should not seed cache when addNewStreams called with empty cache", () => {
const newStreams = [{ name: "new-stream", stream_type: "logs" }];
mockStore.state.streams.logs = null;
streamsInstance.addNewStreams("logs", newStreams);
// Should NOT dispatch — partial results must not seed the cache,
// otherwise a full fetch on the logs page would be skipped.
expect(mockStore.dispatch).not.toHaveBeenCalled();
});
});
// Test 41-45: Error Handling and Edge Cases
describe("Error Handling and Edge Cases", () => {
it("should handle StreamService.nameList rejection", async () => {
vi.mocked(StreamService.nameList).mockRejectedValueOnce(new Error("Service error"));
await expect(streamsInstance.getStreams("logs", false, false)).rejects.toThrow(
"Service error",
);
});
it("should handle StreamService.schema rejection in getStream", async () => {
mockStore.state.streams.logs = { list: [{ name: "test", schema: [] }] };
mockStore.state.streams.streamsIndexMapping.logs["test"] = 0;
vi.mocked(StreamService.schema).mockRejectedValueOnce(new Error("Schema error"));
await expect(streamsInstance.getStream("test", "logs", true)).rejects.toThrow(
"Error while fetching schema: Schema error",
);
});
it("should handle getPaginatedStreams service error", async () => {
vi.mocked(StreamService.nameList).mockRejectedValueOnce(new Error("Pagination error"));
await expect(streamsInstance.getPaginatedStreams("logs", false, true)).rejects.toThrow(
"Pagination error",
);
});
it("should handle getMultiStreams with schema fetch error", async () => {
const streams = [{ streamName: "test", streamType: "logs", schema: true }];
mockStore.state.streams.logs = { list: [{ name: "test", schema: null }] };
mockStore.state.streams.streamsIndexMapping.logs["test"] = 0;
vi.mocked(StreamService.schema).mockRejectedValueOnce(new Error("Schema error"));
await expect(streamsInstance.getMultiStreams(streams)).rejects.toThrow("Schema error");
});
it("should handle isStreamExists with store access error", () => {
const consoleWarnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
// Create a store with an invalid structure to trigger the try-catch
const originalStore = mockStore;
mockStore.state.streams.streamsIndexMapping = {
get logs() {
throw new Error("Access error");
},
};
const result = streamsInstance.isStreamExists("test", "logs");
expect(result).toBe(false);
expect(consoleWarnSpy).toHaveBeenCalledWith(
"Error checking if stream exists:",
expect.any(Error),
);
// Restore
mockStore = originalStore;
consoleWarnSpy.mockRestore();
});
});
// Test 46-50: Promise and Async Behavior
describe("Promise and Async Behavior", () => {
it("should handle concurrent getStreams calls properly", async () => {
mockStore.state.streams.logs = null;
const promise1 = streamsInstance.getStreams("logs", false, false);
const promise2 = streamsInstance.getStreams("logs", false, false);
const results = await Promise.all([promise1, promise2]);
expect(results).toHaveLength(2);
// Due to the implementation, each composable instance may call the service
expect(StreamService.nameList).toHaveBeenCalled();
});
it("should handle notification dismissal in getStreams", async () => {
mockStore.state.streams.logs = null;
await streamsInstance.getStreams("logs", false, true);
expect(vi.mocked(toast)).toHaveBeenCalledWith({
variant: "loading",
message: "Please wait while loading streams...",
timeout: 0,
});
expect(mockToastDismiss).toHaveBeenCalled();
});
it("should handle notification dismissal in getPaginatedStreams", async () => {
await streamsInstance.getPaginatedStreams("logs", false, true);
expect(mockToastDismiss).toHaveBeenCalled();
});
it("should handle getStreams with notify=false", async () => {
mockStore.state.streams.logs = null;
await streamsInstance.getStreams("logs", false, false);
expect(vi.mocked(toast)).not.toHaveBeenCalled();
});
it("should handle Promise.allSettled with mixed results in getStreams", async () => {
mockStore.state.streams.areStreamsFetched = false;
// Mock one success and one failure
vi.mocked(StreamService.nameList)
.mockResolvedValueOnce({ data: { list: [{ name: "success", stream_type: "logs" }] } })
.mockRejectedValueOnce(new Error("Failed"));
const result = await streamsInstance.getStreams("all", false, false);
expect(result.name).toBe("all");
});
});
// Test 51-54: Data Ingestion and Store Integration
describe("Data Ingestion and Store Integration", () => {
it("should set isDataIngested when streams are added", () => {
mockStore.state.organizationData.isDataIngested = false;
const streamList = [{ name: "test", stream_type: "logs" }];
streamsInstance.setStreams("logs", streamList);
expect(mockStore.dispatch).toHaveBeenCalledWith("setIsDataIngested", true);
});
it("should not set isDataIngested for empty stream list", () => {
mockStore.state.organizationData.isDataIngested = false;
streamsInstance.setStreams("logs", []);
expect(mockStore.dispatch).not.toHaveBeenCalledWith("setIsDataIngested", expect.anything());
});
it("should not set isDataIngested for internal streams or non-user types", () => {
mockStore.state.organizationData.isDataIngested = false;
streamsInstance.setStreams("logs", [
{ name: "usage", stream_type: "logs" },
{ name: "_o2_db_stats", stream_type: "logs" },
]);
streamsInstance.setStreams("metadata", [{ name: "default", stream_type: "metadata" }]);
expect(mockStore.dispatch).not.toHaveBeenCalledWith("setIsDataIngested", expect.anything());
});
it("should set isDataIngested for a RUM stream", () => {
mockStore.state.organizationData.isDataIngested = false;
streamsInstance.setStreams("logs", [{ name: "_rumdata", stream_type: "logs" }]);
expect(mockStore.dispatch).toHaveBeenCalledWith("setIsDataIngested", true);
});
it("should handle deepEqual with nested objects", () => {
const obj1 = { a: { b: { c: 1 } } };
const obj2 = { a: { b: { c: 1 } } };
const obj3 = { a: { b: { c: 2 } } };
expect(streamsInstance.deepEqual(obj1, obj2)).toBe(true);
expect(streamsInstance.deepEqual(obj1, obj3)).toBe(false);
});
/// A write always lands. The caller only reaches setStreams with a list the
/// query just produced, so declining to store it would discard a refetch.
it("stores the list every time, including over an existing one", () => {
const streamList = [{ name: "test", stream_type: "logs" }];
vi.clearAllMocks();
mockStore.state.streams.logs = { list: [{ name: "already-here" }] };
streamsInstance.setStreams("logs", streamList);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({
streamType: "logs",
streams: expect.objectContaining({ list: streamList, name: "logs" }),
}),
);
// And again, over the list it just wrote.
vi.clearAllMocks();
streamsInstance.setStreams("logs", streamList);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({ streamType: "logs" }),
);
// An empty list is a real answer too — a type whose streams were all deleted.
vi.clearAllMocks();
const emptyStreamList: any[] = [];
streamsInstance.setStreams("metrics", emptyStreamList);
expect(mockStore.dispatch).toHaveBeenCalledWith(
"streams/setStreams",
expect.objectContaining({
streamType: "metrics",
streams: expect.objectContaining({ name: "metrics", list: emptyStreamList }),
}),
);
});
});
describe("Forced full read and fetched-at", () => {
it("a forced read of the whole list fetches every type, memory hits included", async () => {
queryClient.clear();
for (const type of [
"logs",
"metrics",
"traces",
"profiles",
"enrichment_tables",
"index",
"metadata",
]) {
(mockStore.state.streams as any)[type] = { list: [{ name: "cached" }] };
}
await streamsInstance.getStreams("all", false, false, true);
expect(StreamService.nameList).toHaveBeenCalledTimes(7);
});
it("getStreamsFetchedAt reports the oldest type list, or nothing when none is cached", async () => {
queryClient.clear();
const key = (type: string) => ["org", "test-org", "streams", "nameList", type];
queryClient.setQueryData(key("logs"), [], { updatedAt: 2000 });
queryClient.setQueryData(key("metrics"), [], { updatedAt: 1000 });
expect(await streamsInstance.getStreamsFetchedAt()).toBe(1000);
expect(await streamsInstance.getStreamsFetchedAt("logs")).toBe(2000);
expect(await streamsInstance.getStreamsFetchedAt("traces")).toBeUndefined();
});
});
});