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(