streaming source fix

Author: jcatt-sf

src/datacustomcode/client.py

@@ -55,17 +55,17 @@ def _streaming_source_name() -> str:
     """Return the streaming transform's read-source name.
 
     Resolved from ``config.streaming_source``, which ``run_entrypoint``
-    populates from config.json's ``permissions.read`` entry.
+    populates from config.json's ``streamingSource`` field.
 
     Raises:
         RuntimeError: If no ``streaming_source`` has been configured (e.g. the
-            transform's config.json has no ``permissions.read`` entry).
+            transform's config.json has no ``streamingSource`` field).
     """
     source = config.streaming_source
     if not source:
         raise RuntimeError(
             "No streaming source configured. A streaming transform must declare "
-            "its read source in config.json under 'permissions.read'."
+            "its read source in config.json under 'streamingSource'."
         )
     return source
 

src/datacustomcode/config.py

@@ -90,7 +90,7 @@ class ClientConfig(BaseConfig):
         SparkProviderConfig[BaseSparkSessionProvider], None
     ] = None
     # Source object name for a streaming (DELTA_SYNC) transform, populated by
-    # ``run_entrypoint`` from config.json's ``permissions.read``
+    # ``run_entrypoint`` from config.json's ``streamingSource`` field
     streaming_source: Union[str, None] = None
 
     def update(self, other: ClientConfig) -> ClientConfig:

src/datacustomcode/run.py

@@ -42,20 +42,14 @@ def _set_config_option(config_obj, key: str, value: Optional[str]) -> None:
         config_obj.options[key] = value
 
 
-def _read_source_from_permissions(config_json: dict) -> Optional[str]:
-    """Return the read-source name from config.json ``permissions.read``.
+def _read_streaming_source(config_json: dict) -> Optional[str]:
+    """Return the streaming source name from config.json's ``streamingSource``.
     """
-    permissions = config_json.get("permissions")
-    if not isinstance(permissions, dict):
+    source = config_json.get("streamingSource")
+    if not isinstance(source, dict):
         return None
-    read = permissions.get("read")
-    if not isinstance(read, dict):
-        return None
-    for layer in ("dlo", "dmo"):
-        names = read.get(layer)
-        if names:
-            return names[0]
-    return None
+    name = source.get("name")
+    return str(name) if name else None
 
 
 def _update_config_options(profile: Optional[str], sf_cli_org: Optional[str]):
@@ -141,7 +135,7 @@ def run_entrypoint(
         _set_config_option(config.reader_config, "dataspace", dataspace)
         _set_config_option(config.writer_config, "dataspace", dataspace)
 
-        config.streaming_source = _read_source_from_permissions(config_json)
+        config.streaming_source = _read_streaming_source(config_json)
 
     _update_config_options(profile, sf_cli_org)
 

tests/test_client.py

@@ -301,7 +301,7 @@ def test_read_dlo_deltas(self, reset_client, mock_spark):
         reader.read_dlo_deltas.return_value = mock_df
 
         client = StreamingClient(reader=reader, writer=writer)
-        
+
         with patch("datacustomcode.client.config") as mock_config:
             mock_config.streaming_source = "Account_std__dll"
             result = client.read_dlo_deltas()
@@ -323,7 +323,7 @@ def test_read_dlo_deltas_without_configured_source_raises(
             with pytest.raises(RuntimeError) as exc_info:
                 client.read_dlo_deltas()
 
-        assert "permissions.read" in str(exc_info.value)
+        assert "streamingSource" in str(exc_info.value)
         reader.read_dlo_deltas.assert_not_called()
 
     def test_read_dmo_deltas(self, reset_client, mock_spark):

tests/test_run.py

@@ -490,51 +490,39 @@ def test_run_entrypoint_empty_dataspace_value(self):
                 os.unlink(config_json_path)
 
 
-class TestReadSourceFromPermissions:
-    """`_read_source_from_permissions` extracts the streaming read source from
-    config.json's `permissions.read`."""
+class TestReadStreamingSource:
+    """`_read_streaming_source` extracts the source name from config.json's
+    `streamingSource` object."""
 
-    def test_returns_single_dlo(self):
-        from datacustomcode.run import _read_source_from_permissions
+    def test_returns_dlo_name(self):
+        from datacustomcode.run import _read_streaming_source
 
-        config_json = {"permissions": {"read": {"dlo": ["Account_std__dll"]}}}
-        assert _read_source_from_permissions(config_json) == "Account_std__dll"
+        config_json = {"streamingSource": {"type": "dlo", "name": "Account_Home__dll"}}
+        assert _read_streaming_source(config_json) == "Account_Home__dll"
 
-    def test_returns_single_dmo(self):
-        from datacustomcode.run import _read_source_from_permissions
-
-        config_json = {"permissions": {"read": {"dmo": ["Account_model__dlm"]}}}
-        assert _read_source_from_permissions(config_json) == "Account_model__dlm"
-
-    def test_dlo_preferred_when_both_present(self):
-        from datacustomcode.run import _read_source_from_permissions
+    def test_returns_dmo_name(self):
+        from datacustomcode.run import _read_streaming_source
 
         config_json = {
-            "permissions": {"read": {"dlo": ["the_dll"], "dmo": ["the_dlm"]}}
+            "streamingSource": {"type": "dmo", "name": "AccountTransformed__dlm"}
         }
-        assert _read_source_from_permissions(config_json) == "the_dll"
-
-    def test_returns_first_of_multiple(self):
-        from datacustomcode.run import _read_source_from_permissions
-
-        config_json = {"permissions": {"read": {"dlo": ["first__dll", "second__dll"]}}}
-        assert _read_source_from_permissions(config_json) == "first__dll"
+        assert _read_streaming_source(config_json) == "AccountTransformed__dlm"
 
     @pytest.mark.parametrize(
         "config_json",
         [
             {},
-            {"permissions": None},
-            {"permissions": {}},
-            {"permissions": {"read": None}},
-            {"permissions": {"read": {}}},
-            {"permissions": {"read": {"dlo": []}}},
+            {"streamingSource": None},
+            {"streamingSource": {}},
+            {"streamingSource": {"type": "dlo"}},
+            {"streamingSource": {"type": "dlo", "name": ""}},
+            {"streamingSource": {"type": "dlo", "name": None}},
         ],
     )
     def test_returns_none_when_absent_or_empty(self, config_json):
-        from datacustomcode.run import _read_source_from_permissions
+        from datacustomcode.run import _read_streaming_source
 
-        assert _read_source_from_permissions(config_json) is None
+        assert _read_streaming_source(config_json) is None
 
 
 class TestStreamingSourceScenarios:
@@ -575,15 +563,15 @@ def _run_capturing_streaming_source(self, config_json_body):
             if os.path.exists("streaming_source_output.txt"):
                 os.unlink("streaming_source_output.txt")
 
-    def test_streaming_source_set_from_permissions_read(self):
+    def test_streaming_source_set_from_streaming_source_field(self):
         content = self._run_capturing_streaming_source(
             {
                 "dataspace": "default",
-                "permissions": {"read": {"dlo": ["Account_std__dll"]}},
+                "streamingSource": {"type": "dlo", "name": "Account_Home__dll"},
             }
         )
-        assert "streaming_source: Account_std__dll" in content
+        assert "streaming_source: Account_Home__dll" in content
 
-    def test_streaming_source_none_for_batch_without_read(self):
+    def test_streaming_source_none_for_batch_without_field(self):
         content = self._run_capturing_streaming_source({"dataspace": "default"})
         assert "streaming_source: None" in content