ARROW-1456: [Python] Run s3fs unit tests in Travis CI

.travis.yml

@@ -105,8 +105,10 @@ matrix:
     - ARROW_TRAVIS_COVERAGE=1
     - ARROW_TRAVIS_FLIGHT=1
     - ARROW_TRAVIS_OPTIONAL_INSTALL=1
+    - ARROW_TRAVIS_PARQUET=1
     - ARROW_TRAVIS_PYTHON_DOCS=1
     - ARROW_TRAVIS_PYTHON_JVM=1
+    - ARROW_TRAVIS_S3=1
     - ARROW_TRAVIS_USE_SYSTEM_JAVA=1
     - ARROW_TRAVIS_USE_TOOLCHAIN=1
     - ARROW_TRAVIS_VALGRIND=1

ci/conda_env_python.yml

@@ -23,5 +23,6 @@ pandas
 pytest
 pytest-faulthandler
 pytz
+s3fs
 setuptools
 setuptools_scm=3.2.0

ci/docker_install_conda.sh

@@ -40,3 +40,9 @@ conda config --add channels conda-forge
 
 # Update packages
 conda update --all -q -y
+
+# Download the Minio S3 server into PATH
+export S3FS_DIR=/usr/local/bin/
+mkdir -p $S3FS_DIR
+wget --directory-prefix $S3FS_DIR https://dl.min.io/server/minio/release/linux-amd64/minio
+chmod +x $S3FS_DIR/minio

python/pyarrow/tests/conftest.py

@@ -114,6 +114,13 @@
 except ImportError:
     pass
 
+try:
+    from distutils.spawn import find_executable  # noga
+    if find_executable('minio') or find_executable('minio.exe'):
+        defaults['s3'] = True
+except ImportError:
+    pass
+
 try:
     import tensorflow  # noqa
     defaults['tensorflow'] = True

python/pyarrow/tests/test_parquet.py

@@ -1808,30 +1808,44 @@ def test_filters_read_table(tempdir):
 
 
 @pytest.yield_fixture
-def s3_example():
-    access_key = os.environ['PYARROW_TEST_S3_ACCESS_KEY']
-    secret_key = os.environ['PYARROW_TEST_S3_SECRET_KEY']
-    bucket_name = os.environ['PYARROW_TEST_S3_BUCKET']
-
+@pytest.mark.s3
+def s3_example(tempdir):
     import s3fs
-    fs = s3fs.S3FileSystem(key=access_key, secret=secret_key)
+    import subprocess
 
-    test_dir = guid()
+    bucket_name = 'minio_bucket'
+    minio_access_key = 'minio'
+    minio_secret_key = 'miniopass'
+    endpoint = '127.0.0.1:9123'
+    client_kwargs = {'endpoint_url': 'http://' + endpoint}
+    os.environ["MINIO_ACCESS_KEY"] = minio_access_key
+    os.environ["MINIO_SECRET_KEY"] = minio_secret_key
+    os.makedirs(tempdir / bucket_name)
+
+    cmd = ['minio', 'server', '--compat', '--address', endpoint, str(tempdir)]
+    pro = subprocess.Popen(cmd, stdout=subprocess.PIPE)
 
+    fs = s3fs.S3FileSystem(key=minio_access_key, secret=minio_secret_key,
+                           client_kwargs=client_kwargs)
+
+    test_dir = guid()
     bucket_uri = 's3://{0}/{1}'.format(bucket_name, test_dir)
     fs.mkdir(bucket_uri)
     yield fs, bucket_uri
     fs.rm(bucket_uri, recursive=True)
+    pro.kill()
 
 
 @pytest.mark.pandas
 @pytest.mark.s3
 def test_read_partitioned_directory_s3fs(s3_example):
+    import pathlib
     from pyarrow.filesystem import S3FSWrapper
 
     fs, bucket_uri = s3_example
     wrapper = S3FSWrapper(fs)
-    _partition_test_for_filesystem(wrapper, bucket_uri)
+    bucket_path = pathlib.PurePosixPath(bucket_uri.replace('s3://', ''))
+    _partition_test_for_filesystem(wrapper, bucket_path)
 
     # Check that we can auto-wrap
     dataset = pq.ParquetDataset(bucket_uri, filesystem=fs)
@@ -1885,7 +1899,14 @@ def _visit_level(base_dir, level, part_keys):
             this_part_keys = part_keys + [(name, value)]
 
             level_dir = base_dir / '{0}={1}'.format(name, value)
-            fs.mkdir(level_dir)
+            is_s3 = hasattr(fs, 'fs')
+
+            if is_s3:
+                fs.mkdir(level_dir, create_parents=False)
+                path = 's3://' + (level_dir / '_SUCCESS').as_posix()
+            else:
+                fs.mkdir(level_dir)
+                path = (level_dir / '_SUCCESS')
 
             if level == DEPTH - 1:
                 # Generate example data
@@ -1897,10 +1918,16 @@ def _visit_level(base_dir, level, part_keys):
                     _write_table(part_table, f)
                 assert fs.exists(file_path)
 
-                (level_dir / '_SUCCESS').touch()
+                if is_s3:
+                    fs.fs.touch(path)
+                else:
+                    path.touch()
             else:
                 _visit_level(level_dir, level + 1, this_part_keys)
-                (level_dir / '_SUCCESS').touch()
+                if is_s3:
+                    fs.fs.touch(path)
+                else:
+                    path.touch()
 
     _visit_level(base_dir, 0, [])