[8.x] [Streams] Schema API integration tests (#204401)

x-pack/test/api_integration/apis/streams/helpers/requests.ts

@@ -60,3 +60,15 @@ export async function deleteStream(supertest: Agent, id: string) {
   const response = await req.send().expect(200);
   return response.body;
 }
+
+export async function getUnmappedFieldsForStream(supertest: Agent, id: string) {
+  const req = supertest.get(`/api/streams/${id}/schema/unmapped_fields`).set('kbn-xsrf', 'xxx');
+  const response = await req.send().expect(200);
+  return response.body;
+}
+
+export async function simulateFieldsForStream(supertest: Agent, id: string, body: JsonObject) {
+  const req = supertest.post(`/api/streams/${id}/schema/fields_simulation`).set('kbn-xsrf', 'xxx');
+  const response = await req.send(body).expect(200);
+  return response.body;
+}

x-pack/test/api_integration/apis/streams/index.ts

@@ -13,5 +13,6 @@ export default function ({ loadTestFile }: FtrProviderContext) {
     loadTestFile(require.resolve('./enrichment'));
     loadTestFile(require.resolve('./classic'));
     loadTestFile(require.resolve('./flush_config'));
+    loadTestFile(require.resolve('./schema'));
   });
 }

x-pack/test/api_integration/apis/streams/schema.ts

@@ -0,0 +1,98 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import expect from '@kbn/expect';
+import {
+  deleteStream,
+  enableStreams,
+  forkStream,
+  getUnmappedFieldsForStream,
+  indexDocument,
+  simulateFieldsForStream,
+} from './helpers/requests';
+import { FtrProviderContext } from '../../ftr_provider_context';
+import { cleanUpRootStream } from './helpers/cleanup';
+import { waitForDocumentInIndex } from '../../../alerting_api_integration/observability/helpers/alerting_wait_for_helpers';
+
+export default function ({ getService }: FtrProviderContext) {
+  const supertest = getService('supertest');
+  const esClient = getService('es');
+  const retryService = getService('retry');
+  const logger = getService('log');
+
+  describe('Streams Schema', () => {
+    after(async () => {
+      await deleteStream(supertest, 'logs.nginx');
+      await cleanUpRootStream(esClient);
+    });
+
+    before(async () => {
+      await enableStreams(supertest);
+
+      const doc = {
+        '@timestamp': '2024-01-01T00:00:10.000Z',
+        message: '2023-01-01T00:00:10.000Z error test',
+        ['some.field']: 'some value',
+        ['another.field']: 'another value',
+        lastField: 'last value',
+        ['log.level']: 'warning',
+      };
+
+      await indexDocument(esClient, 'logs', doc);
+      await waitForDocumentInIndex({ esClient, indexName: 'logs', retryService, logger });
+    });
+
+    describe('Unmapped fields API', () => {
+      it('Returns unmapped fields', async () => {
+        const response = await getUnmappedFieldsForStream(supertest, 'logs');
+        expect(response.unmappedFields).to.eql(['another.field', 'lastField', 'some.field']);
+      });
+    });
+
+    describe('Fields simulation API', () => {
+      it('Returns failure status when simulation would fail', async () => {
+        const response = await simulateFieldsForStream(supertest, 'logs', {
+          field_definitions: [{ name: 'message', type: 'boolean' }],
+        });
+
+        expect(response.status).to.be('failure');
+        expect(response.simulationError).to.be.a('string');
+        expect(response.documentsWithRuntimeFieldsApplied).to.be(null);
+      });
+      it('Returns success status when simulation would succeed', async () => {
+        const response = await simulateFieldsForStream(supertest, 'logs', {
+          field_definitions: [{ name: 'message', type: 'keyword' }],
+        });
+
+        expect(response.status).to.be('success');
+        expect(response.simulationError).to.be(null);
+        expect(response.documentsWithRuntimeFieldsApplied).length(1);
+      });
+      it('Returns unknown status when documents are missing and status cannot be determined', async () => {
+        const forkBody = {
+          stream: {
+            name: 'logs.nginx',
+          },
+          condition: {
+            field: 'log.logger',
+            operator: 'eq',
+            value: 'nginx',
+          },
+        };
+
+        await forkStream(supertest, 'logs', forkBody);
+        const response = await simulateFieldsForStream(supertest, 'logs.nginx', {
+          field_definitions: [{ name: 'message', type: 'keyword' }],
+        });
+
+        expect(response.status).to.be('unknown');
+        expect(response.simulationError).to.be(null);
+        expect(response.documentsWithRuntimeFieldsApplied).to.be(null);
+      });
+    });
+  });
+}