refactor(pipelines/cdp): modularize test suite to mirror pipeline structure - #275
Merged
Merged
Conversation
…ucture - Decompose monolithic test_customer_data_platform.py into dedicated test modules: - test_models.py: tests for NamedTuple data models, payload parsing, and Beam schema compatibility - test_options.py: tests for pipeline and GCP options parsing and defaults - test_schemas.py: tests for packaged and custom BigQuery table schema loading - test_parsing.py: tests for ParseRecordDoFn and AssignEventTimestampDoFn - test_sessionization.py: tests for ProcessCustomerSessionDoFn aggregation and metrics - test_sinks.py: tests for Storage Write API timestamp formatting and BigQuery sink attachment - test_pipeline.py: tests for build_pipeline DAG assembly and in-memory execution - Expand unit test coverage for sinks and timestamp assignment (41 tests passing, up from 21) - Verify full compliance with Google Python style (yapf) and PyLint (10.00/10) TAG=agy CONV=59306c7c-80e2-4454-9ec5-e496e3b85156
iht
commented
Sep 9, 2026
…uites - Refactor test_sessionization.py to run ProcessCustomerSessionDoFn inside TestPipeline with Sessions windowing and GroupByKey - Refactor test_parsing.py to execute ParseRecordDoFn and AssignEventTimestampDoFn inside TestPipeline with assert_that - Update test_sinks.py to use TestPipeline for DAG construction tests - Set TestPipeline.__test__ = False to ensure seamless pytest discovery TAG=agy CONV=59306c7c-80e2-4454-9ec5-e496e3b85156
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
The test suite for the Customer Data Platform (CDP) pipeline currently consolidated all tests into a single monolithic 443-line file (
pipelines/cdp/tests/test_customer_data_platform.py). While the main pipeline was previously refactored into cohesive submodules inpipelines/cdp/cdp_pipeline/, the tests did not mirror this structure.This PR modularizes
pipelines/cdp/tests/to establish exact 1:1 parity withpipelines/cdp/cdp_pipeline/:test_models.py: unit tests forTransactionItem,CouponRedemption,DeadLetterRecord,CustomerInteractionEvent,UnifiedTransactionRecord,CustomerSessionProfile, and Beam schema compatibilitytest_options.py: tests forMyPipelineOptionsdefaults, CLI flag parsing, and GCP optionstest_schemas.py: tests for default and custom BigQuery schema loading and missing file handlingtest_parsing.py: tests forParseRecordDoFn(transactions, coupons, malformed payloads, missing keys) andAssignEventTimestampDoFn(ISO strings, micros fallback, UTC now fallback)test_sessionization.py: tests forProcessCustomerSessionDoFnwindow aggregation, session profiling, and multi-coupon joinstest_sinks.py: tests for Storage Write API timestamp conversions (_to_beam_timestamp), row dict formatters, and BigQuery sink graph attachmenttest_pipeline.py: tests forbuild_pipelineDAG assembly and in-memory execution usingTestPipelineVerification
pytest tests/ -v(up from 21 tests previously)yapfpylint --rcfile ../pylintrc cdp_pipeline testspython setup.py sdistTAG=agy
CONV=59306c7c-80e2-4454-9ec5-e496e3b85156