Merge a916114 into 2f034c1

packtivity/handlers/environment_handlers.py

@@ -4,9 +4,14 @@
 
 @environment('docker-encapsulated')
 def docker(environment,parameters,state):
+
+    jsonpars = parameters.json()
+    for p,v in parameters.leafs():
+        p.set(jsonpars, v)
+
     for i,x in enumerate(environment['par_mounts']):
         script = x.pop('jqscript')
-        x['mountcontent'] = jq.jq(script).transform(parameters, text_output = True)
+        x['mountcontent'] = jq.jq(script).transform(jsonpars, text_output = True)
 
     if environment['workdir'] is not None:
         environment['workdir'] = state.contextualize_value(environment['workdir'])

packtivity/handlers/execution_handlers.py

@@ -25,11 +25,22 @@ def sourcepath(path):
 def state_context_to_mounts(state):
     readwrites  = state.readwrite
     readonlies = state.readonly
-    mounts = ''
+
+    mounts = []
     for rw in readwrites:
-        mounts += '-v {}:{}:rw'.format(sourcepath(os.path.abspath(rw)),rw)
+        mounts.append({
+            'type': 'bind',
+            'source': sourcepath(os.path.abspath(rw)),
+            'destination': rw,
+            'readonly': False
+        })
     for ro in readonlies:
-        mounts += ' -v {}:{}:ro'.format(sourcepath(ro),ro)
+        mounts.append({
+            'type': 'bind',
+            'source': sourcepath(os.path.abspath(ro)),
+            'destination': ro,
+            'readonly': False
+        })
     return mounts
 
 def prepare_par_mounts(parmounts,state):
@@ -39,10 +50,12 @@ def prepare_par_mounts(parmounts,state):
         with open(parmountfile,'w') as f:
             f.write(x['mountcontent'])
 
-        mounts.append(' -v {}:{}'.format(
-            sourcepath(os.path.abspath(parmountfile)),
-            x['mountpath']
-        ))
+        mounts.append({
+            'type': 'bind',
+            'source': sourcepath(os.path.abspath(parmountfile)),
+            'destination': x['mountpath'],
+            'readonly': False
+        })
 
     return mounts
 
@@ -51,13 +64,26 @@ def cvmfs_from_volume_plugin(cvmfs_repos = None):
         cvmfs_repos = yaml.load(os.environ.get('PACKTIVITY_CVMFS_REPOS','null'))
     if not cvmfs_repos:
         cvmfs_repos  = ['atlas.cern.ch','atlas-condb.cern.ch','sft.cern.ch']
-    command_line = ' --security-opt label:disable'
+
+    options = '--security-opt label:disable'
+    mounts = []
     for repo in cvmfs_repos:
-        command_line += ' --volume-driver cvmfs -v {cvmfs_repo}:/cvmfs/{cvmfs_repo}'.format(cvmfs_repo = repo)
-    return command_line
+        mounts.append({
+            'type': 'volume',
+            'source': repo,
+            'destination': '/cvmfs/{}'.format(repo),
+            'readonly': False
+        })
+
+    return options, mounts
 
 def cvmfs_from_external_mount():
-    return ' -v {}:/cvmfs'.format(os.environ.get('PACKTIVITY_CVMFS_LOCATION','/cvmfs'))
+    return '', [{
+        'type': 'volume',
+        'source': os.environ.get('PACKTIVITY_CVMFS_LOCATION','/cvmfs'),
+        'destination': '/cvmfs',
+        'readonly': False
+    }]
 
 def cvmfs_mount():
     cvmfs_source = os.environ.get('PACKTIVITY_CVMFS_SOURCE','external')
@@ -69,10 +95,12 @@ def cvmfs_mount():
         raise RuntimeError('unknown CVMFS location requested')
 
 def auth_mount():
