From 76eed0d4e9c6a14563867f5190fb55cac6c6a2ab Mon Sep 17 00:00:00 2001 From: Laurent Savaete Date: Thu, 1 Sep 2022 04:48:36 +0300 Subject: [PATCH] feat(taps): Add checks for primary keys, replication keys and state partitioning keys to standard tap tests (#829) * Add code validation tests to standard tests * Add extra test for replication keys * Apply suggestions from code review Co-authored-by: Edgar R. M. * Fix formatting for ci check Co-authored-by: Edgar R. M. --- singer_sdk/testing.py | 47 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/singer_sdk/testing.py b/singer_sdk/testing.py index d2fa94d91..b0c8a926d 100644 --- a/singer_sdk/testing.py +++ b/singer_sdk/testing.py @@ -40,7 +40,52 @@ def _test_stream_connections() -> None: tap1: Tap = tap_class(config=config, parse_env_config=True) tap1.run_connection_test() - return [_test_cli_prints, _test_discovery, _test_stream_connections] + def _test_pkeys_in_schema() -> None: + """Verify that primary keys are actually in the stream's schema.""" + tap = tap_class(config=config, parse_env_config=True) + for name, stream in tap.streams.items(): + pkeys = stream.primary_keys or [] + schema_props = set(stream.schema["properties"].keys()) + for pkey in pkeys: + error_message = ( + f"Coding error in stream '{name}': " + f"primary_key '{pkey}' is missing in schema" + ) + assert pkey in schema_props, error_message + + def _test_state_partitioning_keys_in_schema() -> None: + """Verify that state partitioning keys are actually in the stream's schema.""" + tap = tap_class(config=config, parse_env_config=True) + for name, stream in tap.streams.items(): + sp_keys = stream.state_partitioning_keys or [] + schema_props = set(stream.schema["properties"].keys()) + for sp_key in sp_keys: + assert sp_key in schema_props, ( + f"Coding error in stream '{name}': state_partitioning_key " + f"'{sp_key}' is missing in schema" + ) + + def _test_replication_keys_in_schema() -> None: + """Verify that the replication key is actually in the stream's schema.""" + tap = tap_class(config=config, parse_env_config=True) + for name, stream in tap.streams.items(): + rep_key = stream.replication_key + if rep_key is None: + continue + schema_props = set(stream.schema["properties"].keys()) + assert rep_key in schema_props, ( + f"Coding error in stream '{name}': replication_key " + f"'{rep_key}' is missing in schema" + ) + + return [ + _test_cli_prints, + _test_discovery, + _test_stream_connections, + _test_pkeys_in_schema, + _test_state_partitioning_keys_in_schema, + _test_replication_keys_in_schema, + ] def get_standard_target_tests(