-import os
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
import glob
+import os
+import re
+import stat
+
+BACKSLASH_ESCAPE_RE = re.compile(r'\\(.)')
class SubstitutionError(Exception):
pass
return None
def sub_file(v):
- return os.path.join(os.environ['TASK_KEEPMOUNT'], v)
+ path = os.path.join(os.environ['TASK_KEEPMOUNT'], v)
+ st = os.stat(path)
+ if st and stat.S_ISREG(st.st_mode):
+ return path
+ else:
+ raise SubstitutionError("$(file {}) is not accessible or is not a regular file".format(path))
def sub_dir(v):
d = os.path.dirname(v)
if d == '':
d = v
- return os.path.join(os.environ['TASK_KEEPMOUNT'], d)
+ path = os.path.join(os.environ['TASK_KEEPMOUNT'], d)
+ st = os.stat(path)
+ if st and stat.S_ISDIR(st.st_mode):
+ return path
+ else:
+ raise SubstitutionError("$(dir {}) is not accessible or is not a directory".format(path))
def sub_basename(v):
return os.path.splitext(os.path.basename(v))[0]
def sub_glob(v):
l = glob.glob(v)
if len(l) == 0:
- raise SubstitutionError("$(glob): No match on '%s'" % v)
+ raise SubstitutionError("$(glob {}) no match found".format(v))
else:
return l[0]
while True:
m = search(c)
if m is None:
- return c
+ return BACKSLASH_ESCAPE_RE.sub(r'\1', c)
v = do_substitution(p, c[m[0]+2 : m[1]])
var = True