Merge branch 'f-winlist' into 'dev'

ENH: Referenced files

See merge request caosdb/caosdb-advanced-user-tools!48

src/caosadvancedtools/table_importer.py

@@ -82,6 +82,15 @@ def date_converter(val, fmt="%Y-%m-%d"):
         return datetime.strptime(val, fmt)
 
 
+def win_path_list_converter(val):
+    """
+    checks whether the value looks like a list of windows paths and converts
+    it to posix paths
+    """
+    paths = val.split(",")
+    return [win_path_converter(p) for p in paths]
+
+
 def win_path_converter(val):
     """
     checks whether the value looks like a windows path and converts it to posix

src/caosadvancedtools/utils.py

@@ -27,6 +27,7 @@ import os
 import pathlib
 
 import caosdb as db
+from caosdb.exceptions import TransactionError
 
 logger = logging.getLogger(__name__)
 
@@ -81,36 +82,51 @@ def read_field_as_list(field):
         return [field]
 
 
-def find_file_included_by(glob):
+def get_referenced_files(glob, prefix=None, filename=None, location=None):
     """
-    Executes a query that looks for files included by a glob. The glob needs
-    to be according to CaosDB rules.
+    queries the database for files referenced by the provided glob
 
-    Returns a container.
-    """
-
-    query_string = "FIND file which is stored at {}".format(glob)
-
-    return db.execute_query(query_string)
-
-
-def assure_absolute_path_in_glob(glob, prefix):
-    """
-    Prefixes a relative globs with some path.
-
-    Some times files are defined by a relative glob (e.g. "scripts/**"). In
-    order to search for such files in CaosDB, these globs are prefixed with the
-    current location.
-
-    A relative glob is identified by a missing "/" in the beginning.
+    Parameters:
+    glob: the glob referencing the file(s)
+    prefix: the glob can be relative to some path, in that case that path needs
+            to be given as prefix
+    filename: the file in which the glob is given (used for error messages)
+    location: the location in the file in which the glob is given (used for
+              error messages)
     """
 
-    if not glob.startswith("/"):
-        glob = os.path.normpath(os.path.join(prefix, glob))
-    else:
-        glob = os.path.normpath(glob)
-
-    return glob
+    orig_glob = glob
+
+    if not glob.startswith("/") and prefix is not None:
+        glob = os.path.join(prefix, glob)
+    glob = os.path.normpath(glob)
+    try:
+        query_string = "FIND file which is stored at {}".format(glob)
+        files = db.execute_query(query_string)
+    except TransactionError:
+        logger.error(
+            "In {} in file \n{}\nthe expression '{}' does not "
+            "allow a search for files. Please make sure "
+            "it is valid.".format(
+                location,
+                filename,
+                orig_glob
+            )
+        )
+
+        return []
+
+    if len(files) == 0:
+        logger.warning(
+            "In {} in file \n{}\nthe expression '{}' does not "
+            "reference any known files".format(
+                location,
+                filename,
+                orig_glob
+            )
+        )
+
+    return files
 
 
 def check_win_path(path, filename=None):

unittests/test_table_importer.py

@@ -28,6 +28,7 @@ from caosadvancedtools.datainconsistency import DataInconsistencyError
 from caosadvancedtools.table_importer import (XLSImporter, assure_name_format,
                                               date_converter,
                                               win_path_converter,
+                                              win_path_list_converter,
                                               yes_no_converter)
 
 
@@ -53,6 +54,11 @@ class ConverterTest(unittest.TestCase):
         self.assertRaises(ValueError, win_path_converter, "/hallo/python")
         self.assertEqual(win_path_converter(r"\this\computer"),
                          "/this/computer")
+        self.assertEqual(win_path_list_converter(r"\this\computer"),
+                         ["/this/computer"])
+        self.assertEqual(win_path_list_converter(
+            r"\this\computer,\this\computer"),
+                         ["/this/computer", "/this/computer"])
 
     def test_date(self):
         test_file = os.path.join(os.path.dirname(__file__), "date.xlsx")

unittests/test_utils.py

@@ -20,23 +20,95 @@
 # along with this program. If not, see <https://www.gnu.org/licenses/>.
 #
 # ** end header
+import logging
 import unittest
+from tempfile import NamedTemporaryFile
 
-from caosadvancedtools.utils import (assure_absolute_path_in_glob,
-                                     check_win_path, string_to_person,
-                                     )
+import caosdb as db
+from caosadvancedtools.utils import (check_win_path, get_referenced_files,
+                                     string_to_person)
+from caosdb import RecordType, configure_connection, get_config
+from caosdb.connection.mockup import MockUpResponse, MockUpServerConnection
+from caosdb.exceptions import TransactionError
 
 
-class Assure_absoluteTest(unittest.TestCase):
+class ReferencesBaseTest(unittest.TestCase):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.entities = (
+            '<Response><File name="test.npy" path="/some/path/test.npy'
+            '" id="1234"/><Query string="find record" results="1">'
+            '</Query></Response>')
 
-    def test_abs(self):
-        assert assure_absolute_path_in_glob("/hi/leute", "/Data") == "/hi/leute"
+    def setUp(self):
+        conlogger = logging.getLogger("connection")
+        conlogger.setLevel(level=logging.ERROR)
+        autlogger = logging.getLogger("authentication")
+        autlogger.setLevel(level=logging.ERROR)
+        connection = configure_connection(
+            url="unittests", username="testuser",
+            password_method="plain",
+            password="testpassword", timeout=200,
+            implementation=MockUpServerConnection)
 
-    def test_rel(self):
-        assert assure_absolute_path_in_glob("hi/leute", "/Data") == "/Data/hi/leute"
+        connection._delegate_connection.resources.append(
+            lambda **kwargs: MockUpResponse(200, {}, self.entities))
 
-    def test_compress(self):
-        assert assure_absolute_path_in_glob("../leute", "/Data") == "/leute"
+        self.logfile = NamedTemporaryFile()
+        logger = logging.getLogger()
+        logger.addHandler(logging.FileHandler(self.logfile.name))
+        logger.setLevel(logging.DEBUG)
+
+    def clear_log(self):
+        with open(self.logfile.name, "w") as lf:
+            lf.write("")
+
+    def get_log(self):
+        with open(self.logfile.name) as lf:
+            log = lf.read()
+        self.clear_log()
+
+        return log
+
+    def test_ref(self):
+        self.clear_log()
+        files = get_referenced_files("test.npy", prefix=None, filename=None,
+                                     location=None)
+        self.assertEqual(len(files), 1)
+        self.assertEqual(files[0].path, "/some/path/test.npy")
+        self.assertEqual(self.get_log(), "")
+
+
+class ReferencesFailTest(ReferencesBaseTest):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.entities = (
+            '<Response><File name="test.npy" path="/some/path/test.npy'
+            '" id="1234"/><Query string="find record" results="1">'
+            '</Query><Error code="0" description="An error occured during the'
+            ' of this query. Maybe you use a wrong syntax?" /></Response>')
+
+    def test_ref(self):
+        self.clear_log()
+        files = get_referenced_files("test.npy", prefix=None, filename=None,
+                                     location=None)
+        self.assertEqual(files, [])
+        assert "does not allow a search" in self.get_log()
+
+
+class ReferencesEmptyTest(ReferencesBaseTest):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.entities = (
+            '<Response><Query string="find record" results="1">'
+            '</Query></Response>')
+
+    def test_ref(self):
+        self.clear_log()
+        files = get_referenced_files("test.npy", prefix=None, filename=None,
+                                     location=None)
+        self.assertEqual(files, [])
+        assert "does not reference any " in self.get_log()
 
 
 def is_dhornung(rec):