small filesystem fixes (#34956)

* Fix typehint for get_filesystem

* Fix bug with creating files in cwd with LocalFileSystem

* lint

* revert precommit config change

* isort

* add file to start PR builder

* Revert "add file to start PR builder"

This reverts commit f80b361.

* fix isort again

Author: hjtran

sdks/python/apache_beam/io/filesystems.py

@@ -125,7 +125,7 @@ def get_scheme(path):
 
   @staticmethod
   def get_filesystem(path):
-    # type: (str) -> FileSystems
+    # type: (str) -> FileSystem
 
     """Get the correct filesystem for the specified path
     """

sdks/python/apache_beam/io/localfilesystem.py

@@ -157,7 +157,9 @@ def create(
 
     Returns: file handle with a close function for the user to use
     """
-    os.makedirs(os.path.dirname(path), exist_ok=True)
+    dirname = os.path.dirname(path)
+    if dirname:
+      os.makedirs(os.path.dirname(path), exist_ok=True)
     return self._path_open(path, 'wb', mime_type, compression_type)
 
   def open(

sdks/python/apache_beam/io/localfilesystem_test.py

@@ -17,9 +17,7 @@
 #
 
 """Unit tests for LocalFileSystem."""
-
-# pytype: skip-file
-
+import contextlib
 import filecmp
 import logging
 import os
@@ -35,6 +33,8 @@
 from apache_beam.io.filesystem import BeamIOError
 from apache_beam.options.pipeline_options import PipelineOptions
 
+# pytype: skip-file
+
 
 def _gen_fake_join(separator):
   """Returns a callable that joins paths with the given separator."""
@@ -65,10 +65,27 @@ def setUp(self):
   def tearDown(self):
     shutil.rmtree(self.tmpdir)
 
+  @contextlib.contextmanager
+  def tmpdir_as_cwd(self):
+    """Context manager that sets the current working directory to a temp dir."""
+    old_cwd = os.getcwd()
+    os.chdir(self.tmpdir)
+    try:
+      yield
+    finally:
+      os.chdir(old_cwd)
+
   def test_scheme(self):
     self.assertIsNone(self.fs.scheme())
     self.assertIsNone(localfilesystem.LocalFileSystem.scheme())
 
+  def test_create_cwd_file(self):
+    with self.tmpdir_as_cwd():
+      with self.fs.create("blah.txt") as f:
+        f.write(b"blah")
+      with open("blah.txt", "rb") as f:
+        assert f.read() == b"blah"
+
   @mock.patch('apache_beam.io.localfilesystem.os')
   def test_unix_path_join(self, *unused_mocks):
     # Test joining of Unix paths.