Haris Habul

Introduction Building on our previous blog Building ETL Pipeline in Snowflake, we will cover different approaches to automated testing for ETL Pipelines in Snowflake. For these kinds of solutions, QA Engineers should use their creativity to implement useful test cases as much as possible. There are  various approaches, and we will go through the following: tasks verification, end-to-end testing, IQR analysis and data quality verification. Tasks verification This part of the automated testing helps the verification of E and T of ETL parts. If we examine the example from the previously mentioned blog, we can see the use of tasks in Snowflake ETL pipelines. This allows us to develop an automated test case that runs a Snowflake query to identify failed tasks within the last 24 hours or a specified time range, depending on your needs. Here is the example written in Node JS with Jest (or any other test runner that you prefer, such as Jasmine, etc.): 1. Initialise repository and install Jest with commands: npm init npm i jest 2. Create Snowflake Connector Util class: const Snowflake = require('snowflake-promise').Snowflake const connectionProperties = { account: 'TEST_ACCOUNT', username: 'TEST_USERNAME', role: 'TEST_ROLE', database: 'TEST_DB', schema: 'TEST_SCHEMA', warehouse: 'TEST_WAREHOUSE', authenticator: 'SNOWFLAKE_JWT', privateKeyPath: process.env.SF_TEST_PRIVATE_KEY, privateKeyPass: process.env.SF_TEST_PRIVATE_KEY_PASS, } const taskHistoryTable = `${connectionProperties.database}.INFORMATION_SCHEMA.TASK_HISTORY` class SnowflakeConnectorUtil { async snowflakePool() { return new Snowflake(connectionProperties) } async createPool() { const pool = await this.snowflakePool() await pool.connect() return pool } async getSnowflakeFailedTasksInLast24Hours() { return `SELECT * FROM ( SELECT distinct name, first_value(completed_time) ignore nulls over (partition by name order by completed_time desc) as last_completed_time, first_value(state) ignore nulls over (partition by name order by completed_time desc) as last_state FROM db.schema.task_history ) WHERE last_state IN ('FAILED', 'CANCELLED', 'FAILED_AND_AUTO_SUSPENDED') AND last_completed_time >= DATEADD(DAY, -1, GETDATE());` } } module.exports = SnowflakeConnectorUtil 3. Create a test file. For your test case use previously created Snowflake Connector Util: const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtil') const snowflakeConnectorUtil = new SnowflakeConnectorUtil() describe('Tasks verification', () => { let pool beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() }, 10000) it('Daily verification tasks', async () => { const query = await snowflakeConnectorUtil.getSnowflakeFailedTasksInLast24Hours() const records = await pool.execute(query) records.map(failedTask => { console.log( `Failed task + ${failedTask.NAME}, error code: ${failedTask.ERROR_CODE}, error message: ${failedTask.ERROR_MESSAGE}` ) }) expect(records).toEqual([]) }, 10000) afterAll(async () => { await pool.destroy() }) }) With this approach, we could have the following benefits: Early detection of ETL failures: issues are caught long before they affect downstream Improved data reliability: leads to higher confidence in the data across dashboards and reports Reduced manual effort: engineers do not have to waste time on a manual daily tasks check Proactive incident management: integration with the communication tools such as Slack, Teams, email, etc. allows proactive response and prevents cases in which users notify problems before the engineers Ability to have an overview picture of ETL health over time Improved SLA compliance: Early automated detection helps teams to maintain data availability and freshness commitments in line with business SLAs End-to-end In this part, we are covering the whole process of ETL by publishing some test data and verifying that the data is being processed as it is required. We will use an ETL pipeline example from a previous blog and develop our end-to-end test automation example based on that. In the previous blog, the main source of the data is the Kafka topic. Therefore, our first test step in some of the test environments would be to prepare the test data by publishing it to the Kafka topic. We will do that with Node.js, but of course, feel free to use any other programming language or tool with which you are comfortable. npm install kafka npm install jest or npm install jasmine Next, we need to produce the Data on our ETL data source that is in this example Kafka. So we will use the Kafka producer: const { Kafka } = require('kafkajs'); class KafkaProducer { kafka; producer; constructor() { this.kafka = new Kafka({ clientId: process.env.YOUR_CLIENT_ID, brokers: process.env.KAFKA_HOST }); this.producer = this.kafka.producer(); } async sendMessage(message) { try { await this.producer.connect(); await this.producer.send({ topic: process.env.YOUR_KAFKA_TOPIC, messages: [ { value: message }, ], }); await this.producer.disconnect() } catch (error) { console.log(error); } } } module.exports = KafkaProducer; Note: Although this data source issue could be resolved on the ETL side by various services, such as AWS S3 buckets and RabbitMQ, for this blog, we decided to test a data pipeline that uses Kafka as the data source. We will still need the Snowflake Connector Util class, but with getters for Snowflake queries for staging and the final table. It should look something as the following: const Snowflake = require('snowflake-promise').Snowflake const connectionProperties = { account: 'TEST_ACCOUNT', username: 'TEST_USERNAME', role: 'TEST_ROLE', database: 'TEST_DB', schema: 'TEST_SCHEMA', warehouse: 'TEST_WAREHOUSE', authenticator: 'SNOWFLAKE_JWT', privateKeyPath: process.env.SF_TEST_PRIVATE_KEY, privateKeyPass: process.env.SF_TEST_PRIVATE_KEY_PASS, } const taskHistoryTable = `${connectionProperties.database}.INFORMATION_SCHEMA.TASK_HISTORY` const packageStagingTable = `${connectionProperties.database}.PACKAGE_STAGING` const packageFinalTable = `${connectionProperties.database}.PACKAGE_FINAL` class SnowflakeConnectorUtil { async snowflakePool() { return new Snowflake(connectionProperties) } async createPool() { let pool = await this.snowflakePool() await pool.connect() return pool } async getSnowflakeFailedTasksInLast24Hours() { return `SELECT * FROM ( SELECT distinct name, first_value(completed_time) ignore nulls over (partition by name order by completed_time desc) as last_completed_time, first_value(state) ignore nulls over (partition by name order by completed_time desc) as last_state FROM ${taskHistoryTable} ) WHERE last_state IN ('FAILED', 'CANCELLED', 'FAILED_AND_AUTO_SUSPENDED') AND last_completed_time >= DATEADD(DAY, -1, GETDATE());` } checkNullValuesQueries = [ { description: 'Package Sender or Recipient ID', query: `select * from dbo.schema.packages_final where sender_id is null or getter_id is null`, expected: 0, }, { description: 'User ID', query: `select * from dbo.schema.user_request where user_id is null`, expected: 0, } ] getPackageStagingTableQuery(packageId) { return `select * from ${packageStagingTable} where package_id = '${packageId}'` } getPackageFinalTableQuery(packageId) { return `select * from ${packageFinalTable} where package_id = '${packageId}'` } } module.exports = SnowflakeConnectorUtil Then we just need to use everything mentioned in the test case: import { v6 as uuidv6 } from 'uuid'; const Kafka = require('./KafkaProducer') const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtill') const kafka = new Kafka() const snowflakeConnectorUtil = new SnowflakeConnectorUtil() describe('E2E test example', () => { let pool let packageId beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() packageId = uuidv6() // producing test data await kafka.sendMessage( `{ 'packageId', ${packageId}, 'senderId': '1', 'recipientId': '2', }`) }) it ('Verify staging table', async () => { const packagesStagingQuery = snowflakeConnectorUtil.getPackageStagingTableQuery(packageId) const packagesStaging = await pool.execute(packagesStagingQuery) expect(packagesStaging).toBeGreaterThan(0) expect(packagesStaging[0].senderFirstName).toBe('John') expect(packagesStaging[0].senderLastName).toBe('Doe') expect(packagesStaging[0].recipientFirstName).toBe('Jane') expect(packagesStaging[0].recipientLastName).toBe('Doe') }) it ('Verify final table', async () => { const packagesFinalQuery = snowflakeConnectorUtil.getPackageFinalTableQuery(packageId) const packagesFinal = await pool.execute(packagesFinalQuery) expect(packagesFinal).toBeGreaterThan(0) expect(packagesFinal[0].senderFirstName).toBe('John') expect(packagesFinal[0].senderLastName).toBe('Doe') expect(packagesFinal[0].recipientFirstName).toBe('Jane') expect(packagesFinal[0].recipientLastName).toBe('Doe') }) afterAll(async () => { await pool.destroy() }) }) By combining various data, we can create multiple test scenarios and achieve confidence in our pipeline after the team makes changes. This approach leads us to the following benefits: Ensuring data integrity across the pipeline Validates transformations and business logic Detects schema or contract breaks early Builds confidence in data freshness and accuracy Reduces manual verification effort IQR analysis You can learn more about the IQR on the following blog IQR in Automation Testing: Unleashing the Data Analytics Potential - Atlantbh Sarajevo. In the context of testing Snowflake ETL pipelines, we can implement this on the appropriate dataset. Test cases based on this approach will provide us more confidence in our ETL process. Significant changes in our data can point to potential issues. A significant increase might indicate that our user base is growing, so we should be prepared for that. A sudden drop could indicate an issue with data processing. For instance, let us say we want to find outliers in the data set for the final table names, which will be called “package_final”. A common use case would be to identify outliers, such as a significant drop or increase in data over the last 30 days. Since it is not valid in many cases to compare Mondays with Saturdays, we should create an appropriate query to retrieve the last 30 Mondays. We will add this query to the SnowflakeConnectorUtil class: getLast30Weeks() { return `WITH OrderedRows AS ( SELECT DATE(created_ts_pst) created_date, count(*) as count, ROW_NUMBER() OVER (ORDER BY created_date) AS row_num, FROM ${packageFinalTable} WHERE created_ts_pst BETWEEN DATEADD(DAY, -210, current_date) AND current_date GROUP BY created_date Order by created_date desc ) SELECT * FROM OrderedRows WHERE row_num % 7 = 0;` } We will wrap our IQR analyze logic into shared service: const iqrAnalyze = async data => { const dataSetValues = data.map(value => value.COUNT) const dataSetSorted = dataSetValues.sort((a, b) => a - b) const quartile1 = dataSetSorted[Math.floor(dataSetSorted.length / 4)] const quartile3 = dataSetSorted[Math.ceil(dataSetSorted.length * (3 / 4))] const iqr = quartile3 - quartile1 const lowerBound = quartile1 - 1.5 * iqr const upperBound = quartile3 + 1.5 * iqr const outliers = dataSetSorted.filter( value => (value < lowerBound || value > upperBound) && value == dataSet[0].COUNT ) expect(outliers.length).toBe(0) } module.exports = { iqrAnalyze: iqrAnalyze, } Next we test it by using data from Snowflake in our test file: const SnowflakeConnectorUtil = require('./db/SnowflakeConnectorUtil') const snowflakeConnectorUtil = new SnowflakeConnectorUtil() const iqrAnalyzer = require('./services/iqrAnalyze') describe('IQR test suite', () => { let pool beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() }, 10000) it('Packages test', async () => { const query = await snowflakeConnectorUtil.getLast30Weeks() const data = await pool.execute(query) await iqrAnalyzer.iqrAnalyze(data) }, 10000) afterAll(async () => { await pool.destroy() }) }) The main advantages of incorporating IQR analysis into automated testing include: Early data anomaly detection Better confidence in data quality Data observability support that leads to better confidence in ETL Data Quality verification In this section, we should verify the quality of data that has undergone the pipeline transformations. One of the most common mistakes in ETL processes is invalid data in columns such as NULL values. To find these defects, we could use the following approach. First, let us define an array of queries, each with a description and the expected number of records that should be returned with the query: checkNullValuesQueries = [ { description: 'Package Sender or Recipient ID', query: `select * from ${packageFinalTable} where sender_id is null or getter_id is null`, expected: 0, }, { description: 'User ID', query: `select * from dbo.schema.user_request where user_id is null`, expected: 0, } ] Next, in the test case we go through the array of queries, execute each one and assert the expected records length: const SnowflakeConnectorUtil = require('/db/SnowflakeConnectorUtil') const snowflakeConnectorUtil = new SnowflakeConnectorUtil() describe('Data quality verification', () => { let pool beforeAll(async () => { pool = await snowflakeConnectorUtil.createPool() }, 10000) snowflakeConnectorUtil.checkNullValuesQueries.map(checkNullValueQuery => { it(`Verify ${checkNullValueQuery.description}`, async () => { const nullFieldsRecords = await pool.execute(checkNullValueQuery.query) expect(nullFieldsRecords.length) .withContext(JSON.stringify(nullFieldsRecords)) .toBe(checkNullValueQuery.expected) await sleep(10000) }, 100000) }) afterAll(async () => { await pool.destroy() }) }) This check strengthens the overall data governance framework by ensuring the standardized quality of the data, guarantees the data completeness, and supports compliance with business and regulatory requests. CI/CD For the CI/CD part, we will use Jenkins; therefore, the first step will be setting up the Jenkins server. The easiest way would be to use Docker and first pull the latest Jenkins Docker image, and run Jenkins on a Docker container by running the following scripts in the terminal: docker pull jenkins/jenkins:lts docker run -d --name jenkins-lts -p 8080:8080 -p 50000:50000 -v jenkins_home:/var/jenkins_home jenkins/jenkins:lts Next, go to your localhost:8080 (or any other port that you have configured) and configure the admin account. Since we are developing our test cases in Node JS, we need to install the appropriate Jenkins plugin by navigating to Manage Jenkins > Manage Plugins > Available plugins. Next, find “NodeJS Plugin” and click on the install button. After the plugin installation, we need to configure the Node JS version by going to Manage Jenkins > Tools, finding NodeJS installations, and clicking on Add “NodeJS”. This will open options for choosing the appropriate Node version that you need, as well as additional packages that you want to have on your Jenkins. Now we will create our Jenkins files in the git repository. The Jenkins file for an end-to-end test case could be such as the following: pipeline { agent any tools { nodejs 'NodeJS 18' } triggers { cron('H 0 * * *') // Run once a day at midnight } stages { stage('Checkout') { steps { checkout scm } } stage('Install Dependencies') { steps { sh 'npm install' } } stage('Run Test') { steps { sh 'npm test e2e.test.js' } } } } Now let us back to the Jenkins dashboard. Click on the “+ New Item” button and then enter the test name for instance: “E2E Snowflake test”, choose “Pipeline” and click the “Ok” button. Afterwards, add description, select the “Github project” checkbox, and paste your remote repository link. In the pipeline section, choose 'Pipeline script from SCM' and set your GitHub repository link, along with other details such as credentials, branch name, and script file path. Then, save the details and back to the main dashboard Jenkins where we should see our job: We can now manually trigger our test or wait for the cron schedule we set up earlier in the script by specifying the appropriate CRON. Next steps include adding other test cases and setting up Jenkins on a server of your choice. You could read more about Jenkins on our other blogs: Jenkins blogs. Reporting For the reporting part, we will introduce the Allure reporting library. First, we need to install the needed dependency in the Node JS repository with the command: npm i -D jest-allure2-reporter allure-commandline Then we should define jest.config.js as: module.exports = { reporters: [ 'default', [require.resolve('jest-allure2-reporter'), { resultsDir: 'allure-results' }], ], testEnvironment: require.resolve('jest-allure2-reporter/environment-node'), testTimeout: 30000, }; In package.json modify change scripts config to be such as: "scripts": { "test:ci": "jest --runInBand" }, We are using runInBand, which guarantees running multiple test cases as worker processes sequentially (one by one). If we do not use this, the test cases will run in parallel, and we could face race conditions while writing them to the allure reports and have a risk of corrupted reports. Then, inside your tests, add allure functions for attaching metadata to every test step that you will later need as an explanation in formatting your report. They could be a story, description, step, attachment, etc. Here is an example so you could use as a reference to your other steps: it ('Verify staging table', async () => { allure.story("Staging Table Check"); allure.description("Verify that staging table data matches expected schema"); allure.step("Querying and verifying staging table...", async () => { const packagesStagingQuery = snowflakeConnectorUtil.getPackageStagingTableQuery(packageId) const packagesStaging = await pool.execute(packagesStagingQuery) expect(packagesStaging).toBeGreaterThan(0) expect(packagesStaging[0].senderFirstName).toBe('John') expect(packagesStaging[0].senderLastName).toBe('Doe') expect(packagesStaging[0].recipientFirstName).toBe('Jane') expect(packagesStaging[0].recipientLastName).toBe('Doe') }); allure.attachment("Staging query result", "SELECT * FROM staging_table", "text/plain"); }) Note: Pay attention to the “allure.step”, it takes a string and also a callback as arguments, so here we are passing our test step logic as a callback. After running the test locally, we should get the allure-results folder in our repository with our report file with some additional details. Let us now include this in our CI/CD part on Jenkins. First, we need to install Allure Jenkins Plugin in Manage Jenkins > Plugins Then we need to install Allure Commandline in Manage Jenkins > Tools Once that is done, let us modify the Jenkins script in our Git repository so we could add the reporting part at the bottom. It should be such as the following: pipeline { agent any tools { nodejs 'NodeJS 18' } triggers { cron('H 0 * * *') // Run once a day at midnight } stages { stage('Checkout') { steps { checkout scm } } stage('Install Dependencies') { steps { sh 'npm install' } } stage('Run Test') { steps { sh 'npm run test:ci e2e.test.js' } } } post { always { archiveArtifacts artifacts: 'allure-results/**', allowEmptyArchive: true } success { allure([ includeProperties: true, jdk: '', reportBuildPolicy: 'ALWAYS', results: [[path: 'allure-results']] ]) } unsuccessful { allure([ includeProperties: true, jdk: '', reportBuildPolicy: 'ALWAYS', results: [[path: 'allure-results']] ]) } } } After running our Jenkins job we should get the following artifacts: If you download allure-report.zip you should find the index.html file as allure report. If you open it in a browser it should look like the following: Of course, you can modify everything by adopting the mentioned allure functions as you need. Note: Maybe your browser will block opening these metrics because of CORS but you could avoid that by running simple local static server from allure-report folder with python or some other technology by running command: python3 -m http.server 8081 If you open http://localhost:8081/ you should see the allure report without CORS errors. Finally, you could add a Jenkins step that sends an allure report as an email to the stakeholders or to make Slack notification. It could be done by shell script or Jenkins plugin “Email Extension. Conclusion We have explored various approaches to automated testing for ETL pipelines built in Snowflake, orchestrated through Jenkins CI/CD, and integrated with Allure reporting. All these approaches can help us detect issues early, maintain high data reliability, and continuously improve pipeline performance. Automated testing of the ETL pipeline in Snowflake has transformed our ETL validation from a manual process to a proactive and self-monitoring system.  

Introduction to AWS Step Function AWS Step Function is an AWS resource that offers serverless function orchestration making it easy to sequence multiple AWS resources into one business workflow. This tool allows us to create and run many checkpointed and event-driven workflows that maintain the application state, ideal for projects such as data pipelines and many others. In this blog, we will explain the concept of development and application of the Step Function, with test automation integrated into its workflow. Pricing of AWS Step Function AWS Step Function pricing works based on ‘you pay what you use’. We have two types of workflows: Standard Workflows and Express Workflows. By default, Step Function uses Standard Workflow, and its price is $0.000025 per state transition. The pricing for Express Workflow is $0.000001 per request. In this blog, we will use the default option. Step Function states AWS Step Function provides a set of states that you can use to build workflows. Each state represents a specific action or behavior within the workflow. Here are some of the commonly used Step Function states: Task State: Represents a single unit of work in a Step Function workflow. It can be used to invoke AWS Lambda functions, run ECS tasks, or perform other activities. Choice State: Allows you to define conditional logic within your workflow. It evaluates a condition and transitions to different states based on the result. Wait State: Delays the execution of the workflow for a specific period. Pass State:  Passes its input to its output without performing work. It is useful when constructing and debugging state machines. Parallel State: Allows us to run multiple branches of execution in parallel. Within the parallel state every branch can be separated. Succeed State: Ends execution as successfully processed. Fail State: Ends execution as failure. It also provides more ways to find out the root cause of errors. Additionally, it could provide us with the ‘Catch’ block if necessary. Map State: You can think of it like a for-each loop for Step Function. It allows us to iterate over an array/list of items and perform the same workflow steps for every item in the array. Let us learn by the example: Demo project description: In this example, we will create a simple demo software with the Step Function, Lambda, DynamoDB, GitHub Actions, and  Serverless framework for easier Step Function creation. In the DynamoDB, we will create one table called ‘messages’. This table is supposed to receive data from other imaginary services. Message record should contain the following:  ID,  First name (fName),  Last name (lName),  Region (US/EU), Points, Column representing flag for processed messages (is_processed) Our Step function should contain a workflow that takes all of the unprocessed messages, separates them by the region of the US and EU users, and puts those messages into the other appropriate tables: US_users and EU_users. Step function implementation: We will use the Serverless framework to easily implement the Step Function, its Lambdas and IAM roles. The Serverless framework offers infrastructure as a code implementation and deployment for the AWS Step function. It uses Lambdas and additional necessary services that we need in the state machine. We will not go in-depth with the Serverless framework, but if you would like to know more, you can find it on their official website. Follow these steps to start with the Serverless framework: Create empty folder 2. Instantiate NPM with the ‘npm init’ command 3. Install serverless framework via NPM ‘npm install -g serverless’ 4. Add the following dependencies that we will further use in package.json: {   "name": "sfn-blog",   "dependencies": {     "@aws-sdk/client-dynamodb": "^3.405.0",     "@aws-sdk/client-s3": "^3.405.0",     "axios": "^1.4.0",     "jest": "^29.4.1",     "jest-junit": "^15.0.0"   },   "version": "1.0.0",   "description": "",   "main": "index.js",   "directories": {     "test": "test"   },   "scripts": {     "test": "jest"   },   "author": "",   "license": "ISC",   "devDependencies": {     "@types/jest": "^29.2.5",     "serverless-step-functions": "^3.13.1"   } }      5. Run ‘npm install’      6. Create serverless.yml which will be used for infrastructure as a code      7. Create handler.js for lambda functions written in Node JS Note: Do not forget to add your AWS credentials (access key and secret key). If you are on a MAC machine, you can do this by running the following command: sls config credentials --provider aws --key <YOUR_ACCESS_KEY> --secret <YOUR_SECRET_KEY>      8. Create DynamoDB table ‘messages” with ID type String as a Partition key:      9. Choose Customize Settings, On-demand capacity mode, and click  ‘Create table’ at the end of the page.            10. In the same way, create two more tables: US_users_events and EU_users_events.      11. In the serverless yml file, we will define our needed Step Function workflow: service: sfn-blog frameworkVersion: "3" # Define  cloud provider settings and IAM roles needed for our SF to work provider:   name: aws   runtime: nodejs18.x   region: eu-central-1   environment: ${file(env.json)}   iamRoleStatements:     - Effect: Allow       Action: dynamodb:*       Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/messages     - Effect: Allow       Action: dynamodb:*       Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/US_users_events     - Effect: Allow       Action: dynamodb:*       Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/EU_users_events     - Effect: Allow       Action: s3:*       Resource: arn:aws:s3:::sfn-blog plugins:   - serverless-step-functions # Define a path to the AWS Lambda Functions functions:   FetchFromDynamoDBState:     handler: handler.FetchFromDynamoDBState   ProcessUSUsersEvents:     handler: handler.ProcessUSUsersEvents   ProcessEUUsersEvents:     handler: handler.ProcessEUUsersEvents   InsertMessageDynamoDBTest:     handler: handler.InsertMessageDynamoDBTest   VerifyStepFunctionOutcomeTest:     handler: handler.VerifyStepFunctionOutcomeTest # Define Step Function and its state machine stepFunctions:   stateMachines:     proceedRewards:       name: proceedRewards       definition:         StartAt: InsertMessageDynamoDBTest         States:           InsertMessageDynamoDBTest:             Type: Task             Resource:               Fn::GetAtt: [InsertMessageDynamoDBTest, Arn]             ResultPath: "$.response"             Next: WaitState           WaitState:             Type: Wait             Seconds: 200             Next: FetchFromDynamoDBState           FetchFromDynamoDBState:             Type: Task             Resource:               Fn::GetAtt: [FetchFromDynamoDBState, Arn]             ResultPath: "$.items"             Next: ProcessDataState           ProcessDataState:             Type: Map             ItemsPath: "$.items.items"             ResultPath: "$.mappedData"             MaxConcurrency: 2             Iterator:               StartAt: ProceedChoiceState               States:                 ProceedChoiceState:                   Type: Choice                   Choices:                     - Variable: $.region.S                       StringEquals: "US"                       Next: ProcessUSUsersEventsState                     - Variable: $.region.S                       StringEquals: "EU"                       Next: ProcessEUUsersEventsState                   Default: DefaultState                 ProcessUSUsersEventsState:                   Type: Task                   Resource:                     Fn::GetAtt: [ProcessUSUsersEvents, Arn]                   End: true                 ProcessEUUsersEventsState:                   Type: Task                   Resource:                     Fn::GetAtt: [ProcessEUUsersEvents, Arn]                   End: true                 DefaultState:                   Type: Fail                   Cause: 'Invalid region value.'                   Error: 'InvalidRegionError'             Next: VerifyStepFunctionOutcomeTest           VerifyStepFunctionOutcomeTest:             Type: Task             Resource:               Fn::GetAtt: [VerifyStepFunctionOutcomeTest, Arn]             End: true       12. Include the following Lambda functions in the handler.js file (if you are wondering why we are using these keys ‘S’, ‘N’ and ‘BOOL’ in params for DynamoDB client it is because the library itself needs that specification of the data types, ‘S’ for string, ‘N’ for number, ‘BOOL’ for boolean and etc.): const { DynamoDBClient, PutItemCommand, ScanCommand, UpdateItemCommand } = require('@aws-sdk/client-dynamodb'); const dynamoDbClient = new DynamoDBClient({ region: 'eu-central-1' }); module.exports.FetchFromDynamoDBState = async () => { try { const params = {             TableName: 'messages',             FilterExpression: 'is_processed = :processed',             ExpressionAttributeValues: { ':processed': { BOOL: false },             }         }; const scanCommand = new ScanCommand(params); let result = await dynamoDbClient.send(scanCommand); const items = result.Items; return {             items         };     } catch (error) { throw error;     } } module.exports.ProcessUSUsersEvents = async (item) => { return await processUsersEvents(item, 'US_users_events'); } module.exports.ProcessEUUsersEvents = async (item) => { return await processUsersEvents(item, 'EU_users_events'); } const processUsersEvents = async (item, tableName) => { let dataProcessed = false; try { await processItem(item, tableName); await updateIsProcessedColumn(item.ID.S);         dataProcessed = true;     } catch (error) { return dataProcessed;     } return dataProcessed; } const processItem = async (item, tableName) => { const eventItem = {         ID: { S: item.ID.S },         fName: { S: item.fName.S },         lName: { S: item.lName.S },         points: { N: item.points.N },     } const params = {         TableName: tableName,         Item: eventItem     }; const command = new PutItemCommand(params); await dynamoDbClient.send(command); } const updateIsProcessedColumn = async (itemId) => { const paramsForUpdate = {         TableName: 'messages',         Key: {             ID: { S: itemId },             UpdateExpression: 'SET is_processed = :value',             ExpressionAttributeValues: { ':value': { BOOL: true } },             ReturnValues: 'UPDATED_NEW'         }     }; const updateItemCommand = new UpdateItemCommand(paramsForUpdate); await dynamoDbClient.send(updateItemCommand); }      13. Run command: ‘sls deploy’ to  the Step Function: Test Automation In real-world scenarios, this Step Function can be scheduled to run a workflow at regular intervals, such as every hour, so we have to split the test automation into two parts: data insertion and verification. Firstly, we will create a test automation script to insert a message with test data to the DynamoDB table ‘messages’ and save this message as a test job artifact to an S3 bucket. AWS S3 is a cloud-based object storage service provided by Amazon Web Services (AWS) (you can learn more here) that we need in this case to store our test artifacts that will be later used from the second script that will do the verification part. You can think of the S3 bucket as a virtual container where our test data will be kept, but of course, you can use other services for this purpose. The good alternatives could be saving the data as the GitHub artifacts, or if you choose to work this with Jenkins, you can have the same concept and also save the data on your own servers. It is also important to explain that we can not save the artifacts locally on AWS Lambdas because they are serverless computing services that are packaged into isolated environments (containers) where it runs on a multi-tenant cluster of machines managed by AWS. We will use Node JS with JEST for this example. const { DynamoDBClient, PutItemCommand } = require('@aws-sdk/client-dynamodb'); const dynamoDbClient = new DynamoDBClient({ region: 'eu-central-1' }); const S3Client = require('./S3Client'); describe('Adding message to the DynamoDB messages table', () => { let item;     beforeAll(async () => {         item = {             ID: { S: new Date().toISOString() },             fName: { S: 'Test_First_Name' },             lName: { S: 'Test_Last_Name' },             points: { N: '500' },             is_processed: { BOOL: false },             region: { S: 'US' }         };     });     it('Adding message', async () => { const params = {             TableName: 'messages',             Item: item         }; try { const command = new PutItemCommand(params); await dynamoDbClient.send(command); console.log('Data inserted successfully.');         } catch (error) { console.log('Error inserting data:', error);         }     }, 10000);     it('Save message to S3 as a job artifact ', async () => { const s3Client = new S3Client(); const bucket = 'sfn-blog'; const key = 'test-message.json'; await s3Client.removeObject(bucket, key); await s3Client.uploadFileToS3(             bucket,             key, JSON.stringify(item), 'application/json'         );     }, 20000); }); Secondly, we will create a script that should be run after the Step Function finishes to verify that the message has been processed. In this script, we will pull the artifact JSON file from S3 bucket, and verify that this record can be found in the ‘US_users_events’ table in DynamoDB. const { DynamoDBClient, ScanCommand } = require('@aws-sdk/client-dynamodb'); const dynamoDbClient = new DynamoDBClient({ region: 'eu-central-1' }); const S3Client = require('./S3Client'); describe('Verify that message has been processed', () => { let item; const s3Client = new S3Client();     beforeAll(async () => {         item = await s3Client.getObject('sfn-blog', 'test-message.json');     }, 200000);     it('Verify DynamoDB US_users_events', async () => { const params = {             TableName: 'US_users_events'         }; const scanCommand = new ScanCommand(params); const response = await dynamoDbClient.send(scanCommand); const records = response.Items; const testRecord = records.filter(it => it.ID.S == item.ID.S);         expect(testRecord.length).toBe(1);         expect(testRecord[0].fName.S).toBe(item.fName.S);         expect(testRecord[0].lName.S).toBe(item.lName.S);         expect(testRecord[0].points.N).toBe(item.points.N);     }, 20000); }); We also need a helper file for S3 related methods. Create S3Client.js file: const { S3, DeleteObjectCommand, GetObjectCommand, PutObjectCommand } = require('@aws-sdk/client-s3'); class S3Client {     s3Client; constructor() { this.s3Client = new S3({ region: 'eu-central-1' });     } async getObject(bucketName, key) { try { const params = {                 Bucket: bucketName,                 Key: key             }; const getObjectCommand = new GetObjectCommand(params); const response = await this.s3Client.send(getObjectCommand); const bodyStream = response.Body; let data = ""; for await (const chunk of bodyStream) {                 data += chunk;             } const object = JSON.parse(data); return object;         } catch (error) { console.log(error); return null;         }     } async removeObject(bucketName, key) { try { const params = {                 Bucket: bucketName,                 Key: key             }; const removeObjectCommand = new DeleteObjectCommand(params); await this.s3Client.send(removeObjectCommand);         } catch (error) { console.log(error); return null;         }     } async uploadFileToS3(bucketName, key, body, contentType) { try { const params = {                 Bucket: bucketName,                 Key: key,                 Body: body,                 ContentType: contentType             } const putObjectCommand = new PutObjectCommand(params); await this.s3Client.send(putObjectCommand);         } catch (error) { console.log(error); return null;         }     } } module.exports = S3Client; We will now create workflows (think of it as a Jenkins job) for these two scripts with GitHub Actions, but you can use any tool you want, for instance, Jenkins, TeamCity, etc. If you want to know more about GitHub Actions, you can find it on their official website. To define workflows in GitHub Actions, follow these steps: Create a repository on GitHub, Upload test files, including package.json, Click on the ‘Actions’ button Click on the ‘New workflow’ button, Choose ‘Node.js’ workflow, Define your workflow for the first test script as a yml file inside folder ‘.github/workflows’: name: insert-message-DynamoDB on:   workflow_dispatch: env: AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }} AWS_REGION: eu-central-1 jobs:   insert_message_in_DynamoDB:     name: Insert message in DynamoDB     runs-on: ubuntu-latest     steps:     - name: Checkout repository       uses: actions/checkout@v2     - name: Set up Node.js       uses: actions/setup-node@v2       with:         node-version: '18'     - name: Configure AWS credentials       run: aws configure set aws_access_key_id ${{ env.AWS_ACCESS_KEY_ID }} && aws configure set aws_secret_access_key ${{ env.AWS_SECRET_ACCESS_KEY }} && aws configure set region ${{ env.AWS_REGION }}     - name: Install dependencies       run: npm install     - name: Run InsertMessageDynamoDB tests       run: npm run test insertMessageDynamoDB.test.js      7. Do the same for the second script: name: verify-step-function-outcome on:   workflow_dispatch: env: AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }} AWS_REGION: eu-central-1 jobs:   verify_outcome:     name: Verify Step Function Outcome     runs-on: ubuntu-latest     steps:     - name: Checkout repository       uses: actions/checkout@v2     - name: Set up Node.js       uses: actions/setup-node@v2       with:         node-version: '18'     - name: Configure AWS credentials       run: aws configure set aws_access_key_id ${{ env.AWS_ACCESS_KEY_ID }} && aws configure set aws_secret_access_key ${{ env.AWS_SECRET_ACCESS_KEY }} && aws configure set region ${{ env.AWS_REGION }}     - name: Install dependencies       run: npm install     - name: Run VerifyStepFunctionOutcome tests       run: npm run test verifyStepFunctionOutcome.test.js      8. Add InsertMessageDynamoDBTest  as a first step in the Step Function States array, which should invoke the Lambda function that will trigger the Github Action workflow for running the first test script that will insert a message in DynamoDB ‘messages’ table,      9. Add a wait state after state for triggering the first test script so we can be sure that script execution on the GitHub side has been finished      10. Add a state at the end of the Step Function that should invoke the Lambda function that will trigger the Github Action workflow for running the second test case that will verify that the message has been processed successfully      11. Add env.json file with the GitHub Actons credentials and trigger IDs: {     "INSERT_ACTION_TRIGGER_ID": <GITHUB_WORKFLOW_ID_FOR_DYNAMODB_TEST_DATA_INSERTION>,     "VERIFY_ACTION_TRIGGER_ID":<GITHUB_WORKFLOW_ID_FOR_DYNAMODB_TEST_DATA_VERIFICATION>,     "GITHUB_OWNER": <YOUR_GITHUB_USERNAME>,     "GITHUB_REPO": <YOUR_GITHUB_REPO>,     "GITHUB_TOKEN": <YOUR_GITHUB_TOKEN> }      12. Specify the environment file inside serverless.yml file under the provider configuration The final version of the serverless.yml file for the Step Function definition should look like the following: service: sfn-blog frameworkVersion: "3" # Define  cloud provider settings and IAM roles needed for our SF to work provider:   name: aws   runtime: nodejs18.x   region: eu-central-1   environment: ${file(env.json)}   iamRoleStatements:     - Effect: Allow       Action: dynamodb:*       Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/messages     - Effect: Allow       Action: dynamodb:*       Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/US_users_events     - Effect: Allow       Action: dynamodb:*       Resource: arn:aws:dynamodb:eu-central-1:178190218027:table/EU_users_events     - Effect: Allow       Action: s3:*       Resource: arn:aws:s3:::sfn-blog plugins:   - serverless-step-functions # Define a path to the AWS Lambda Functions functions:   FetchFromDynamoDBState:     handler: handler.FetchFromDynamoDBState   ProcessUSUsersEvents:     handler: handler.ProcessUSUsersEvents   ProcessEUUsersEvents:     handler: handler.ProcessEUUsersEvents   InsertMessageDynamoDBTest:     handler: handler.InsertMessageDynamoDBTest   VerifyStepFunctionOutcomeTest:     handler: handler.VerifyStepFunctionOutcomeTest # Define Step Function and its state machine stepFunctions:   stateMachines:     proceedRewards:       name: proceedRewards       definition:         StartAt: InsertMessageDynamoDBTest         States:           InsertMessageDynamoDBTest:             Type: Task             Resource:               Fn::GetAtt: [InsertMessageDynamoDBTest, Arn]             ResultPath: "$.response"             Next: WaitState           WaitState:             Type: Wait             Seconds: 200             Next: FetchFromDynamoDBState           FetchFromDynamoDBState:             Type: Task             Resource:               Fn::GetAtt: [FetchFromDynamoDBState, Arn]             ResultPath: "$.items"             Next: ProcessDataState           ProcessDataState:             Type: Map             ItemsPath: "$.items.items"             ResultPath: "$.mappedData"             MaxConcurrency: 2             Iterator:               StartAt: ProceedChoiceState               States:                 ProceedChoiceState:                   Type: Choice                   Choices:                     - Variable: $.region.S                       StringEquals: "US"                       Next: ProcessUSUsersEventsState                     - Variable: $.region.S                       StringEquals: "EU"                       Next: ProcessEUUsersEventsState                   Default: DefaultState                 ProcessUSUsersEventsState:                   Type: Task                   Resource:                     Fn::GetAtt: [ProcessUSUsersEvents, Arn]                   End: true                 ProcessEUUsersEventsState:                   Type: Task                   Resource:                     Fn::GetAtt: [ProcessEUUsersEvents, Arn]                   End: true                 DefaultState:                   Type: Fail                   Cause: 'Invalid region value.'                   Error: 'InvalidRegionError'             Next: VerifyStepFunctionOutcomeTest           VerifyStepFunctionOutcomeTest:             Type: Task             Resource:               Fn::GetAtt: [VerifyStepFunctionOutcomeTest, Arn]             End: true The additional Lambdas for triggering GitHub workflows should look like this: const axios = require('axios'); module.exports.InsertMessageDynamoDBTest = async () => { await triggerWorkflow(process.env.INSERT_ACTION_TRIGGER_ID); } module.exports.VerifyStepFunctionOutcomeTest = async () => { await triggerWorkflow(process.env.VERIFY_ACTION_TRIGGER_ID); } const triggerWorkflow = async (workFlowId) => { const owner = process.env.GITHUB_OWNER; const repo = process.env.GITHUB_REPO; const token = process.env.GITHUB_TOKEN; let response; try {         response = await axios.post( `https://api.github.com/repos/${owner}/${repo}/actions/workflows/${workFlowId}/dispatches`,             {                 ref: 'main'             },             {                 headers: { 'Content-Type': 'application/json', 'Authorization': `Bearer ${token}`                 }             }         ); console.log('Job triggered successfully:', response.data);     } catch (error) { console.error('Failed to trigger job:', error.response.data);     } return response; } The final version of the Step Function should look like the following: This way, we will ensure that our Step Function continuously processes the messages successfully. Note: The testing states should be present only in the testing environments. Alternatives We could implement test automation for this kind of system by using other approaches. For example, we could create scheduled triggers for GitHub Actions Workflows. The first one for data insertion to the DynamoDB could be scheduled at the beginning of the day, and the second one for the verification of processed data could be scheduled at a time when we expect that at least one Step Function Workflow execution has been finished. The advantage of this approach is that we could avoid using the first and the last lambda functions for triggering the GitHub Actions Workflows that we specified in our Test Automation part. The advantage of the first approach that we explained is that if someone makes some changes on the Step Function, it will be verified on the next execution. We could also implement both approaches with Jenkins and any other CI/CD tool. The main challenges The main challenges for the implementation of test automation on systems based on Step Functions (systems that run periodically or per some event) are unique, and here are some of the main ones: Time-dependent testing: Systems triggered periodically often involve time-based operations. We may need to control the passage of time during tests to validate time-dependent behaviors. Data Security: We must ensure that test data are handled securely and that the test environments are adequately isolated, especially if we are working with sensitive data. End-to-End Testing: We need to ensure that all components work together, and it often requires two or more separate scripts that will orchestrate together by exchanging the data as artifacts for testing just one system behavior.  Continuous Integration/Continuous Deployment (CI/CD): We must ensure that tests run automatically with each deployment. Handling Large Data Sets: Systems with Step Functions may process large data sets. Testing with large data can take time and resources, so it needs a particular approach. Environment Consistency: To catch environment-specific issues, test environments should closely resemble production environments, including Cloud configurations.  Resource Setup and Teardown: Setting up and tearing down AWS resources, such as databases, queues, or S3 buckets, for each test can be time-consuming and costly. Data Generation: Creating test data to simulate real-world scenarios and edge cases can be challenging. We must ensure that the data is consistent with the state transitions in our Step Function workflows. Connection with External Services: Establishing connections to the different services like databases and etc., can sometimes be very challenging based on many different situations, including security configurations, etc. Conclusion We have gone through the overall concept of development and application of the Step Function and the integration of automation testing within its workflow. Through all this, we were introduced to additional tools from the domain of cloud computing, CI/CD, and test automation. You can find the full project on this GitHub repository. Sources: https://docs.aws.amazon.com/step-functions/latest/dg/concepts-states.html If you found this useful, check out other Atlantbh blogs!

A data processing-based solution refers to a system or approach that involves data collection, manipulation, analysis, and utilization to solve a particular problem or address a specific need. Data processing is crucial in various fields and industries, including business, healthcare, finance, science, and technology. It enables companies to turn raw data into valuable information for business. These kinds of solutions require a particular approach from the quality assurance standpoint. This article will explain QA professionals’ challenges with data processing-based solutions and introduce strategies to conquer them. Data Processing Stages From the QA standpoint, we must ensure that every data processing stage meets quality standards. We often hear the term ETL process (extract, transform, and load)  and its initial part of the data processing-based solutions. ETL uses a set of business rules to clean and organize raw data and prepare it for the  next steps in data processing according to the What is ETL? - Extract Transform Load Explained - AWS. It is possible to have many steps that are part of the data processing workflow, but these are the crucial ones that are always present in data processing according to the 5 Best Data Processing Software: Complete Guide: Data Collection: This starting point in data processing includes collecting raw data from correct sources, such as message brokers, file storage, etc. This step refers to the ‘extract’ part of the ETL. Data Preparation: This step represents filtering invalid, unneeded, or inaccurate data and converting data to the  format needed for further processing, and it refers to the ‘transform’ part of the ETL. Data Input: In this step, prepared data is provided to the processing step, and it refers to the ‘load’ part of the ETL, in which the data is moved from staging(initial) to the target area. Processing: The data undergoes different transformations to get the desired output, including calculations and various data processing methods (machine learning or other algorithms). Data output/interpretation: In this step, final processing starts. Data teams display it on some UI in easy-to-read formats for users, such as graphs, widgets, dashboards, tables, video, audio, etc. Sometimes, engineers develop applications (web, mobile, desktop, etc.) that present data; sometimes, they generate reports based on the data. There are several main data types used to present data in a usable way: - Text: telling story for data   - Chart: showing trends such as growth or decline   - Table: presenting statistical data   - Image: images are also widely used for data presentations Data storage: Data teams store data and metadata for many reasons, such as quick access when needed, further processing, keeping backup data, etc. We can save it in databases, data warehouses, file storage, etc. In most cases this whole process is automated by using various tools, and it can be based on many processing models, and these are the most common ones according to the Difference between Batch Processing and Real Time Processing System - GeeksforGeeks: Batch Processing: Processes large volumes of data in batches by schedule. Real-Time Processing: Processes data as it arrives, in real-time or near-real-time. Main QA Challenges We have several challenges from a Quality Assurance standpoint when working with data processing-based solutions. Here are the main challenges: Ensuring data quality and integrity: QA teams ensure the accuracy and integrity of data throughout the data pipelines. They verify that data is not lost, duplicated, or corrupted during the ETL process (including data loads, transformations, etc., through the data pipelines). Data validation and verification: QA teams need to create data verification and validation test cases (manual or automated) that verify that data meets quality standards and that data transformations comply with business needs. Sometimes, these transformations can be complex and include calculations and aggregations.Example: Let us say that we have a system that consumes data from some message broker like Kafka or RabbitMQ, then saves the message to the initial data area (S3 bucket, SQL database, etc.), and then goes to the target area. After that, we do some data transformations, including mathematical operations, to get the final data outcome which is saved in a report file as an Excel sheet. For this kind of solution, we can have the following test case steps: -  Provide data to the message broker. - Connect to the initial data area and verify that the data sent to the message broker is stored and valid. - Connect to the target data area and verify that data has been transformed according to the accepted criteria. - Verify that the report file is created in the needed place. - Verify that the content of the file meets accepted criteria. Handling large volumes of data: This kind of solution, in most cases, includes big data sets. From a QA standpoint, verifying that the system will stay stable under a large volume of data without bottlenecks or data loss is essential. A good solution for this verification is to develop load tests (using frameworks such as JMeter, k6, etc.). It can also be tricky while developing automated test cases because large data can take time and resources, so it needs a particular approach. QA teams can not perform all test case scenarios in lower environments: Often, mocked data on the lower environments is not enough to test some features thoroughly, so the QA teams are completing testing upper environments with the production data. One solution that can help in this case is to have one test environment that consumes the same data as the production environment. Paying attention and developing practices to keep mocked data distinct from production data is important. Insufficient data to test the feature: Often, a lot of data is needed in the system to test some features thoroughly, and we can not predict how the system will behave when more data comes in. One of the options to overcome this challenge is to generate mocked data temporarily or to postpone the deployment to the production environment until the data is accurate. Data privacy and security: Attention to data privacy regulations(GDPR, HIPAA) is essential. QA teams must store the data securely, focus on access control, and adequately isolate the test environments. Documentation and knowledge transfer: QA teams should collaborate with Product Owners on creating and maintaining up-to-date documentation. For solution quality maintenance, it is necessary to ensure knowledge transfer and documentation of test cases. Cross-functional Collaboration: Effective collaboration between QA teams, product owners, software engineers, data engineers, data scientists, and other professionals included in development is crucial. Through well-established communication, it is ensured that everyone has a shared understanding of quality requirements. Sometimes, more than one team works on data processing-based solutions, which requires well-established contact through cross-team channels and documentation. For example, one team works on the data collection and preparation, and the other team works on the other data processing stages. Working with various data processing tools: QA professionals need to work with multiple data processing tools. The most popular ones are AWS (with its data processing resources such as Redshift, Glue, Lambda, etc.), Google Cloud Platform (with its data processing resources such as BigQuery, Cloud Dataprep, etc.), Azure Cloud (with its data processing resources such as SQL Warehouse, Data Factory, Azure Functions, etc.), and Snowflake, among others. They need to be capable of following trends in data processing by continuously improving their skills and keeping up to date with new tools and practices. Time-dependent testing: This requires a particular approach for the manual verification and test automation.If the system is based on a batch processing model, it triggers periodically, so we must adapt our test cases. For example, the processing stage runs every hour, so after providing new data, we have to wait for the processing schedule to verify that the data is successfully processed.On the other hand, if the system is based on a real-time processing model we may encounter other difficulties related to the continuous data streams, especially during the test automation. For example, if our automated test case expects some output based on calculations, and the new data stream enters the system it could cause a false test failure. Conclusion In the continued development of data processing-based solutions, we must recognize the role of Quality Assurance. The challenges of QA in data processing-based solutions may be complex, but they are solvable with the right strategies and commitment to excellence. All in all, QA is the unsung hero in ensuring the integrity and reliability of data. The data-driven future holds limitless opportunities, and with a strong QA, we can be sure that we are approaching with precision and confidence. "QA Challenges on Data Processing-based Solutions" Tech Bite was brought to you by Haris Habul, Senior Quality Assurance Engineer at Atlantbh. (more…)