Fix dataframe CSV tests on Windows (#39563)

Author: Guflly

sdks/python/apache_beam/dataframe/io.py

@@ -736,7 +736,7 @@ def open(self, file_handle):
     self.empty = self.header = self.footer = None
     if not self.binary:
       file_handle = TextIOWrapper(
-          file_handle, encoding=self.kwargs.get("encoding", None))
+          file_handle, encoding=self.kwargs.get("encoding", None), newline='')
     self.file_handle = file_handle
 
   def write_to(self, df, file_handle=None):

sdks/python/apache_beam/dataframe/io_test.py

@@ -18,7 +18,6 @@
 import importlib
 import math
 import os
-import platform
 import shutil
 import tempfile
 import typing
@@ -65,9 +64,6 @@ class MyRow(typing.NamedTuple):
   value: int
 
 
-@unittest.skipIf(
-    platform.system() == 'Windows',
-    'https://github.com/apache/beam/issues/20642')
 class IOTest(unittest.TestCase):
   def setUp(self):
     self._temp_roots = []
@@ -431,6 +427,11 @@ def test_file_not_found(self):
 
   def test_windowed_write(self):
     output = self.temp_dir()
+
+    def no_colon_file_naming(*args):
+      file_name = fileio.default_file_naming('out.csv')(*args)
+      return file_name.replace(':', '_')
+
     with beam.Pipeline() as p:
       pc = (
           p | beam.Create([MyRow(timestamp=i, value=i % 3) for i in range(20)])
@@ -440,18 +441,18 @@ def test_windowed_write(self):
               beam.window.FixedWindows(10)).with_output_types(MyRow))
 
       deferred_df = convert.to_dataframe(pc)
-      deferred_df.to_csv(output + 'out.csv', index=False)
+      deferred_df.to_csv(output, file_naming=no_colon_file_naming, index=False)
 
     first_window_files = (
         f'{output}out.csv-'
-        f'{datetime.utcfromtimestamp(0).isoformat()}*')
+        f'{datetime.utcfromtimestamp(0).isoformat().replace(":", "_")}*')
     self.assertCountEqual(
         ['timestamp,value'] + [f'{i},{i % 3}' for i in range(10)],
         set(self.read_all_lines(first_window_files, delete=True)))
 
     second_window_files = (
         f'{output}out.csv-'
-        f'{datetime.utcfromtimestamp(10).isoformat()}*')
+        f'{datetime.utcfromtimestamp(10).isoformat().replace(":", "_")}*')
     self.assertCountEqual(
         ['timestamp,value'] + [f'{i},{i%3}' for i in range(10, 20)],
         set(self.read_all_lines(second_window_files, delete=True)))