File write feature (#6)

* Read File Working

Files can be read from local file system. Seek works as well. More thorough tests needed.

* Linted with flake8

* Typed everything

* Mostly There

* sign

* Checks passed

* Should pass now

* Add fsspec dependency to library

Cannot add XRootD dependency because it is not properly packaged :(

* Revert "Merge branch 'file-system-ops-2' into File-Open-Feature"

This reverts commit 2a27584, reversing
changes made to 33d710e.

* Add file open and read tests to CI

* Remove windows and pypy CI tests

xrootd is not available in conda for windows, and pypy not in conda

* Upload chunk added

* Flush and upload chunk finalized

* lint

* lint

* Tried write test, fails

* Writing works

* lint

* Append Added

* More read tests

* lint

* readuntil test

byte mode only

* close file condition

* xrootd version limit

* changed python version

* version change test

* prints added

* write raise

* ping test added

* lint

* lint

* removed macos

Co-authored-by: Nick Smith <[email protected]>

.github/workflows/ci.yml

@@ -38,7 +38,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ["3.7", "3.10"]
-        runs-on: [ubuntu-latest, macos-latest]
+        runs-on: [ubuntu-latest]
 
     steps:
       - uses: actions/checkout@v3
@@ -54,7 +54,7 @@ jobs:
       - name: Install build environment
         shell: bash -l {0}
         run: |
-          mamba install -c conda-forge python==${{ matrix.python-version }} xrootd
+          mamba install -c conda-forge python=${{ matrix.python-version }} "xrootd>=5.4.3"
       - name: Install package
         shell: bash -l {0}
         run: python -m pip install .[test]

src/fsspec_xrootd/xrootd.py

@@ -170,7 +170,7 @@ def __init__(
         if "x" in mode:
             self.mode = OpenFlags.NEW
         elif "a" in mode:
-            self.mode = OpenFlags.APPEND
+            self.mode = OpenFlags.UPDATE
         elif "+" in mode:
             self.mode = OpenFlags.UPDATE
         elif "w" in mode:
@@ -192,6 +192,11 @@ def __init__(
         if not status.ok:
             raise OSError(f"File did not open properly: {status.message}")
 
+        self.metaOffset = 0
+        if "a" in mode:
+            _stats, _deets = self._myFile.stat()
+            self.metaOffset = _deets.size
+
         self.path = path
         self.fs = fs
         self.mode = mode
@@ -230,20 +235,53 @@ def __init__(
             )
         else:
             self.buffer = io.BytesIO()
-            self.offset = None
             self.forced = False
             self.location = None
+            self.offset = 0
 
     def _fetch_range(self, start: int, end: int) -> Any:
-        status, data = self._myFile.read(start, end - start)
+        status, data = self._myFile.read(
+            self.metaOffset + start, self.metaOffset + end - start
+        )
         if not status.ok:
             raise OSError(f"File did not read properly: {status.message}")
         return data
 
+    def flush(self, force: bool = False) -> None:
+        if self.closed:
+            raise ValueError("Flush on closed file")
+        if force and self.forced:
+            raise ValueError("Force flush cannot be called more than once")
+        if force:
+            self.forced = True
+
+        if self.mode not in {"wb", "ab"}:
+            # no-op to flush on read-mode
+            return
+
+        if not force and self.buffer.tell() < self.blocksize:
+            # Defer write on small block
+            return
+
+        if self._upload_chunk(final=force) is not False:
+            self.offset += self.buffer.seek(0, 2)
+            self.buffer = io.BytesIO()
+
+    def _upload_chunk(self, final: bool = False) -> Any:
+        status, _n = self._myFile.write(
+            self.buffer.getvalue(), self.offset + self.metaOffset, self.buffer.tell()
+        )
+        if final:
+            self.closed
+            self.close()
+        if not status.ok:
+            raise OSError(f"File did not write properly: {status.message}")
+        return status.ok
+
     def close(self) -> None:
         if getattr(self, "_unclosable", False):
             return
-        if self.closed:
+        if self.closed or not self._myFile.is_open():
             return
         if self.mode == "rb":
             self.cache = None

tests/test_basicio.py

@@ -8,7 +8,8 @@
 import fsspec
 import pytest
 
-TESTDATA = "apple banana orange grape"
+TESTDATA = "apple\nbanana\norange\ngrape"
+TESTWRITEDATA = "the end is never the end is never the end"
 
 
 @pytest.fixture(scope="module")
@@ -32,6 +33,15 @@ def test_broken_server():
         _ = fsspec.open("root://localhost:12345/")
 
 
+def test_ping(localserver):
+    from XRootD import client
+
+    fs = client.FileSystem(localserver)
+    status, _n = fs.ping()
+    if not status.ok:
+        raise OSError(f"Server did not run properly: {status.message}")
+
+
 def test_read_xrd(localserver):
     from XRootD import client
 
@@ -49,3 +59,32 @@ def test_read_xrd(localserver):
 def test_read_fsspec(localserver):
     with fsspec.open(localserver + "/testfile.txt", "rt") as f:
         assert f.read() == TESTDATA
+        f.seek(0)
+        assert f.readline() == "apple\n"
+        f.seek(0)
+        lns = f.readlines()
+        assert lns[2] == "orange\n"
+        f.seek(1)
+        assert f.read(1) == "p"
+
+    with fsspec.open(localserver + "/testfile.txt", "rb") as f:
+        assert f.readuntil(b"e") == b"apple"
+
+    fs, token, path = fsspec.get_fs_token_paths(localserver + "/testfile.txt", "rt")
+    assert fs.read_block(path[0], 0, 4) == b"appl"
+
+
+def test_write_fsspec(localserver):
+    with fsspec.open(localserver + "/testfile2.txt", "wt") as f:
+        f.write(TESTWRITEDATA)
+        f.flush()
+    with fsspec.open(localserver + "/testfile2.txt", "rt") as f:
+        assert f.read() == TESTWRITEDATA
+
+
+def test_append_fsspec(localserver):
+    with fsspec.open(localserver + "/testfile2.txt", "at") as f:
+        f.write(TESTWRITEDATA)
+        f.flush()
+    with fsspec.open(localserver + "/testfile2.txt", "rt") as f:
+        assert f.read() == TESTWRITEDATA + TESTWRITEDATA