[Python] Add ValidatesRunner coverage for UnboundedSource SDF wrapper

Mark EndToEndTest with it_validatesrunner and no_sickbay_batch so the Dataflow
ValidatesRunner streaming suite exercises the wrapper end to end; the unbounded
source requires --streaming, so it is excluded from the batch suite.

Add test_unbounded_source_read to PortableRunnerTest so portable ValidatesRunner
covers the wrapper. Prism and the DirectRunner execute it; Flink and Spark skip
with documented runner limitations: SDF watermark tracking (BEAM-2939) and bundle
finalization (#19526) on Flink, and SDF execution on portable Spark (#19468).

Author: Eliaaazzz

sdks/python/apache_beam/io/unbounded_source_test.py

@@ -24,6 +24,7 @@
 import logging
 import unittest
 
+import pytest
 from typing_extensions import override
 
 import apache_beam as beam
@@ -732,6 +733,8 @@ def test_finalize_checkpoint_invoked(self):
     self.assertEqual(finalize_log, [1])
 
 
+@pytest.mark.it_validatesrunner
+@pytest.mark.no_sickbay_batch
 class EndToEndTest(unittest.TestCase):
   def test_direct_runner_emits_all_in_order(self):
     with TestPipeline() as p:

sdks/python/apache_beam/runners/portability/portable_runner_test.py

@@ -30,6 +30,7 @@
 import grpc
 
 import apache_beam as beam
+from apache_beam.io.unbounded_source_test import UnboundedCountingSource
 from apache_beam.options.pipeline_options import DebugOptions
 from apache_beam.options.pipeline_options import DirectOptions
 from apache_beam.options.pipeline_options import PipelineOptions
@@ -47,6 +48,7 @@
 from apache_beam.testing.util import equal_to
 from apache_beam.transforms import environments
 from apache_beam.transforms import userstate
+from apache_beam.transforms import window
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -209,6 +211,40 @@ def process(self, kv, index=beam.DoFn.StateParam(index_state_spec)):
           | beam.ParDo(AddIndex()),
           equal_to(expected))
 
+  def test_unbounded_source_read(self):
+    """Reads a self-terminating UnboundedSource through the job service.
+
+    Complements the DirectRunner tests in apache_beam.io.unbounded_source_test
+    by exercising the runner-API round trip and the EOF ``MAX_TIMESTAMP``
+    watermark propagation that lets the downstream window fire.
+    """
+    if type(self).__name__ in {
+        'FlinkRunnerTest',
+        'FlinkRunnerTestOptimized',
+        'FlinkRunnerTestStreaming',
+        'SparkRunnerTest',
+    }:
+      # Flink lacks the SDF watermark tracking this source relies on
+      # (BEAM-2939; all Flink variants, including Streaming, skip
+      # test_sdf_with_watermark_tracking) and bundle finalization
+      # (https://github.com/apache/beam/issues/19526); portable Spark does not
+      # execute SDFs (https://github.com/apache/beam/issues/19468).
+      raise unittest.SkipTest(
+          'Requires SDF watermark tracking and bundle finalization (Flink) '
+          'and SDF execution (Spark).')
+    with self.create_pipeline() as p:
+      out = p | beam.io.Read(UnboundedCountingSource(5))
+      self.assertFalse(out.is_bounded)
+      assert_that(out, equal_to([0, 1, 2, 3, 4]), label='assert_elements')
+      windowed = (
+          out
+          | beam.WindowInto(window.FixedWindows(100))
+          | beam.Map(lambda v: ('all', v))
+          | beam.GroupByKey()
+          | beam.MapTuple(lambda _key, values: sorted(values)))
+      assert_that(
+          windowed, equal_to([[0, 1, 2, 3, 4]]), label='assert_windowed')
+
   # Inherits all other tests from fn_api_runner_test.FnApiRunnerTest
 
   def test_sdf_default_truncate_when_bounded(self):