-
Notifications
You must be signed in to change notification settings - Fork 56
Test VDK Data Jobs with pytest
This tutorial guides you through the process of testing Versatile Data Kit (VDK) data jobs to ensure they run correctly and efficiently. Testing is crucial for maintaining data integrity, ensuring data processing logic is correct, and catching errors early in the development cycle. Upon completion, you'll understand how to implement tests for VDK data jobs, improving your data pipelines' reliability and robustness.
Audience: This tutorial is intended for data engineers and developers working with VDK to build and manage data processing jobs.
Estimated Time: Completing this tutorial should take less than 30 minutes.
- Familiarity with Python and pytest framework.
- Basic understanding of VDK and its components.
- VDK and vdk-test-utils installed in your development environment.
Before writing tests, ensure you have vdk-test-utils
and pytest
installed. These tools provide functionalities to write and run tests for VDK data jobs.
pip install vdk-test-utils pytest
VDK provides CliEntryBasedTestRunner, a tool for creating unit or functional tests for your plugins or data jobs in an isolated environment. It's useful for testing data jobs without affecting your production environment by stubbing external dependencies
Let's say we want to test a simple data transformation job. First, initialize the test runner with any plugins you're using
from vdk.internal.test_utils import CliEntryBasedTestRunner
list_of_plugins_i_am_using = []
runner = CliEntryBasedTestRunner(list_of_plugins_i_am_using)
Then, invoke the data job you wish to test:
result = runner.invoke(["run", "path/to/your-data-job"])
cli_assert_equal(0, result)
assert 'expected_output' in result.output
Tip: Use
cli_assert_equal(0, result)
to ensure the job ran successfully without errors and on error to easier to debug logs
To test a job that ingests data over HTTP:
- Mock an HTTP server response.
- Use CliEntryBasedTestRunner with the HTTP ingestion plugin.
- Run your data job and assert the expected outcome
from vdk.internal.test_utils import mock, CliEntryBasedTestRunner
def test_http_ingestion(httpserver):
httpserver.expect_request("/ingest").respond_with_response(Response(status=200))
# Set up environment variables for the test
with mock.patch.dict(os.environ, {"VDK_INGEST_METHOD_DEFAULT": "http", ...}):
runner = CliEntryBasedTestRunner(ingest_http_plugin)
result = runner.invoke(["run", "path/to/ingest-job"])
cli_assert_equal(0, result)
assert len(httpserver.log) == expected_batches
For SQL transformation jobs:
- Run the job using CliEntryBasedTestRunner.
- Execute a query against the transformed data.
- Assert the expected data is present.
def test_sql_transformation():
runner = CliEntryBasedTestRunner([])
result = runner.invoke(["run", "path/to/sql-job"])
cli_assert_equal(0, result)
query_result = runner.invoke(["sql-query", "--query", "SELECT * FROM table"])
assert "expected_value" in query_result.output
Congratulations on completing this tutorial! You've learned how to set up a testing environment for VDK data jobs, write basic and advanced tests, and ensure your data processing logic works as expected.
Testing is a vital part of the data job development process, helping to catch errors early and ensure data integrity. By following this tutorial, you've equipped yourself with the knowledge to write effective tests for your VDK data jobs.
Explore further by testing different types of data jobs, incorporating continuous integration processes for automated testing, and reviewing the VDK documentation for more advanced features and best practices .
SDK - Develop Data Jobs
SDK Key Concepts
Control Service - Deploy Data Jobs
Control Service Key Concepts
- Scheduling a Data Job for automatic execution
- Deployment
- Execution
- Production
- Properties and Secrets
Operations UI
Community
Contacts