-    if 'PACKTIVITY_AUTH_LOCATION' not in os.environ:
-        return ' -v /home/recast/recast_auth:/recast_auth'
-    else:
-        return ' -v {}:/recast_auth'.format(os.environ['PACKTIVITY_AUTH_LOCATION'])
+    return [{
+        'type': 'bind',
+        'source': os.environ.get('PACKTIVITY_AUTH_LOCATION','/home/recast/recast_auth'),
+        'destination': '/recast_auth',
+        'readonly': False
+    }]
 
 def resource_mounts(state,environment,log):
     report = '''\n\
@@ -91,15 +119,16 @@ def resource_mounts(state,environment,log):
     do_auth  = ('GRIDProxy'  in environment['resources']) or ('KRB5Auth' in environment['resources'])
     log.debug('do_auth: %s do_cvmfs: %s',do_auth,do_cvmfs)
 
+    options, mounts = '', []
 
-    resource_mounts = ''
     if do_cvmfs:
-        resource_mounts+=cvmfs_mount()
-
+        cvfms_options, cvmfs_mounts = cvmfs_mount()
+        options += cvfms_options
+        mounts  += cvmfs_mounts
     if do_auth:
-        resource_mounts+=auth_mount()
+        mounts  += auth_mount()
 
-    return resource_mounts
+    return options, mounts
 
 def docker_execution_cmdline(state,environment,log,metadata,stdin,cmd_argv):
     quoted_string = ' '.join(map(pipes.quote,cmd_argv))
@@ -119,23 +148,31 @@ def docker_execution_cmdline(state,environment,log,metadata,stdin,cmd_argv):
 
     # volume mounts (resources, parameter mounts and state mounts)
     state_mounts = state_context_to_mounts(state)
-    rsrcs_mounts = resource_mounts(state,environment,log)
-    par_mounts = ' '.join(prepare_par_mounts(environment['par_mounts'], state))
+    par_mounts   = prepare_par_mounts(environment['par_mounts'], state)
+    rsrcs_opts, rsrcs_mounts = resource_mounts(state,environment,log)
+
+
+    mount_args = ''
+    for s in state_mounts + par_mounts + rsrcs_mounts:
+        mount_args += ' -v {source}:{destination}:{mode}'.format(
+            source = s['source'],
+            destination = s['destination'],
+            mode = 'ro' if s['readonly'] else 'rw'
+        )
 
-    return 'docker run --rm {stdin} {cid} {workdir} {custom} {state_mounts} {rsrcs} {par_mounts} {img}:{tag} {command}'.format(
+    return 'docker run --rm {stdin} {cid} {workdir} {custom} {mount_args} {rsrcs_opts} {img}:{tag} {command}'.format(
         stdin = '-i' if stdin else '',
         cid = cid_file,
         workdir = workdir_flag,
         custom = custom_mod,
-        state_mounts = state_mounts,
-        rsrcs = rsrcs_mounts,
-        par_mounts = par_mounts,
+        mount_args = mount_args,
+        rsrcs_opts = rsrcs_opts,
         img = image,
         tag = imagetag,
         command = quoted_string
     )
 
-def run_docker_with_script(state,environment,job,log):
+def run_docker_with_script(environment,job,log):
     script = job['script']
     interpreter = job['interpreter']
 
@@ -147,19 +184,19 @@ def run_docker_with_script(state,environment,job,log):
     indocker = interpreter
     envmod = 'source {} && '.format(environment['envscript']) if environment['envscript'] else ''
     in_docker_cmd = envmod+indocker
-    return in_docker_cmd, script
+    return ['sh', '-c', in_docker_cmd], script
 
-def run_docker_with_oneliner(state,environment,command,log):
+def run_docker_with_oneliner(environment,job,log):
     log.debug('''\n\
 --------------
 running one liner in container.
 command: {command}
 --------------
-    '''.format(command = command))
+    '''.format(command = job['command']))
 
-    envmod = 'source {} &&'.format(environment['envscript']) if environment['envscript'] else ''
-    in_docker_cmd = '{envmodifier} {command}'.format(envmodifier = envmod, command = command)
-    return in_docker_cmd, None
+    envmod = 'source {} && '.format(environment['envscript']) if environment['envscript'] else ''
+    in_docker_cmd = envmod + job['command']
+    return ['sh', '-c', in_docker_cmd], None
 
 def execute_docker(metadata,state,log,docker_run_cmd_str,stdin_content = None):
     log.debug('container execution command: \n%s',docker_run_cmd_str)
@@ -247,16 +284,13 @@ def docker_enc_handler(environment,state,job,metadata):
     with logutils.setup_logging_topic(metadata,state,'step',return_logger = True) as log:
         if 'command' in job:
             stdin = False
