[Python] Add UnboundedSource ValidatesRunner test on portable runners

Add test_unbounded_source_read to PortableRunnerTest so the portable
ValidatesRunner suites exercise the UnboundedSource SDF wrapper end to
end: read a self-terminating source through the job service, assert the
elements and that the EOF MAX_TIMESTAMP watermark lets a downstream
FixedWindows + GroupByKey fire.

The embedded portable runner variants, the Flink suites, and
PrismRunnerTest inherit the test. SparkRunnerTest skips it because
portable Spark does not execute SDFs (#19468), matching its other SDF
test skips.

Author: Eliaaazzz

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,30 @@ 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__ == 'SparkRunnerTest':
+      # Portable Spark does not execute SDFs, so SparkRunnerTest skips every
+      # SDF test (see its test_sdf* overrides).
+      raise unittest.SkipTest('https://github.com/apache/beam/issues/19468')
+    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):