{"type":"doc","content":[{"type":"extension","attrs":{"layout":"default","extensionType":"com.atlassian.confluence.macro.core","extensionKey":"toc","parameters":{"macroParams":{"minLevel":{"value":"1"},"maxLevel":{"value":"2"}},"macroMetadata":{"macroId":{"value":"a5e2968003747a832698e875be764e3e"},"schemaVersion":{"value":"1"},"title":"Table of Contents"}},"localId":"9161784f-1e1c-4b95-820c-60621d157ab3"}},{"type":"heading","attrs":{"level":1},"content":[{"text":"Introduction","type":"text"}]},{"type":"paragraph","content":[{"text":"Onesait Platform's Dataflow module is a low-code tool for defining and executing data flows. In this article we will see how JUnit can be used to automate dataflow testing.","type":"text"}]},{"type":"paragraph","content":[{"text":"To test the pipelines we need an instance of Dataflow where to execute them. To keep this test completely independent of external resources, we use the TestContainers library to run a Dataflow instance in the test itself. It would be more efficient and faster when running the tests to have a dedicated instance with all the pipelines that need to be tested.","type":"text"}]},{"type":"paragraph","content":[{"text":"The purpose of this example is to show how this kind of tests can be performed, and therefore the pipeline we are going to test is very simple. Any other pipeline would be tested in a similar way.","type":"text"}]},{"type":"paragraph","content":[{"text":"The strategy followed in this example is to use the dataflow client library to manage the flows of an instance, execute a preview remotely and validate the values of the records in each of the stages. In other words, the same thing we would do manually and visually when using the preview of a pipeline, but automated.","type":"text"}]},{"type":"paragraph","content":[{"text":"Other types of tests could be performed, such as the complete execution of a pipeline and validating that the number of records in the outputs and the number and type of errors are as expected, but the information provided by this type of test is less accurate.","type":"text"}]},{"type":"paragraph","content":[{"text":"The complete code for this example is at ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/tree/main/test-examples/pipelines"}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Dataflow Instance Configuration","type":"text","marks":[{"type":"strong"}]}]},{"type":"paragraph","content":[{"text":"As mentioned earlier, this example uses the TestContainers library, which allows for automatically starting and stopping containers in a test.","type":"text"}]},{"type":"paragraph","content":[{"text":"To use it with JUnit 5, you need the following dependencies:","type":"text"}]},{"type":"codeBlock","content":[{"text":"\n org.testcontainers\n testcontainers\n 1.18.3\n test\n\n\n org.testcontainers\n junit-jupiter\n 1.18.3\n test\n","type":"text"}]},{"type":"paragraph","content":[{"text":"Configuring the container in a basic way is very simple; you just need to annotate the test class with ","type":"text"},{"text":"@Testcontainers","type":"text","marks":[{"type":"code"}]},{"text":" and create an attribute with the ","type":"text"},{"text":"@Container","type":"text","marks":[{"type":"code"}]},{"text":". annotation. For example, in our case:","type":"text"}]},{"type":"codeBlock","attrs":{"language":"java"},"content":[{"text":"private static final String IMAGE = \"registry.onesaitplatform.com/onesaitplatform/streamsets:4.1.0-overlord-323\";\nprivate static final int PORT = 18630;\n\n@Container\npublic static GenericContainer dataflow = new GenericContainer<>(IMAGE)\n .withExposedPorts(PORT)\n\t\t.withStartupTimeout(Duration.ofSeconds(120));","type":"text"}]},{"type":"paragraph","content":[{"text":"The TestContainers library will automatically create and destroy a container for each test that runs. This can slow down the execution of tests. Additionally, each new container will run without any pre-installed pipelines or additional libraries, so these will need to be installed in the test process itself. There are several strategies to address this:","type":"text"}]},{"type":"bulletList","content":[{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"On one hand, you can have an image configured with all the required pipelines and libraries installed. This is as simple as creating an image from an existing container that has been configured using the command ","type":"text"},{"text":"docker commit ","type":"text","marks":[{"type":"code"}]},{"text":".","type":"text"}]}]},{"type":"listItem","content":[{"type":"paragraph","content":[{"text":"On the other hand, you can have a ready and configured Dataflow instance in an environment to be used during test execution.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Testing with JUnit","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Creating a Dataflow Client","type":"text"}]},{"type":"paragraph","content":[{"text":"The first step needed to run the tests is to create an instance of the Dataflow client.","type":"text"}]},{"type":"codeBlock","content":[{"text":"ApiClient apiClient = new ApiClient(authType);\napiClient.setUserAgent(\"SDC CLI\");\napiClient.setBasePath(getUrl(port) + \"/rest\");\napiClient.setUsername(user);\napiClient.setPassword(password);\n\nSimpleModule module = new SimpleModule();\nmodule.addDeserializer(FieldJson.class, new FieldDeserializer());\napiClient.getJson().getMapper().registerModule(module);","type":"text"}]},{"type":"paragraph","content":[{"text":"In addition to creating the client, you can see in the code that a Deserializer is being registered. This is not mandatory. In this case, it is used to leverage the Dataflow Field classes to facilitate querying the values of the records, but the values could also be checked using the original JSON format.","type":"text"}]},{"type":"panel","attrs":{"panelType":"note"},"content":[{"type":"paragraph","content":[{"text":"The code for creating a client is encapsulated in a utility class so that it can be reused across multiple tests. ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/blob/main/test-examples/pipelines/3.23.0/src/test/java/com/minsait/onesaitplatform/dataflow/examples/ApiUtils.java"}}]},{"text":".","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Import the pipeline to be tested.","type":"text"}]},{"type":"paragraph","content":[{"text":"The next step will be to import the pipeline to be tested.","type":"text"}]},{"type":"codeBlock","content":[{"text":"TypeRef returnType = new TypeRef() {};\nPipelineEnvelopeJson pipelineEnvelopeJson = apiClient.getJson().deserialize(new File(getPipelinePath(filename)), returnType);\n\nStoreApi storeApi = new StoreApi(apiClient);\nPipelineEnvelopeJson importPipeline = storeApi.importPipeline(\n\t\tpipelineId, \n\t\trev, \n\t\tfalse, \n\t\tfalse, \n\t\tfalse, \n\t\tfalse, \n\t\tpipelineEnvelopeJson);","type":"text"}]},{"type":"paragraph","content":[{"text":"Ver la sección ","type":"text"},{"text":"Configurar Dataflow","type":"text","marks":[{"type":"link","attrs":{"href":"https://onesaitplatform.atlassian.net/wiki/spaces/DOC/pages/3850469377#Configuraci%C3%B3n-de-la-instancia-de-Dataflow"}}]},{"text":", para ver cómo configurar escenarios más complejos.","type":"text"}]},{"type":"paragraph","content":[{"text":"See Configuring Dataflow to learn how to set up more complex scenarios.","type":"text"}]},{"type":"panel","attrs":{"panelType":"note"},"content":[{"type":"paragraph","content":[{"text":"The code for creating a pipeline is encapsulated in a utility class so that it can be reused across multiple tests. ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/blob/main/test-examples/pipelines/3.23.0/src/test/java/com/minsait/onesaitplatform/dataflow/examples/ApiUtils.java"}}]},{"text":".","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Run a pipeline preview.","type":"text"}]},{"type":"paragraph","content":[{"text":"Once the pipeline is in the Dataflow instance, we will run a preview.","type":"text"}]},{"type":"paragraph","content":[{"text":"The following code snippet does exactly that. It runs a preview, waits for the result to be ready, and retrieves the preview data.","type":"text"}]},{"type":"paragraph"},{"type":"codeBlock","content":[{"text":"PreviewApi previewApi = new PreviewApi(apiClient);\n\nPreviewInfoJson previewOutput = previewApi.previewWithOverride(\n\t\tpipelineId, \n\t\tCollections.emptyList(), \n\t\trev, \n\t\tbatchSize, \n\t\tbatches, \n\t\twriteTarget, \n\t\tnull, \n\t\tnull);\nString previewerId = previewOutput.getPreviewerId();\n\nPreviewStatus status = null;\ndo {\n\tPreviewInfoJson previewStatus = previewApi.getPreviewStatus(pipelineId, previewerId);\n\tTimeUnit.SECONDS.sleep(5);\n\tStatusEnum statusEnum = previewStatus.getStatus();\n\tstatus = PreviewStatus.valueOf(statusEnum.name());\n} while (status.isActive());\n\nassertTrue(PreviewStatus.FINISHED.equals(status));\n\nPreviewOutputJson previewData = previewApi.getPreviewData(pipelineId, previewerId); ","type":"text"}]},{"type":"panel","attrs":{"panelType":"note"},"content":[{"type":"paragraph","content":[{"text":"The code for running a pipeline preview is encapsulated in a utility class so that it can be reused across multiple tests. ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/blob/main/test-examples/pipelines/3.23.0/src/test/java/com/minsait/onesaitplatform/dataflow/examples/ApiUtils.java"}}]},{"text":".","type":"text"}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Validating the Preview Result","type":"text"}]},{"type":"paragraph","content":[{"text":"Once the preview has been run, we can analyze the results to determine if they are as expected. ","type":"text"}]},{"type":"paragraph","content":[{"text":"Essentially, this involves checking batch by batch if the records in the stages of interest have the expected values. In other words, we verify that the expected data is obtained and that the expected transformations are performed before sending to destination.","type":"text"}]},{"type":"codeBlock","content":[{"text":"List> batchesOutput = previewData.getBatchesOutput();\n\n// This test only use one batch.\nList stageOutputs = batchesOutput.get(0);\n\nfor (StageOutputJson stageOutput : stageOutputs) {\n\t// Assert status of each stage for the selected batch.\n\tswitch (stageOutput.getInstanceName()) {\n\tcase \"DevRawDataSource_01\":\n\t\tMap> outputs = stageOutput.getOutput();\n\t\tfor (List records : outputs.values()) {\n\t\t\tfor (RecordJson record : records) {\n\n\t\t\t\t// If needed, test record header values\n\t\t\t\t// Dummy test as example\n\t\t\t\tHeaderJson header = record.getHeader();\n\t\t\t\tassertTrue(header.getSourceId().equals(\"rawData::0\"));\n\n\t\t\t\tField field = ApiUtils.getFieldFromRecord(record, apiClient.getJson());\n\t\t\t\tMap root = field.getValueAsMap();\n\n\t\t\t\t// We can test any data of the records, including the data type.\n\t\t\t\tassertTrue(root.get(\"value1\").getType().equals(Field.Type.STRING));\n\t\t\t\tassertTrue(root.get(\"value1\").getValueAsString().equals(\"abc\"));\n\t\t\t\tassertTrue(root.get(\"value2\").getType().equals(Field.Type.STRING));\n\t\t\t\tassertTrue(root.get(\"value2\").getValueAsString().equals(\"xyz\"));\n\t\t\t\tassertTrue(root.get(\"value3\").getType().equals(Field.Type.STRING));\n\t\t\t\tassertTrue(root.get(\"value3\").getValueAsString().equals(\"lmn\"));\n\t\t\t}\n\t\t}\n\t\tbreak;\n\tcase \"Trash_01\":\n\t\t// trash does not have output\n\t\tbreak;\n\tdefault:\n\t\t// stage not tested\n\t\tbreak;\n\t}\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"The example we have included only generates one batch of data. In most cases, this will be sufficient. If tests with multiple batches are required, it is advisable to use a manageable number of them to simplify the validation of results. For each batch obtained, there will be a set of records at each stage of the pipeline. The test should verify that the values correspond to what is expected at each stage.","type":"text"}]},{"type":"paragraph","content":[{"text":"As we can see in the following example, the pipeline shows a visualization of the data in the output of each stage.","type":"text"}]},{"type":"mediaSingle","attrs":{"layout":"center"},"content":[{"type":"media","attrs":{"width":946,"id":"b1573fdc-f5b8-4312-92e9-238b31cf197f","collection":"contentId-4357357569","type":"file","height":607}},{"type":"caption","content":[{"text":"Pipeline de ejemplo para tests","type":"text"}]}]},{"type":"paragraph","content":[{"text":"The names ","type":"text"},{"text":"DevRawDataSource_01","type":"text","marks":[{"type":"code"}]},{"text":" and ","type":"text"},{"text":"Trash_01","type":"text","marks":[{"type":"code"}]},{"text":" used in the case block of the previous code snippet to determine the execution stage can be seen in the information of each stage in the pipeline editor:","type":"text"}]},{"type":"mediaSingle","attrs":{"layout":"center"},"content":[{"type":"media","attrs":{"width":946,"id":"3c02653d-2acd-4311-a9e3-bbe6985fe180","collection":"contentId-4357357569","type":"file","height":607}},{"type":"caption","content":[{"text":"Ejemplo de nombre de etapa en un pipeline","type":"text"}]}]}],"version":1}

Browser not supported