Repository: iterative/dvc

Issue:
Cannot add file having name with substring of a folder as prefix in s3
### Steps to reproduce
1. Upload two files in s3: `folder/data/data.csv` and` folder/datasets.md`.
2. Setup remotes and caches.
```sh
dvc remote add -f s3 s3://dvc-temp/folder
dvc remote add -f cache remote://s3/cache
dvc config cache.s3 cache
```
3. `dvc run -d remote://s3/data 'echo hello world'`

### Outcome
```
Running command:
        echo hello world
hello world
ERROR: unexpected error - '/folder/datasets.md' does not start with '/folder/data'
```

#### Version
```
$ dvc version
Python version: 3.6.6
Platform: Linux-5.3.12-arch1-1-x86_64-with-arch
Binary: False
Package: None
Filesystem type (cache directory): ('ext4', '/dev/sda9')
Filesystem type (workspace): ('ext4', '/dev/sda9')
```

#### Script to reproduce
```sh
#! /usr/bin/env bash

export AWS_ACCESS_KEY_ID='testing'
export AWS_SECRET_ACCESS_KEY='testing'
export AWS_SECURITY_TOKEN='testing'
export AWS_SESSION_TOKEN='testing'

moto_server s3 &> /dev/null &

python -c '
import boto3

session = boto3.session.Session()
s3 = session.client("s3", endpoint_url="http://localhost:5000")
s3.create_bucket(Bucket="dvc-temp")

s3.put_object(Bucket="dvc-temp", Key="folder/data/data.csv")
s3.put_object(Bucket="dvc-temp", Key="folder/datasets.md", Body="### Datasets")
'

temp=$(mktemp -d)
cd $temp

dvc init --no-scm
dvc remote add -f s3 s3://dvc-temp/folder
dvc remote modify s3 endpointurl http://localhost:5000
dvc remote add -f cache remote://s3/cache
dvc config cache.s3 cache

dvc run -d remote://s3/data 'echo hello world'
```

#### Analysis:
1. This is due to `walk_files` implementation in `RemoteS3` looking via prefix instead of `/<prefix`> to walk files. Either, `walk_files` should get directory path or should just append it itself.

https://github.com/iterative/dvc/blob/0404a2324e497667a8b7d0ab0bd2b37db8c97e4c/dvc/remote/s3.py#L282

Or, I'd prefer it to be handled when collecting the directory.
https://github.com/iterative/dvc/blob/caa67c725e1e351ed122bdad17db0f29a8e73c39/dvc/remote/base.py#L196

2. Again, the logic of `exists` looks flawed. Say, you have `data/subdir-file.txt` and `data/subdir/1` files. When adding `data/subdir`, the first result could be `subdir-file.txt` which matches `startswith`, therefore, the `exists()` will return True, but in reality, `subdir` does not exist.
So, the function should check if it's a directory, and should loop through all results of `_list_paths()` till it finds the exact match (not sure, how expensive this will be).

https://github.com/iterative/dvc/blob/caa67c725e1e351ed122bdad17db0f29a8e73c39/dvc/remote/s3.py#L208-L211

Discussion:

Patch:
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
index 8013f9bb9..8dd2bdf48 100644
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -5,6 +5,7 @@ import logging
 import os
 import threading
 
+from botocore.exceptions import ClientError
 from funcy import cached_property, wrap_prop
 
 from dvc.config import Config
@@ -207,10 +208,26 @@ class RemoteS3(RemoteBASE):
     def list_cache_paths(self):
         return self._list_paths(self.path_info)
 
+    def isfile(self, path_info):
+        if path_info.path.endswith("/"):
+            return False
+
+        try:
+            self.s3.head_object(Bucket=path_info.bucket, Key=path_info.path)
+        except ClientError as exc:
+            if exc.response["Error"]["Code"] != "404":
+                raise
+            return False
+
+        return True
+
     def exists(self, path_info):
-        dir_path = path_info / ""
-        fname = next(self._list_paths(path_info, max_items=1), "")
-        return path_info.path == fname or fname.startswith(dir_path.path)
+        """Check if the blob exists. If it does not exist,
+        it could be a part of a directory path.
+
+        eg: if `data/file.txt` exists, check for `data` should return True
+        """
+        return self.isfile(path_info) or self.isdir(path_info)
 
     def makedirs(self, path_info):
         # We need to support creating empty directories, which means
@@ -279,7 +296,7 @@ class RemoteS3(RemoteBASE):
         )
 
     def walk_files(self, path_info, max_items=None):
-        for fname in self._list_paths(path_info, max_items):
+        for fname in self._list_paths(path_info / "", max_items):
             if fname.endswith("/"):
                 continue
 


Test patch:
diff --git a/tests/func/test_api.py b/tests/func/test_api.py
index a970d3cd5..ddf101479 100644
--- a/tests/func/test_api.py
+++ b/tests/func/test_api.py
@@ -75,7 +75,7 @@ def run_dvc(*argv):
 
 
 @pytest.mark.parametrize("remote_url", remote_params, indirect=True)
