diff --git a/src/caosadvancedtools/table_importer.py b/src/caosadvancedtools/table_importer.py
index 7506e81527768af2ebdb65e43d48f03d90fd5cb2..5268332c352c030d2c4d68903d0ab532217073fd 100755
--- a/src/caosadvancedtools/table_importer.py
+++ b/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
diff --git a/src/caosadvancedtools/utils.py b/src/caosadvancedtools/utils.py
index 8622a7b4b87104a9f777c96f0029bd33af36778c..7f51fa4158c612a8f7d5e517ffb4f479a34fe08b 100644
--- a/src/caosadvancedtools/utils.py
+++ b/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):
diff --git a/unittests/test_table_importer.py b/unittests/test_table_importer.py
index 18a3d0175ff10ac02bbbe3ca7f805cf2bf705954..3d44b0dc5fb9a229e1b79fe114082b2fc33a8c23 100644
--- a/unittests/test_table_importer.py
+++ b/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")
diff --git a/unittests/test_utils.py b/unittests/test_utils.py
index df1e491cf0f3fefeb5233dc13212534177d630c9..b259ad4ad9d2209a0be113e3dd386490dd78db9c 100644
--- a/unittests/test_utils.py
+++ b/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):