Check for rglobs in include/exclude filters

blobxfer/models/__init__.py

@@ -71,10 +71,24 @@ def add_includes(self, includes):
                 includes = list(includes)
             else:
                 includes = [includes]
+        # remove any starting rglob spec
+        incl = []
+        for inc in includes:
+            tmp = pathlib.Path(inc).parts
+            if tmp[0] == '**':
+                if len(tmp) == 1:
+                    continue
+                else:
+                    incl.append(str(pathlib.Path(*tmp[1:])))
+            else:
+                incl.append(inc)
+        # check for any remaining rglob specs
+        if any(['**' in x for x in incl]):
+            raise ValueError('invalid include specification containing "**"')
         if self._include is None:
-            self._include = includes
+            self._include = incl
         else:
-            self._include.extend(includes)
+            self._include.extend(incl)
 
     def add_excludes(self, excludes):
         # type: (_BaseSourcePaths, list) -> None
@@ -87,10 +101,24 @@ def add_excludes(self, excludes):
                 excludes = list(excludes)
             else:
                 excludes = [excludes]
+        # remove any starting rglob spec
+        excl = []
+        for exc in excludes:
+            tmp = pathlib.Path(exc).parts
+            if tmp[0] == '**':
+                if len(tmp) == 1:
+                    continue
+                else:
+                    excl.append(str(pathlib.Path(*tmp[1:])))
+            else:
+                excl.append(exc)
+        # check for any remaining rglob specs
+        if any(['**' in x for x in excl]):
+            raise ValueError('invalid exclude specification containing "**"')
         if self._exclude is None:
-            self._exclude = excludes
+            self._exclude = excl
         else:
-            self._exclude.extend(excludes)
+            self._exclude.extend(excl)
 
     def add_path(self, path):
         # type: (_BaseSourcePaths, str) -> None

tests/test_blobxfer_models_upload.py

@@ -87,19 +87,28 @@ def test_localsourcepaths_files(tmpdir):
     defpath.join('moo.cow').write('y')
 
     a = upload.LocalSourcePath()
+    a.add_includes('**')
     a.add_includes('*.txt')
     a.add_includes(('moo.cow', '*blah*'))
+    with pytest.raises(ValueError):
+        a.add_includes('**/**/*')
+    a.add_excludes('**')
     a.add_excludes('**/blah.x')
+    with pytest.raises(ValueError):
+        a.add_excludes('**/**/blah.x')
     a.add_excludes(['world.txt'])
     a.add_path(str(tmpdir))
     a_set = set()
     for file in a.files():
         sfile = str(file.parent_path / file.relative_path)
         a_set.add(sfile)
 
+    assert len(a._include) == 3
+    assert len(a._exclude) == 2
+
     assert not a.can_rename()
     assert len(a.paths) == 1
-    assert str(abcpath.join('blah.x')) not in a_set
+    assert str(abcpath.join('blah.x')) in a_set
     assert str(defpath.join('world.txt')) in a_set
     assert str(defpath.join('moo.cow')) not in a_set