-            in_docker_cmd, stdin = run_docker_with_oneliner(state,environment,job['command'],log)
+            container_argv, container_stdin = run_docker_with_oneliner(environment,job,log)
         elif 'script' in job:
             stdin = True
-            in_docker_cmd, stdin = run_docker_with_script(state,environment,job,log)
+            container_argv, container_stdin = run_docker_with_script(environment,job,log)
         else:
             raise RuntimeError('do not know yet how to run this...')
 
-        container_argv  = ['sh', '-c', in_docker_cmd]
-        container_stdin = stdin
-
         cmdline = docker_execution_cmdline(
             state,environment,log,metadata,
             stdin = stdin,

packtivity/typedleafs.py

@@ -6,25 +6,32 @@
 import base64
 import importlib
 import collections
+from six import string_types
 
 class LeafModel(object):
     def __init__(self, spec):
         self.datamodel = spec or {'keyword': None, 'types': {}}
         self._types2str, self._str2types = {}, {}
-        for name,module_class in self.datamodel['types'].items():
-            m, c = module_class.split(':')
-            c = getattr(importlib.import_module(m),c)
-            self._types2str[c] = name
-            self._str2types[name] = c
+        for name,class_def in self.datamodel['types'].items():
+            if type(class_def)==type:
+                self._types2str[class_def] = name
+                self._str2types[name] = class_def
+            elif isinstance(class_def, string_types):
+                m, c = class_def.split(':')
+                c = getattr(importlib.import_module(m),c)
+                self._types2str[c] = name
+                self._str2types[name] = c
+            else:
+                raise RuntimeError('not sure how to interpret type def %s',class_def)
         self.keyword = self.datamodel['keyword']
         self.leaf_magic = '___leaf___'
 
     def leaf_encode(self,obj):
-        return self.leaf_magic + base64.b64encode(json.dumps(self.dumper(obj)))
+        return self.leaf_magic + base64.b64encode(json.dumps(self.dumper(obj)).encode('utf-8'))
 
     @staticmethod
     def leaf_decode(str):
-        return json.loads(base64.b64decode(str))
+        return json.loads(base64.b64decode(str).decode('utf-8'))
 
     def loader(self, spec, idleafs):
         if not self.keyword: return spec
@@ -95,7 +102,7 @@ def _jsonable(self, value):
 
     @classmethod
     def fromJSON(cls, data, deserialization_opts):
-        return cls(data, deserialization_opts['leafmodel'], deserialization_opts['idleafs'])
+        return cls(data, deserialization_opts.get('leafmodel',None), deserialization_opts.get('idleafs',False))
 
     def _load_from_string(self,jsonstring, typed = True, idleafs = False):
         if typed:
@@ -121,20 +128,23 @@ def copy(self):
         return TypedLeafs(copy.deepcopy(self.typed()), self.leafmodel)
 
     def asrefs(self):
-        data = copy.deepcopy(self.typed())
+        data = copy.deepcopy(self.json())
         for p, v in self.leafs():
             p.set(data, p)
         return data
 
     ### QUERY methods
     def resolve_ref(self, reference):
-        return reference.resolve(self.typed())
+        return reference.get(self.typed())
 
     def jsonpointer(self,pointer_str):
         return jsonpointer.JsonPointer(pointer_str).resolve( self.typed() )
 
-    def jsonpath(self,jsonpath_expression):
-        return jsonpath_rw.parse(jsonpath_expression).find( self.typed() )[0].value
+    def jsonpath(self,jsonpath_expression, multiple_output = False):
+        if not multiple_output:
+            return jsonpath_rw.parse(jsonpath_expression).find( self.typed() )[0].value
+        else:
+            return [x.value for x in jsonpath_rw.parse(jsonpath_expression).find( self.typed())]
 
     def jq(self,jq_program, *args, **kwargs):
         return TypedLeafs(jq.jq(jq_program).transform(self.typed(idleafs = True), *args, **kwargs), self.leafmodel, idleafs = True)

tests/conftest.py

@@ -45,6 +45,14 @@ def dockeproc_script_pack(tmpdir):
 def docker_touchfile_workdir(tmpdir):
 	return packtivity.pack_object.fromspec('tests/testspecs/environment_tests/touchfile_docker_inworkdir.yml')
 
+@pytest.fixture()
+def docker_env_resources(tmpdir):
+	return packtivity.pack_object.fromspec('tests/testspecs/environment_tests/resources_docker.yml')
+
+@pytest.fixture()
+def docker_env_parmounts(tmpdir):
+	return packtivity.pack_object.fromspec('tests/testspecs/environment_tests/resources_parmounts.yml')
+
 @pytest.fixture()
 def fromjq_pub_default(tmpdir):
 	return packtivity.pack_object.fromspec('tests/testspecs/publisher_tests/fromjq-pub-default.yml')

tests/test_environments.py

@@ -0,0 +1,15 @@
+from packtivity.handlers.publisher_handlers import handlers
+from packtivity.typedleafs import TypedLeafs
+
+from packtivity.handlers.environment_handlers import handlers
+from packtivity.syncbackends import finalize_inputs
+
+import logging
+
+def test_docker_parmounts(tmpdir,basic_localfs_state, docker_env_parmounts):
+	state = basic_localfs_state
+	environment = docker_env_parmounts.spec['environment']
+
+	parameters, state = finalize_inputs(TypedLeafs({'outputfile': '{workdir}/hello.txt'}), state)
+	env = handlers[environment['environment_type']]['default'](environment,parameters,state)
+	assert env['par_mounts'][0]['mountcontent'] == '"{}"'.format(parameters['outputfile'])

tests/test_executions.py

@@ -0,0 +1,68 @@
+from packtivity.handlers.publisher_handlers import handlers
+from packtivity.typedleafs import TypedLeafs
+
+from packtivity.handlers.execution_handlers import run_docker_with_oneliner, docker_execution_cmdline
+import logging
+
+def test_docker_cvmfs(tmpdir,basic_localfs_state, docker_env_resources, monkeypatch):
+	state = basic_localfs_state
+	environment = docker_env_resources.spec['environment']
+	log = logging.getLogger('test')
+	job = {'command': 'echo hello world'}
+	container_argv, stdin = run_docker_with_oneliner(environment,job,log)
+
+	assert container_argv == ['sh','-c','echo hello world']
+	assert stdin == None
+
+	cmdline = docker_execution_cmdline(
+		state,environment,log,{'name':'myname'},
+		stdin = stdin,
+		cmd_argv = container_argv
+	)
+	assert '-v /cvmfs:/cvmfs' in cmdline
+
+	monkeypatch.setenv('PACKTIVITY_CVMFS_LOCATION','/here/cvmfs')
+	cmdline = docker_execution_cmdline(
+		state,environment,log,{'name':'myname'},
+		stdin = stdin,
+		cmd_argv = container_argv
+	)
+	assert '-v /here/cvmfs:/cvmfs' in cmdline
+
+	monkeypatch.setenv('PACKTIVITY_CVMFS_SOURCE','voldriver')
+	cmdline = docker_execution_cmdline(
+		state,environment,log,{'name':'myname'},
+		stdin = stdin,
+		cmd_argv = container_argv
+	)
+
+	assert '-v atlas-condb.cern.ch:/cvmfs/atlas-condb.cern.ch:rw' in cmdline
+	assert '-v sft.cern.ch:/cvmfs/sft.cern.ch:rw' in cmdline
+	assert '-v atlas.cern.ch:/cvmfs/atlas.cern.ch:rw' in cmdline
+	assert '--security-opt label:disable' in cmdline
+
+
+def test_docker_auth(tmpdir,basic_localfs_state, docker_env_resources, monkeypatch):
+	state = basic_localfs_state
+	environment = docker_env_resources.spec['environment']
+	log = logging.getLogger('test')
+	job = {'command': 'echo hello world'}
+	container_argv, stdin = run_docker_with_oneliner(environment,job,log)
+
+	assert container_argv == ['sh','-c','echo hello world']
+	assert stdin == None
+
+	cmdline = docker_execution_cmdline(
+		state,environment,log,{'name':'myname'},
+		stdin = stdin,
+		cmd_argv = container_argv
+	)
+	assert '-v /home/recast/recast_auth:/recast_auth:rw' in cmdline
+
+	monkeypatch.setenv('PACKTIVITY_AUTH_LOCATION','/here')
+	cmdline = docker_execution_cmdline(
+		state,environment,log,{'name':'myname'},
+		stdin = stdin,
+		cmd_argv = container_argv
+	)
+	assert '-v /here:/recast_auth:rw' in cmdline