-def test_get_url(remote_url, repo_dir, dvc_repo):
+def test_get_url(repo_dir, dvc_repo, remote_url):
     run_dvc("remote", "add", "-d", "upstream", remote_url)
     dvc_repo.add(repo_dir.FOO)
 
@@ -84,7 +84,7 @@ def test_get_url(remote_url, repo_dir, dvc_repo):
 
 
 @pytest.mark.parametrize("remote_url", remote_params, indirect=True)
-def test_get_url_external(remote_url, dvc_repo, erepo):
+def test_get_url_external(dvc_repo, erepo, remote_url):
     _set_remote_url_and_commit(erepo.dvc, remote_url)
 
     # Using file url to force clone to tmp repo
@@ -94,7 +94,7 @@ def test_get_url_external(remote_url, dvc_repo, erepo):
 
 
 @pytest.mark.parametrize("remote_url", all_remote_params, indirect=True)
-def test_open(remote_url, repo_dir, dvc_repo):
+def test_open(repo_dir, dvc_repo, remote_url):
     run_dvc("remote", "add", "-d", "upstream", remote_url)
     dvc_repo.add(repo_dir.FOO)
     run_dvc("push")
@@ -107,7 +107,7 @@ def test_open(remote_url, repo_dir, dvc_repo):
 
 
 @pytest.mark.parametrize("remote_url", all_remote_params, indirect=True)
-def test_open_external(remote_url, dvc_repo, erepo):
+def test_open_external(dvc_repo, erepo, remote_url):
     erepo.dvc.scm.checkout("branch")
     _set_remote_url_and_commit(erepo.dvc, remote_url)
     erepo.dvc.scm.checkout("master")
@@ -127,7 +127,7 @@ def test_open_external(remote_url, dvc_repo, erepo):
 
 
 @pytest.mark.parametrize("remote_url", all_remote_params, indirect=True)
-def test_missing(remote_url, repo_dir, dvc_repo):
+def test_missing(repo_dir, dvc_repo, remote_url):
     run_dvc("add", repo_dir.FOO)
     run_dvc("remote", "add", "-d", "upstream", remote_url)
 
diff --git a/tests/unit/remote/test_s3.py b/tests/unit/remote/test_s3.py
index 49fc3dbb8..bb0abe594 100644
--- a/tests/unit/remote/test_s3.py
+++ b/tests/unit/remote/test_s3.py
@@ -13,10 +13,12 @@ def remote():
         ├── data
         │  ├── alice
         │  ├── alpha
+        │  ├── subdir-file.txt
         │  └── subdir
         │     ├── 1
         │     ├── 2
         │     └── 3
+        ├── data1.txt
         ├── empty_dir
         ├── empty_file
         └── foo
@@ -26,6 +28,7 @@ def remote():
         s3 = remote.s3
 
         s3.create_bucket(Bucket="bucket")
+        s3.put_object(Bucket="bucket", Key="data1.txt", Body=b"")
         s3.put_object(Bucket="bucket", Key="empty_dir/")
         s3.put_object(Bucket="bucket", Key="empty_file", Body=b"")
         s3.put_object(Bucket="bucket", Key="foo", Body=b"foo")
@@ -34,6 +37,9 @@ def remote():
         s3.put_object(Bucket="bucket", Key="data/subdir/1", Body=b"1")
         s3.put_object(Bucket="bucket", Key="data/subdir/2", Body=b"2")
         s3.put_object(Bucket="bucket", Key="data/subdir/3", Body=b"3")
+        s3.put_object(
+            Bucket="bucket", Key="data/subdir-file.txt", Body=b"subdir"
+        )
 
         yield remote
 
@@ -66,6 +72,7 @@ def test_exists(remote):
         (True, "data/subdir/1"),
         (False, "data/al"),
         (False, "foo/"),
+        (True, "data1.txt"),
     ]
 
     for expected, path in test_cases:
@@ -76,9 +83,11 @@ def test_walk_files(remote):
     files = [
         remote.path_info / "data/alice",
         remote.path_info / "data/alpha",
+        remote.path_info / "data/subdir-file.txt",
         remote.path_info / "data/subdir/1",
         remote.path_info / "data/subdir/2",
         remote.path_info / "data/subdir/3",
+        remote.path_info / "data1.txt",
         remote.path_info / "empty_file",
         remote.path_info / "foo",
     ]
@@ -109,3 +118,23 @@ def test_makedirs(remote):
     assert not remote.exists(empty_dir)
     remote.makedirs(empty_dir)
     assert remote.exists(empty_dir)
+
+
+def test_isfile(remote):
+    test_cases = [
+        (False, "empty_dir/"),
+        (True, "empty_file"),
+        (True, "foo"),
+        (True, "data/alice"),
+        (True, "data/alpha"),
+        (True, "data/subdir/1"),
+        (True, "data/subdir/2"),
+        (True, "data/subdir/3"),
+        (False, "data/subdir/empty_dir/"),
+        (False, "data/subdir/1/"),
+        (False, "something-that-does-not-exist"),
+        (False, "empty_dir"),
+    ]
+
+    for expected, path in test_cases:
+        assert remote.isfile(remote.path_info / path) == expected