add test for filesystem plugins

Author: vnghia

tests/test_fs_plugins.py

@@ -0,0 +1,106 @@
+# Copyright 2020 The TensorFlow Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License.  You may obtain a copy of
+# the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations under
+# the License.
+# ==============================================================================
+"""Tests for file system plugins"""
+
+import os
+import time
+
+import pytest
+import tensorflow as tf
+import tensorflow_io as tfio  # pylint: disable=unused-import
+
+S3_URI = "s3e"
+AZ_URI = "az"
+HDFS_URI = "hdfse"
+VIEWFS_URI = "viewfse"
+HAR_URI = "hare"
+GCS_URI = "gse"
+
+
+def setup_env(uri, envs):
+    # when `envs is None`, it is the default case without
+    # additional envs and should run with every plugins.
+    if envs is None:
+        pytest.skip()
+
+    uri_env, additional_env = envs
+    if uri != uri_env:
+        pytest.skip()
+
+    # TODO(vnvo2409): Use monkeypatch
+    # ------------------------------------ s3 ------------------------------------ #
+    if uri == S3_URI:
+        os.environ["AWS_REGION"] = "us-east-1"
+        os.environ["AWS_ACCESS_KEY_ID"] = "ACCESS_KEY"
+        os.environ["AWS_SECRET_ACCESS_KEY"] = "SECRET_KEY"
+        os.environ["S3_ENDPOINT"] = "localhost:4566"
+        os.environ["S3_USE_HTTPS"] = "0"
+        os.environ["S3_VERIFY_SSL"] = "0"
+
+    for key, value in additional_env.items():
+        os.environ[key] = value
+
+
+@pytest.fixture(
+    scope="session",
+    autouse=True,
+    params=[
+        S3_URI,
+        pytest.param(AZ_URI, marks=pytest.mark.skip(reason="TODO")),
+        pytest.param(HDFS_URI, marks=pytest.mark.skip(reason="TODO")),
+        pytest.param(VIEWFS_URI, marks=pytest.mark.skip(reason="TODO")),
+        pytest.param(HAR_URI, marks=pytest.mark.skip(reason="TODO")),
+        pytest.param(GCS_URI, marks=pytest.mark.skip(reason="TODO")),
+    ],
+)
+def uri_builder(request):
+    uri = request.param
+    bucket_name = "{}{}e/tf-io-test".format(uri, time.time())
+
+    def get_uri(object_name):
+        return "{}://{}/{}".format(uri, bucket_name, object_name)
+
+    tf.io.gfile.makedirs("{}://{}/".format(uri, bucket_name))
+    return uri, bucket_name, get_uri
+
+
+@pytest.mark.parametrize("envs", [None])
+def test_init(uri_builder, envs):
+    uri, _, get_uri = uri_builder
+    setup_env(uri, envs)
+    assert tf.io.gfile.exists(get_uri("")) is True
+
+
+@pytest.mark.parametrize(
+    "envs", [None, (S3_URI, {"S3_DISABLE_MULTI_PART_DOWNLOAD": "1"})]
+)
+def test_write_read_file(uri_builder, envs):
+    """Test write/read file."""
+    uri, _, get_uri = uri_builder
+    setup_env(uri, envs)
+
+    # Setup and check preconditions.
+    file_name = get_uri("writereadfile")
+    if tf.io.gfile.exists(file_name):
+        tf.io.gfile.remove(file_name)
+
+    # Write data.
+    with tf.io.gfile.GFile(file_name, "w") as w:
+        w.write("Hello\n, world!")
+
+    # Read data.
+    with tf.io.gfile.GFile(file_name, "r") as r:
+        file_read = r.read()
+        assert file_read == "Hello\n, world!"