Test about info

singer_sdk/target_base.py

@@ -571,6 +571,38 @@ def get_singer_command(cls: type[Target]) -> click.Command:
 
         return command
 
+    @classmethod
+    def append_builtin_config(cls: type[SQLTarget], config_jsonschema: dict) -> None:
+        """Appends built-in config to `config_jsonschema` if not already set.
+
+        To customize or disable this behavior, developers may either override this class
+        method or override the `capabilities` property to disabled any unwanted
+        built-in capabilities.
+
+        For all except very advanced use cases, we recommend leaving these
+        implementations "as-is", since this provides the most choice to users and is
+        the most "future proof" in terms of taking advantage of built-in capabilities
+        which may be added in the future.
+
+        Args:
+            config_jsonschema: [description]
+        """
+
+        def _merge_missing(source_jsonschema: dict, target_jsonschema: dict) -> None:
+            # Append any missing properties in the target with those from source.
+            for k, v in source_jsonschema["properties"].items():
+                if k not in target_jsonschema["properties"]:
+                    target_jsonschema["properties"][k] = v
+
+        capabilities = cls.capabilities
+
+        if PluginCapabilities.BATCH in capabilities:
+            _merge_missing(BATCH_CONFIG, config_jsonschema)
+
+        super().append_builtin_config(config_jsonschema)
+
+    pass
+
 
 class SQLTarget(Target):
     """Target implementation for SQL destinations."""
@@ -615,9 +647,6 @@ def _merge_missing(source_jsonschema: dict, target_jsonschema: dict) -> None:
         if TargetCapabilities.TARGET_SCHEMA in capabilities:
             _merge_missing(TARGET_SCHEMA_CONFIG, config_jsonschema)
 
-        if PluginCapabilities.BATCH in capabilities:
-            _merge_missing(BATCH_CONFIG, config_jsonschema)
-
         super().append_builtin_config(config_jsonschema)
 
     pass

tests/conftest.py

@@ -11,12 +11,16 @@
 from sqlalchemy import __version__ as sqlalchemy_version
 
 from singer_sdk import typing as th
+from singer_sdk.helpers.capabilities import PluginCapabilities
 from singer_sdk.sinks import BatchSink
 from singer_sdk.target_base import Target
 
 if t.TYPE_CHECKING:
     from _pytest.config import Config
 
+    from singer_sdk.helpers.capabilities import CapabilitiesEnum
+
+
 SYSTEMS = {"linux", "darwin", "windows"}
 
 pytest_plugins = ("singer_sdk.testing.pytest_plugin",)
@@ -103,6 +107,10 @@ class TargetMock(Target):
     name = "target-mock"
     config_jsonschema = th.PropertiesList().to_dict()
     default_sink_class = BatchSinkMock
+    capabilities: t.ClassVar[CapabilitiesEnum] = [
+        *Target.capabilities,
+        PluginCapabilities.BATCH,
+    ]
 
     def __init__(self, *args, **kwargs):
         """Create the Mock target sync."""

tests/core/test_target_base.py

@@ -5,6 +5,7 @@
 import pytest
 
 from singer_sdk.exceptions import MissingKeyPropertiesError
+from singer_sdk.helpers.capabilities import PluginCapabilities
 from tests.conftest import BatchSinkMock, TargetMock
 
 
@@ -53,3 +54,21 @@ def test_validate_record():
     # Test invalid record
     with pytest.raises(MissingKeyPropertiesError):
         sink._singer_validate_message({"name": "test"})
+
+
+def test_target_about_info():
+    target = TargetMock()
+    about = target._get_about_info()
+
+    assert about.capabilities == [
+        PluginCapabilities.ABOUT,
+        PluginCapabilities.STREAM_MAPS,
+        PluginCapabilities.FLATTENING,
+        PluginCapabilities.BATCH,
+    ]
+
+    assert "stream_maps" in about.settings["properties"]
+    assert "stream_map_config" in about.settings["properties"]
+    assert "flattening_enabled" in about.settings["properties"]
+    assert "flattening_max_depth" in about.settings["properties"]
+    assert "batch_config" in about.settings["properties"]