From 00b4954b704c7df2982ed81d67b78ac76025793c Mon Sep 17 00:00:00 2001
From: Guflly <145608489+Guflly@users.noreply.github.com>
Date: Fri, 31 Jul 2026 05:52:11 -0700
Subject: [PATCH] Fix dataframe CSV tests on Windows

---
 sdks/python/apache_beam/dataframe/io.py      |  2 +-
 sdks/python/apache_beam/dataframe/io_test.py | 15 ++++++++-------
 2 files changed, 9 insertions(+), 8 deletions(-)

diff --git a/sdks/python/apache_beam/dataframe/io.py b/sdks/python/apache_beam/dataframe/io.py
index 21eab0b82faf..bc39a40403fb 100644
--- a/sdks/python/apache_beam/dataframe/io.py
+++ b/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):
diff --git a/sdks/python/apache_beam/dataframe/io_test.py b/sdks/python/apache_beam/dataframe/io_test.py
index 4cd502d1b8d7..dd7b8db497ce 100644
--- a/sdks/python/apache_beam/dataframe/io_test.py
+++ b/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)))