You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2021/08/11 13:11:10 UTC
[incubator-streampipes] branch STREAMPIPES-319 updated:
[STREAMPIPES-404] Refactor pipeline managent endpoint
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch STREAMPIPES-319
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/STREAMPIPES-319 by this push:
new c1d9e4f [STREAMPIPES-404] Refactor pipeline managent endpoint
new ea2f6ef Merge branch 'STREAMPIPES-319' of github.com:apache/incubator-streampipes into STREAMPIPES-319
c1d9e4f is described below
commit c1d9e4f3c12ceac2242c1ce32abf97538d88a4a7
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Wed Aug 11 15:09:54 2021 +0200
[STREAMPIPES-404] Refactor pipeline managent endpoint
---
streampipes-pipeline-management/pom.xml | 15 +++
.../manager/pipeline/PipelineManager.java | 108 +++++++++++++++
.../manager/pipeline/TestPipelineExecution.java | 33 -----
.../manager/pipeline/TestPipelineManager.java | 148 +++++++++++++++++++++
.../manager/pipeline/TestPipelineStorage.java | 42 ------
.../streampipes/rest/impl/PipelineResource.java | 40 +++---
.../rest/management/PipelineManagement.java | 1 +
ui/cypress/support/utils/AdapterUtils.ts | 4 -
ui/cypress/support/utils/PipelineUtils.ts | 2 +-
ui/cypress/tests/pipeline/pipelineTest.ts | 50 +++----
10 files changed, 318 insertions(+), 125 deletions(-)
diff --git a/streampipes-pipeline-management/pom.xml b/streampipes-pipeline-management/pom.xml
index 4bcb781..8b0568f 100644
--- a/streampipes-pipeline-management/pom.xml
+++ b/streampipes-pipeline-management/pom.xml
@@ -139,5 +139,20 @@
<version>0.69.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-module-junit4</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.powermock</groupId>
+ <artifactId>powermock-api-mockito2</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java
new file mode 100644
index 0000000..56b7e3e
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.pipeline;
+
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.manager.storage.UserManagementService;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.storage.api.IPipelineStorage;
+import org.apache.streampipes.storage.management.StorageDispatcher;
+
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+
+public class PipelineManager {
+
+ /**
+ * Returns all the pipelines of the user
+ * @param username
+ * @return all pipelines
+ */
+ public static List<Pipeline> getOwnPipelines(String username) {
+ return UserManagementService.getUserService().getOwnPipelines(username);
+ }
+
+ /**
+ * Returns the stored pipeline with the given pipeline Id
+ * @param pipelineId
+ * @return pipeline
+ */
+ public static Pipeline getPipeline(String pipelineId) {
+ return getPipelineStorage().getPipeline(pipelineId);
+ }
+
+ /**
+ * Adds a new pipeline for the user with the username to the storage
+ * @param username
+ * @param pipeline
+ * @return the pipelineId
+ */
+ public static String addPipeline(String username, Pipeline pipeline) {
+
+ String pipelineId = UUID.randomUUID().toString();
+ pipeline.setPipelineId(pipelineId);
+ pipeline.setRunning(false);
+ pipeline.setCreatedByUser(username);
+ pipeline.setCreatedAt(new Date().getTime());
+ pipeline.getSepas().forEach(processor -> processor.setCorrespondingUser(username));
+ pipeline.getActions().forEach(action -> action.setCorrespondingUser(username));
+ Operations.storePipeline(pipeline);
+
+ return pipelineId;
+ }
+
+ /**
+ * Starts all processing elements of the pipeline with the pipelineId
+ * @param pipelineId
+ * @return pipeline status of the start operation
+ */
+ public static PipelineOperationStatus startPipeline(String pipelineId) {
+ Pipeline pipeline = getPipeline(pipelineId);
+ PipelineOperationStatus status = Operations.startPipeline(pipeline);
+ return status;
+ }
+
+ /**
+ * Stops all processing elements of the pipeline
+ * @param pipelineId
+ * @param forceStop when it is true, the pipeline is stopped, even if not all processing element containers could be reached
+ * @return pipeline status of the start operation
+ */
+ public static PipelineOperationStatus stopPipeline(String pipelineId,
+ boolean forceStop) {
+ Pipeline pipeline = getPipeline(pipelineId);
+ PipelineOperationStatus status = Operations.stopPipeline(pipeline, forceStop);
+
+ return status;
+ }
+
+ /**
+ * Deletes the pipeline with the pipeline Id
+ * @param pipelineId
+ */
+ public static void deletePipeline(String pipelineId) {
+ getPipelineStorage().deletePipeline(pipelineId);
+ }
+
+ public static IPipelineStorage getPipelineStorage() {
+ return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI();
+ }
+}
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineExecution.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineExecution.java
deleted file mode 100644
index 22ba23c..0000000
--- a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineExecution.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.manager.pipeline;
-
-import org.apache.streampipes.manager.operations.Operations;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.storage.management.StorageDispatcher;
-
-public class TestPipelineExecution {
-
- public static void main(String[] args) {
- Pipeline pipeline = StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getPipeline
- ("850c2850-a27c-4026-939c-171db57c50f0");
-
- Operations.startPipeline(pipeline);
- }
-}
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineManager.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineManager.java
new file mode 100644
index 0000000..2f4190b
--- /dev/null
+++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineManager.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.streampipes.manager.pipeline;
+
+import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.manager.storage.UserManagementService;
+import org.apache.streampipes.manager.storage.UserService;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.storage.couchdb.impl.PipelineStorageImpl;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static junit.framework.TestCase.assertEquals;
+import static junit.framework.TestCase.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({
+ UserManagementService.class,
+ PipelineManager.class,
+ Operations.class})
+public class TestPipelineManager {
+
+ @Before
+ public void before() {
+ PowerMockito.mockStatic(
+ UserManagementService.class);
+ }
+
+ @Test
+ public void testGetOwnPipelines() {
+ // Prepare
+ Pipeline expectedPipeline = this.getPipeline();
+ List<Pipeline> expected = Arrays.asList(expectedPipeline);
+
+ UserService userService = mock(UserService.class);
+ when(userService.getOwnPipelines(any(String.class))).thenReturn(expected);
+ when(UserManagementService.getUserService()).thenReturn(userService);
+
+ // Test
+ List<Pipeline> result = PipelineManager.getOwnPipelines("user@test.com");
+
+ // Assertions
+ assertEquals(1, result.size());
+ assertEquals(this.getPipelineName(), result.get(0).getName());
+ }
+
+ @Test
+ public void testGetPipeline() {
+ // Prepare
+ Pipeline expectedPipeline = this.getPipeline();
+
+ PipelineStorageImpl pipelineStorageImpl = mock(PipelineStorageImpl.class);
+ when(pipelineStorageImpl.getPipeline(any(String.class))).thenReturn(expectedPipeline);
+ PowerMockito.stub(PowerMockito.method(PipelineManager.class, "getPipelineStorage")).toReturn(pipelineStorageImpl);
+
+ // Test
+ Pipeline result = PipelineManager.getPipeline("pipelineid");
+
+ // Assertions
+ assertNotNull(result);
+ assertEquals(this.getPipelineName(), result.getName());
+ }
+
+
+ @Test
+ public void testStartPipeline() {
+ // Prepare
+ PipelineOperationStatus expectedPipelineOperationStatus = getPipelineOperationStatus();
+ PowerMockito.stub(PowerMockito.method(PipelineManager.class, "getPipeline", String.class)).toReturn(getPipeline());
+ PowerMockito.stub(PowerMockito.method(Operations.class, "startPipeline", Pipeline.class)).toReturn(expectedPipelineOperationStatus);
+
+ // Test
+ PipelineOperationStatus result = PipelineManager.startPipeline("pipelineId");
+
+ // Assertions
+ assertNotNull(result);
+ assertEquals(expectedPipelineOperationStatus.getPipelineName(), result.getPipelineName());
+ }
+
+ @Test
+ public void testStopPipeline() {
+ // Prepare
+ PipelineOperationStatus expectedPipelineOperationStatus = getPipelineOperationStatus();
+ PowerMockito.stub(PowerMockito.method(PipelineManager.class, "getPipeline", String.class)).toReturn(getPipeline());
+ PowerMockito.stub(PowerMockito.method(Operations.class, "stopPipeline", Pipeline.class, boolean.class)).toReturn(expectedPipelineOperationStatus);
+
+ // Test
+ PipelineOperationStatus result = PipelineManager.stopPipeline("pipelineId", true);
+
+ // Assertions
+ assertNotNull(result);
+ assertEquals(expectedPipelineOperationStatus.getPipelineName(), result.getPipelineName());
+ }
+
+ @Test
+ public void testAddPipeline() {
+ // Prepare
+ PipelineOperationStatus expectedPipelineOperationStatus = getPipelineOperationStatus();
+ PowerMockito.mockStatic(Operations.class);
+
+ // Test
+ String result = PipelineManager.addPipeline("test@user.com", getPipeline());
+
+ // Assertions
+ assertNotNull(result);
+ }
+
+ private String getPipelineName() {
+ return "Test Pipeline";
+ }
+
+ private Pipeline getPipeline() {
+ Pipeline pipeline = new Pipeline();
+ pipeline.setPipelineId("testId");
+ pipeline.setName(this.getPipelineName());
+ return pipeline;
+ }
+
+ private PipelineOperationStatus getPipelineOperationStatus() {
+ return new PipelineOperationStatus("", getPipelineName(),"", new ArrayList<>());
+ }
+}
\ No newline at end of file
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineStorage.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineStorage.java
deleted file mode 100644
index 6a239d8..0000000
--- a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/TestPipelineStorage.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.manager.pipeline;
-
-import com.google.common.base.Charsets;
-import com.google.common.io.Resources;
-import org.apache.streampipes.manager.operations.Operations;
-import org.apache.streampipes.model.pipeline.Pipeline;
-import org.apache.streampipes.serializers.json.Utils;
-
-import java.io.IOException;
-import java.net.URL;
-
-public class TestPipelineStorage {
-
- public static void main(String[] args) {
- URL url = Resources.getResource("pipeline.json");
- try {
- String pipelineString = Resources.toString(url, Charsets.UTF_8);
- Pipeline pipeline = Utils.getGson().fromJson(pipelineString, Pipeline.class);
- Operations.storePipeline(pipeline);
- } catch (IOException e) {
- e.printStackTrace();
- }
- }
-}
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
index 53fd658..2a9b702 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
@@ -28,6 +28,7 @@ import io.swagger.v3.oas.annotations.responses.ApiResponse;
import org.apache.streampipes.commons.exceptions.*;
import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
import org.apache.streampipes.manager.operations.Operations;
+import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.model.SpDataSet;
import org.apache.streampipes.model.client.exception.InvalidConnectionException;
import org.apache.streampipes.model.message.Notification;
@@ -36,7 +37,6 @@ import org.apache.streampipes.model.message.Notifications;
import org.apache.streampipes.model.message.SuccessMessage;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
-import org.apache.streampipes.rest.management.PipelineManagement;
import org.apache.streampipes.rest.shared.annotation.JacksonSerialized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,8 +44,6 @@ import org.slf4j.LoggerFactory;
import javax.ws.rs.*;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
-import java.util.Date;
-import java.util.UUID;
@Path("/v2/users/{username}/pipelines")
public class PipelineResource extends AbstractAuthGuardedRestResource {
@@ -75,8 +73,7 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
array = @ArraySchema(schema = @Schema(implementation = Pipeline.class)))
})})
public Response getOwn(@PathParam("username") String username) {
- return ok(getUserService()
- .getOwnPipelines(username));
+ return ok(PipelineManager.getOwnPipelines(username));
}
@GET
@@ -115,8 +112,8 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
@JacksonSerialized
@Operation(summary = "Delete a pipeline with a given id",
tags = {"Pipeline"})
- public Response removeOwn(@PathParam("username") String username, @PathParam("pipelineId") String elementUri) {
- getPipelineStorage().deletePipeline(elementUri);
+ public Response removeOwn(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) {
+ PipelineManager.deletePipeline(pipelineId);
return statusMessage(Notifications.success("Pipeline deleted"));
}
@@ -127,7 +124,7 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
@Operation(summary = "Get a specific pipeline with the given id",
tags = {"Pipeline"})
public Response getElement(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) {
- return ok(getPipelineStorage().getPipeline(pipelineId));
+ return ok(PipelineManager.getPipeline(pipelineId));
}
@Path("/{pipelineId}/start")
@@ -138,9 +135,8 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
tags = {"Pipeline"})
public Response start(@PathParam("username") String username, @PathParam("pipelineId") String pipelineId) {
try {
- Pipeline pipeline = getPipelineStorage()
- .getPipeline(pipelineId);
- PipelineOperationStatus status = Operations.startPipeline(pipeline);
+ PipelineOperationStatus status = PipelineManager.startPipeline(pipelineId);
+
return ok(status);
} catch (Exception e) {
e.printStackTrace();
@@ -158,8 +154,15 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
@PathParam("pipelineId") String pipelineId,
@QueryParam("forceStop") @DefaultValue("false") boolean forceStop) {
logger.info("User: " + username + " stopped pipeline: " + pipelineId);
- PipelineManagement pm = new PipelineManagement();
- return pm.stopPipeline(pipelineId, forceStop);
+
+ try {
+ PipelineOperationStatus status = PipelineManager.stopPipeline(pipelineId, forceStop);
+ return ok(status);
+ } catch
+ (Exception e) {
+ e.printStackTrace();
+ return constructErrorMessage(new Notification(NotificationType.UNKNOWN_ERROR.title(), NotificationType.UNKNOWN_ERROR.description(), e.getMessage()));
+ }
}
@POST
@@ -168,15 +171,8 @@ public class PipelineResource extends AbstractAuthGuardedRestResource {
@Operation(summary = "Store a new pipeline",
tags = {"Pipeline"})
public Response addPipeline(@PathParam("username") String username, Pipeline pipeline) {
- String pipelineId = UUID.randomUUID().toString();
- pipeline.setPipelineId(pipelineId);
- pipeline.setRunning(false);
- pipeline.setCreatedByUser(username);
- pipeline.setCreatedAt(new Date().getTime());
- pipeline.getSepas().forEach(processor -> processor.setCorrespondingUser(username));
- pipeline.getActions().forEach(action -> action.setCorrespondingUser(username));
- //userService.addOwnPipeline(username, pipeline);
- Operations.storePipeline(pipeline);
+
+ String pipelineId = PipelineManager.addPipeline(username, pipeline);
SuccessMessage message = Notifications.success("Pipeline stored");
message.addNotification(new Notification("id", pipelineId));
return ok(message);
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/management/PipelineManagement.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/management/PipelineManagement.java
index 1b80671..f90fd89 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/management/PipelineManagement.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/management/PipelineManagement.java
@@ -27,6 +27,7 @@ import org.apache.streampipes.rest.impl.AbstractRestResource;
import javax.ws.rs.core.Response;
+@Deprecated
public class PipelineManagement extends AbstractRestResource {
public Response stopPipeline(String pipelineId,
diff --git a/ui/cypress/support/utils/AdapterUtils.ts b/ui/cypress/support/utils/AdapterUtils.ts
index 3a79c9c..56f68c8 100644
--- a/ui/cypress/support/utils/AdapterUtils.ts
+++ b/ui/cypress/support/utils/AdapterUtils.ts
@@ -96,10 +96,6 @@ export class AdapterUtils {
}
private static goToConnect() {
- it('Login', () => {
- cy.login();
- });
-
it('Go to StreamPipes connect', () => {
cy.visit('#/connect');
});
diff --git a/ui/cypress/support/utils/PipelineUtils.ts b/ui/cypress/support/utils/PipelineUtils.ts
index 5c52f07..f1b2a72 100644
--- a/ui/cypress/support/utils/PipelineUtils.ts
+++ b/ui/cypress/support/utils/PipelineUtils.ts
@@ -43,7 +43,7 @@ export class PipelineUtils {
private static goToPipelineEditor() {
- it('Go to StreamPipes connect', () => {
+ it('Go to StreamPipes editor', () => {
cy.visit('#/editor');
});
}
diff --git a/ui/cypress/tests/pipeline/pipelineTest.ts b/ui/cypress/tests/pipeline/pipelineTest.ts
index d3bd2a6..630012a 100644
--- a/ui/cypress/tests/pipeline/pipelineTest.ts
+++ b/ui/cypress/tests/pipeline/pipelineTest.ts
@@ -21,29 +21,33 @@ import { PipelineUtils } from '../../support/utils/PipelineUtils';
import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
+const adapterName = 'simulator';
+
+before('Setup Test', () => {
+ it('Login', () => {
+ cy.login();
+ });
+
+ AdapterUtils.addMachineDataSimulator(adapterName);
+});
+
describe('Test Random Data Simulator Stream Adapter', () => {
- it('Login', () => {
- cy.login();
- });
-
- const adapterName = 'simulator';
- AdapterUtils.addMachineDataSimulator(adapterName);
-
- const pipelineInput = PipelineBuilder.create('Pipeline Test')
- .addSource(adapterName)
- .addProcessingElement(
- PipelineElementBuilder.create('field_renamer')
- .addInput('drop-down', 'convert-property', 'timestamp')
- .addInput('input', 'field-name', 't')
- .build())
- .addSink(
- PipelineElementBuilder.create('dashboard_sink')
- .addInput('input', 'visualization-name', 'Demo')
- .build())
- .build();
-
- PipelineUtils.testPipeline(pipelineInput);
-
- AdapterUtils.deleteAdapter();
+ const pipelineInput = PipelineBuilder.create('Pipeline Test')
+ .addSource(adapterName)
+ .addProcessingElement(
+ PipelineElementBuilder.create('field_renamer')
+ .addInput('drop-down', 'convert-property', 'timestamp')
+ .addInput('input', 'field-name', 't')
+ .build())
+ .addSink(
+ PipelineElementBuilder.create('dashboard_sink')
+ .addInput('input', 'visualization-name', 'Demo')
+ .build())
+ .build();
+
+ PipelineUtils.testPipeline(pipelineInput);
+
+ AdapterUtils.deleteAdapter();
});
+