Integration Tests

Author: lukewhiting

modules/streams/src/internalClusterTest/java/org/elasticsearch/rest/streams/TestToggleIT.java

@@ -0,0 +1,61 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.rest.streams;
+
+import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
+import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.rest.streams.logs.LogsStreamsActivationToggleAction;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.hamcrest.ElasticsearchAssertions;
+import org.elasticsearch.test.transport.MockTransportService;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.is;
+
+public class TestToggleIT extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(StreamsPlugin.class, MockTransportService.TestPlugin.class);
+    }
+
+    public void testLogStreamToggle() throws IOException, ExecutionException, InterruptedException {
+        boolean[] testParams = new boolean[] { true, false, true };
+        for (boolean enable : testParams) {
+            doLogStreamToggleTest(enable);
+        }
+    }
+
+    private void doLogStreamToggleTest(boolean enable) throws IOException, ExecutionException, InterruptedException {
+        LogsStreamsActivationToggleAction.Request request = new LogsStreamsActivationToggleAction.Request(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            enable
+        );
+
+        AcknowledgedResponse acknowledgedResponse = client().execute(LogsStreamsActivationToggleAction.INSTANCE, request).get();
+        ElasticsearchAssertions.assertAcked(acknowledgedResponse);
+
+        ClusterStateRequest state = new ClusterStateRequest(TEST_REQUEST_TIMEOUT);
+        ClusterStateResponse clusterStateResponse = client().admin().cluster().state(state).get();
+        ProjectMetadata projectMetadata = clusterStateResponse.getState().metadata().getProject(ProjectId.DEFAULT);
+
+        assertThat(projectMetadata.<StreamsMetadata>custom(StreamsMetadata.TYPE).isLogsEnabled(), is(enable));
+    }
+
+}

modules/streams/src/main/java/org/elasticsearch/rest/streams/StreamsMetadata.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.cluster.AbstractNamedDiffable;
+import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -22,6 +23,7 @@
 import java.io.IOException;
 import java.util.EnumSet;
 import java.util.Iterator;
+import java.util.Objects;
 
 /**
  * Metadata for the Streams feature, which allows enabling or disabling logs for data streams.
@@ -46,10 +48,6 @@ public boolean isLogsEnabled() {
         return logsEnabled;
     }
 
-    public void setLogsEnabled(boolean logsEnabled) {
-        this.logsEnabled = logsEnabled;
-    }
-
     @Override
     public EnumSet<Metadata.XContentContext> context() {
         return Metadata.ALL_CONTEXTS;
@@ -65,6 +63,12 @@ public TransportVersion getMinimalSupportedVersion() {
         return TransportVersions.STREAMS_LOGS_SUPPORT;
     }
 
+    public static NamedDiff<Metadata.ProjectCustom> readDiffFrom(StreamInput in) throws IOException {
+        return readDiffFrom(Metadata.ProjectCustom.class, TYPE, in);
+    }
+
+
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(logsEnabled);
@@ -75,4 +79,17 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
         return Iterators.concat(ChunkedToXContentHelper.chunk((builder, bParams) -> builder.field("logs_enabled", logsEnabled)));
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if ((o instanceof StreamsMetadata that)) {
+            return logsEnabled == that.logsEnabled;
+        } else {
+            return false;
+        }
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(logsEnabled);
+    }
 }

modules/streams/src/main/java/org/elasticsearch/rest/streams/StreamsPlugin.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.rest.streams;
 
+import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
@@ -68,6 +69,9 @@ public List<ActionHandler> getActions() {
 
     @Override
     public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
-        return List.of(new NamedWriteableRegistry.Entry(Metadata.ProjectCustom.class, StreamsMetadata.TYPE, StreamsMetadata::new));
+        return List.of(
+            new NamedWriteableRegistry.Entry(Metadata.ProjectCustom.class, StreamsMetadata.TYPE, StreamsMetadata::new),
+            new NamedWriteableRegistry.Entry(NamedDiff.class, StreamsMetadata.TYPE, StreamsMetadata::readDiffFrom)
+        );
     }
 }