locate(), update_locate_database(): path to database may be specified

osex/__init__.py

@@ -41,13 +41,21 @@ def symlink(source, link_name, relative = False, override = False, backup = True
         else:
             raise OSError("link's path '%s' already exists." % link_name)
 
-def update_locate_database():
+def update_locate_database(root_path = None, database_path = None, require_visibility = None):
     import subprocess
-    subprocess.call(['sudo', 'updatedb'])
+    cmd = ['updatedb']
+    if root_path:
+        cmd.append('--database-root')
+        cmd.append(root_path)
+    if database_path:
+        cmd.append('--output')
+        cmd.append(database_path)
+    if require_visibility is not None:
+        cmd.append('--require-visibility')
+        cmd.append('yes' if require_visibility else 'no')
+    subprocess.check_call(cmd)
 
-def locate(patterns, match_all = False, ignore_case = False, update_database = False):
-    if update_database:
-        update_locate_database()
+def locate(patterns, match_all = False, ignore_case = False, database_path = None):
     if type(patterns) is not list:
         patterns = [str(patterns)]
     patterns = [str(p) for p in patterns]
@@ -56,6 +64,9 @@ def locate(patterns, match_all = False, ignore_case = False, update_database = F
         options.append("--all")
     if ignore_case:
         options.append("--ignore-case")
+    if database_path:
+        options.append('--database')
+        options.append(database_path)
     import subprocess
     try:
         output = subprocess.check_output(["locate"] + options + patterns)
@@ -64,4 +75,4 @@ def locate(patterns, match_all = False, ignore_case = False, update_database = F
             return []
         else:
             raise e
-    return output.strip().split('\n')
+    return output.decode(sys.getfilesystemencoding()).strip().split('\n')

scripts/select-locate

@@ -4,17 +4,25 @@
 import sys
 import osex
 import curses
+import pprint
+import locale
 import argparse
 import argcomplete
 import ioex, ioex.selector
 
-def locate_select(patterns, match_all, ignore_case, update_database, multiple):
+def locate_select(patterns, match_all, ignore_case, update_database, multiple, database_path, update_require_visibility):
+
+    if update_database:
+        osex.update_locate_database(
+                database_path = database_path,
+                require_visibility = update_require_visibility
+                )
 
     paths = osex.locate(
                 patterns, 
                 match_all = match_all, 
-                ignore_case = ignore_case, 
-                update_database = update_database
+                ignore_case = ignore_case,
+                database_path = database_path
                 )
 
     selected_paths = ioex.curses_tty_wrapper(
@@ -28,10 +36,12 @@ def locate_select(patterns, match_all, ignore_case, update_database, multiple):
 def _init_argparser():
 
     argparser = argparse.ArgumentParser(description = None)
+    argparser.add_argument('--database-path')
     argparser.add_argument('--ignore-case', action='store_true')
     argparser.add_argument('--match-all', action='store_true')
     argparser.add_argument('--multiple', action='store_true')
     argparser.add_argument('--update-database', action='store_true')
+    argparser.add_argument('--update-require-visibility', choices = ['yes', 'no'])
     argparser.add_argument('patterns', metavar = 'pattern', nargs = '+')
     return argparser
 
@@ -40,13 +50,18 @@ def main(argv):
     argparser = _init_argparser()
     argcomplete.autocomplete(argparser)
     args = argparser.parse_args(argv)
+    params = vars(args)
 
-    paths = locate_select(**vars(args))
+    if args.update_require_visibility == 'yes':
+        args.update_require_visibility = True
+    elif args.update_require_visibility == 'no':
+        args.update_require_visibility = False
 
+    paths = locate_select(**vars(args))
     if paths is None:
         return 1
     else:
-        print('\n'.join(paths))
+        print('\n'.join(paths).encode(locale.getpreferredencoding()))
         return 0
 
 if __name__ == "__main__":

setup.py

@@ -5,12 +5,12 @@ import glob
 setup(
     name = 'osex',
     packages = ['osex'],
-    version = '0.3',
+    version = '0.3.1',
     description = 'extension for python\'s build-in operating system interface',
     author = 'Fabian Peter Hammerle',
     author_email = 'fabian.hammerle@gmail.com',
     url = 'https://github.com/fphammerle/osex',
-    download_url = 'https://github.com/fphammerle/osex/tarball/0.3',
+    download_url = 'https://github.com/fphammerle/osex/tarball/0.3.1',
     keywords = [],
     classifiers = [],
     scripts = glob.glob('scripts/*'),

tests/test_locate.py

@@ -0,0 +1,97 @@
+# coding=utf-8
+
+import pytest
+
+import osex
+
+import os
+
+def test_local_db_files(tmpdir):
+    test_dir_path = tmpdir.strpath
+    os.chdir(test_dir_path)
+    # prepare dir
+    open('file1', 'w').close()
+    open('file2', 'w').close()
+    os.makedirs(os.path.join('dir', 'subdir1'))
+    os.makedirs(os.path.join('dir', 'subdir2'))
+    # update db
+    database_path = os.path.join(test_dir_path, 'db')
+    osex.update_locate_database(database_path = database_path, root_path = test_dir_path, require_visibility = False)
+    # locate
+    results = osex.locate(
+        [test_dir_path + '*file*'],
+        database_path = database_path
+        )
+    assert results == [os.path.join(test_dir_path, p) for p in [
+        'file1',
+        'file2',
+        ]]
+
+def test_local_db_files_unicode(tmpdir):
+    test_dir_path = tmpdir.strpath
+    os.chdir(test_dir_path)
+    # prepare dir
+    open(u'file-ä', 'w').close()
+    open(u'file-Ö', 'w').close()
+    open(u'file-ß', 'w').close()
+    # update db
+    database_path = os.path.join(test_dir_path, 'db')
+    osex.update_locate_database(database_path = database_path, root_path = test_dir_path, require_visibility = False)
+    # locate
+    results = osex.locate(
+        [test_dir_path + '*file*'],
+        database_path = database_path
+        )
+    expected_results = [os.path.join(test_dir_path, p) for p in [
+        u'file-ä',
+        u'file-Ö',
+        u'file-ß',
+        ]]
+    assert results.sort() == expected_results.sort()
+
+def test_local_db_dirs(tmpdir):
+    test_dir_path = tmpdir.strpath
+    os.chdir(test_dir_path)
+    # prepare dir
+    open('file1', 'w').close()
+    open('file2', 'w').close()
+    os.makedirs(os.path.join('dir', 'subdir1'))
+    os.makedirs(os.path.join('dir', 'subdir2'))
+    # update db
+    database_path = os.path.join(test_dir_path, 'db')
+    osex.update_locate_database(database_path = database_path, root_path = test_dir_path, require_visibility = False)
+    # locate
+    results = osex.locate(
+        [test_dir_path + '*dir*'],
+        database_path = database_path
+        )
+    assert results == [os.path.join(test_dir_path, p) for p in [
+        'dir',
+        os.path.join('dir', 'subdir1'),
+        os.path.join('dir', 'subdir2'),
+        ]]
+
+def test_local_db_subtree(tmpdir):
+    test_dir_path = tmpdir.strpath
+    os.chdir(test_dir_path)
+    # prepare dir
+    os.makedirs(os.path.join('dir1', 'subdir1'))
+    os.makedirs(os.path.join('dir1', 'subdir2'))
+    os.makedirs(os.path.join('dir2', 'subdir3'))
+    os.makedirs(os.path.join('dir2', 'subdir4'))
+    # update db
+    database_path = os.path.join(test_dir_path, 'db')
+    osex.update_locate_database(
+            database_path = database_path,
+            root_path = os.path.join(test_dir_path, 'dir2'),
+            require_visibility = False
+            )
+    # locate
+    results = osex.locate(
+        [test_dir_path + '*subdir*'],
+        database_path = database_path
+        )
+    assert results == [os.path.join(test_dir_path, p) for p in [
+        os.path.join('dir2', 'subdir3'),
+        os.path.join('dir2', 'subdir4'),
+        ]]