Let me query for direct children of a directory

This commit is contained in:
Jean-Paul Calderone 2019-02-15 15:51:49 -05:00
parent d4752bde1a
commit de9c681fe2
2 changed files with 92 additions and 0 deletions

View File

@ -88,6 +88,15 @@ def get_magicfolderdb(dbfile, stderr=sys.stderr,
print >>stderr, e
return None
class LocalPath(object):
@classmethod
def fromrow(self, row):
p = LocalPath()
p.relpath_u = row[0]
p.entry = PathEntry(*row[1:])
return p
class MagicFolderDB(object):
VERSION = 1
@ -121,6 +130,42 @@ class MagicFolderDB(object):
last_downloaded_uri=last_downloaded_uri,
last_downloaded_timestamp=last_downloaded_timestamp)
def get_direct_children(self, relpath_u):
"""
Given the relative path to a directory, return ``LocalPath`` instances
representing all direct children of that directory.
"""
# It would be great to not be interpolating data into query
# statements. However, query parameters are not supported in the
# position where we need them.
sqlitesafe_relpath_u = relpath_u.replace(u"'", u"''")
statement = (
"""
SELECT
path, size, mtime_ns, ctime_ns, version, last_uploaded_uri,
last_downloaded_uri, last_downloaded_timestamp
FROM
local_files
WHERE
-- The "_" used here ensures there is at least one character
-- after the /. This prevents matching the path itself.
path LIKE '{path}/_%' AND
-- The "_" used here serves a similar purpose. This allows
-- matching directory children but avoids matching their
-- children.
path NOT LIKE '{path}/_%/_%'
"""
).format(path=sqlitesafe_relpath_u)
self.cursor.execute(statement)
rows = self.cursor.fetchall()
return list(
LocalPath.fromrow(row)
for row
in rows
)
def get_all_relpaths(self):
"""
Retrieve a set of all relpaths of files that have had an entry in magic folder db

View File

@ -534,6 +534,53 @@ class MagicFolderDbTests(SyncTestCase):
self.assertTrue(entry is not None)
self.assertEqual(entry.last_downloaded_uri, content_uri)
def test_get_direct_children(self):
"""
``get_direct_children`` returns a list of ``PathEntry`` representing each
local file in the database which is a direct child of the given path.
"""
def add_file(relpath_u):
self.db.did_upload_version(
relpath_u=relpath_u,
version=0,
last_uploaded_uri=None,
last_downloaded_uri=None,
last_downloaded_timestamp=1234,
pathinfo=get_pathinfo(self.temp),
)
paths = [
u"some_random_file",
u"the_target_directory_is_elsewhere",
u"the_target_directory_is_not_this/",
u"the_target_directory_is_not_this/and_not_in_here",
u"the_target_directory/",
u"the_target_directory/foo",
u"the_target_directory/bar",
u"the_target_directory/baz",
u"the_target_directory/quux/",
u"the_target_directory/quux/exclude_grandchildren",
u"the_target_directory/quux/and_great_grandchildren/",
u"the_target_directory/quux/and_great_grandchildren/foo",
u"the_target_directory_is_over/stuff",
u"please_ignore_this_for_sure",
]
for relpath_u in paths:
add_file(relpath_u)
expected_paths = [
u"the_target_directory/foo",
u"the_target_directory/bar",
u"the_target_directory/baz",
u"the_target_directory/quux/",
]
actual_paths = list(
localpath.relpath_u
for localpath
in self.db.get_direct_children(u"the_target_directory")
)
self.assertEqual(expected_paths, actual_paths)
def iterate_downloader(magic):
return magic.downloader._processing_iteration()