From edc92da43723b4a45c7f530758a9f1e4cd1f593e Mon Sep 17 00:00:00 2001
From: Kerry Gallagher <kerry.gallagher@elastic.co>
Date: Thu, 2 Jan 2025 15:34:29 +0000
Subject: [PATCH] [Streams] Schema API integration tests (#204401)

## Summary

Just API integration tests for the APIs added in
https://github.com/elastic/kibana/pull/202372.
---
 .../apis/streams/helpers/requests.ts          | 12 +++
 .../api_integration/apis/streams/index.ts     |  1 +
 .../api_integration/apis/streams/schema.ts    | 98 +++++++++++++++++++
 3 files changed, 111 insertions(+)
 create mode 100644 x-pack/test/api_integration/apis/streams/schema.ts

diff --git a/x-pack/test/api_integration/apis/streams/helpers/requests.ts b/x-pack/test/api_integration/apis/streams/helpers/requests.ts
index ae3a325b5f9b4..f26c53aaaf0aa 100644
--- a/x-pack/test/api_integration/apis/streams/helpers/requests.ts
+++ b/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;
+}
diff --git a/x-pack/test/api_integration/apis/streams/index.ts b/x-pack/test/api_integration/apis/streams/index.ts
index 14decb2400196..6c4cf358b8ac3 100644
--- a/x-pack/test/api_integration/apis/streams/index.ts
+++ b/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'));
   });
 }
diff --git a/x-pack/test/api_integration/apis/streams/schema.ts b/x-pack/test/api_integration/apis/streams/schema.ts
new file mode 100644
index 0000000000000..c8b90e90939c1
--- /dev/null
+++ b/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);
+      });
+    });
+  });
+}