autotest: Add a VSIFile wrapper class

autotest/gcore/vsifile.py

@@ -37,7 +37,7 @@
 import pytest
 from lxml import etree
 
-from osgeo import gdal
+from osgeo import gdal, ogr
 
 
 ###############################################################################
@@ -1364,3 +1364,227 @@ def test_vsifile_CopyFileRestartable(tmp_vsimem):
     assert retcode == 0
     assert output_payload is None
     assert gdal.VSIStatL(dstfilename).size == 3
+
+
+###############################################################################
+# Test VSIFile helper class
+
+
+def test_vsifile_class_write_ascii(tmp_path):
+
+    fname = tmp_path / "test.txt"
+
+    lines = ["permission is hereby granted", "free of charge", "to any person"]
+
+    with gdaltest.vsi_open(fname, "w") as f:
+        assert f.tell() == 0
+
+        for line in lines:
+            f.write(line)
+            f.write("\n")
+
+    with open(fname, "r") as f:
+        assert [line.strip() for line in f.readlines()] == lines
+
+
+def test_vsifile_class_read_ascii(tmp_path):
+
+    fname = str(tmp_path / "test.txt")
+
+    lines = ["permission is hereby granted", "free of charge", "to any person"]
+
+    with open(fname, "w", newline="\n") as f:
+        for line in lines:
+            f.write(line)
+            f.write("\n")
+
+        with pytest.raises(Exception):
+            f.write(b"some bytes")
+
+    # read entire file
+    with gdaltest.vsi_open(fname, "r") as f:
+        contents = f.read()
+
+        assert type(contents) is str
+
+        lines_in = [line.strip() for line in contents.strip().split("\n")]
+        assert lines_in == lines
+
+    # read some characters
+    f = gdaltest.vsi_open(fname)
+    assert f.read(10) == "permission"
+
+    # skip a character
+    assert f.seek(1, os.SEEK_CUR) == 0
+    assert f.read(9) == "is hereby"
+
+    # seek backwards
+    assert f.seek(-2, os.SEEK_CUR) == 0
+    assert f.read(2) == "by"
+
+    # jump to beginning
+    assert f.seek(0, os.SEEK_SET) == 0
+    assert f.read(10) == "permission"
+
+    # can't jump before the beginning
+    pos = f.tell()
+    assert f.seek(-2, os.SEEK_SET) == -1
+    assert pos == f.tell()
+
+    # jump to end
+    assert f.seek(0, os.SEEK_END) == 0
+    assert f.read(10) == ""
+
+    assert f.seek(-7, os.SEEK_END) == 0
+    assert f.read() == "person\n"
+
+    f.close()
+    f.close()  # no harm in closing an already-closed file
+
+
+def test_vsifile_class_read_binary(tmp_path):
+
+    fname = tmp_path / "test.wkb"
+
+    g = ogr.CreateGeometryFromWkt("POINT (15 17)")
+    wkb = g.ExportToWkb()
+
+    with open(fname, "wb") as f:
+        f.write(wkb)
+
+    # read entire file
+    with gdaltest.vsi_open(fname, "rb") as f:
+        contents = f.read()
+
+        assert type(contents) is bytes
+
+        assert contents == wkb
+
+    # read some bytes
+    f = gdaltest.vsi_open(fname, "rb")
+    assert f.read(5) == wkb[:5]
+
+    f.seek(10, os.SEEK_SET)
+    assert f.read(5) == wkb[10:15]
+
+
+def test_vsifile_class_write_binary(tmp_path):
+
+    fname = tmp_path / "test.wkb"
+
+    g = ogr.CreateGeometryFromWkt("POINT (15 17)")
+    wkb = g.ExportToWkb()
+
+    with gdaltest.vsi_open(fname, "wb") as f:
+        f.write(wkb[:8])
+        f.write(wkb[8:])
+
+    with open(fname, "rb") as f:
+        assert f.read() == wkb
+
+
+def random_lines():
+    import random
+    import string
+
+    lines = []
+    for i in range(50):
+        lines.append(
+            "".join([random.choice(string.ascii_letters) for j in range(20 + 3 * i)])
+        )
+    lines.append(" ")
+    lines.append("")
+    lines.append("theend")
+    lines.append("")
+
+    return lines
+
+
+@pytest.mark.parametrize("terminating_newline", (True, False))
+def test_vsifile_class_line_iteration(tmp_path, terminating_newline):
+
+    fname = str(tmp_path / "test.txt")
+
+    lines_out = random_lines()
+
+    with open(fname, "w") as f:
+        for line in lines_out:
+            f.write(line)
+            f.write("\n")
+
+        if not terminating_newline:
+            f.write("lastline")
+            lines_out.append("lastline")
+
+    with gdaltest.vsi_open(fname) as f:
+        lines_in = [line for line in f]
+
+    assert lines_in == lines_out
+
+
+def test_vsifile_class_binary_line_iteration(tmp_path):
+
+    fname = str(tmp_path / "test.txt")
+
+    lines_out = [x.encode() for x in random_lines()]
+
+    with open(fname, "wb") as f:
+        for line in lines_out:
+            f.write(line)
+            f.write(b"\n")
+
+    with gdaltest.vsi_open(fname, "rb") as f:
+        lines_in = [line for line in f]
+
+    assert lines_in == lines_out
+
+
+def test_vsifile_class_zipped_csv_reader(tmp_path):
+
+    test_csv = str(tmp_path / "input.csv")
+    test_zip = str(tmp_path / "input.zip")
+
+    import csv
+    import shutil
+    import zipfile
+
+    shutil.copy("../ogr/data/prime_meridian.csv", test_csv)
+
+    with zipfile.ZipFile(test_zip, "w") as zf:
+        zf.write(test_csv, arcname="input.csv")
+
+    with gdaltest.vsi_open(f"/vsizip/{test_zip}/input.csv") as f:
+        records = [x for x in csv.DictReader(f)]
+
+    assert len(records) == 4
+    assert (
+        records[2]["INFORMATION_SOURCE"]
+        == "Institut Geographique National (IGN), Paris"
+    )
+
+
+def test_vsifile_class_file_does_not_exist(tmp_path):
+
+    with pytest.raises(OSError, match="No such file or directory"):
+        gdaltest.vsi_open(tmp_path / "does_not_exist.txt")
+
+
+def test_vsifile_class_read_from_closed_file(tmp_path):
+
+    with gdaltest.vsi_open(tmp_path / "out.txt", "w") as f:
+        f.write("abc")
+
+    with pytest.raises(ValueError, match="closed file"):
+        f.seek(0)
+
+
+def test_vsifile_class_append(tmp_vsimem):
+
+    fname = tmp_vsimem / "out.txt"
+
+    with gdaltest.vsi_open(fname, "w") as f:
+        f.write("abc")
+    with gdaltest.vsi_open(fname, "a") as f:
+        f.write("def")
+    with gdaltest.vsi_open(fname) as f:
+        assert f.read() == "abcdef"