sdk/cwl/tests/test_container.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: Apache-2.0
   4
   5 from builtins import str
   6 from builtins import object
   7
   8 import arvados_cwl
   9 import arvados_cwl.context
  10 import arvados_cwl.util
  11 from arvados_cwl.arvdocker import arv_docker_clear_cache
  12 import copy
  13 import arvados.config
  14 import logging
  15 import mock
  16 import unittest
  17 import os
  18 import functools
  19 import cwltool.process
  20 import cwltool.secrets
  21 from cwltool.update import INTERNAL_VERSION
  22 from schema_salad.ref_resolver import Loader
  23 from schema_salad.sourceline import cmap
  24
  25 from .matcher import JsonDiffMatcher, StripYAMLComments
  26 from .mock_discovery import get_rootDesc
  27
  28 if not os.getenv('ARVADOS_DEBUG'):
  29     logging.getLogger('arvados.cwl-runner').setLevel(logging.WARN)
  30     logging.getLogger('arvados.arv-run').setLevel(logging.WARN)
  31
  32 class CollectionMock(object):
  33     def __init__(self, vwdmock, *args, **kwargs):
  34         self.vwdmock = vwdmock
  35         self.count = 0
  36
  37     def open(self, *args, **kwargs):
  38         self.count += 1
  39         return self.vwdmock.open(*args, **kwargs)
  40
  41     def copy(self, *args, **kwargs):
  42         self.count += 1
  43         self.vwdmock.copy(*args, **kwargs)
  44
  45     def save_new(self, *args, **kwargs):
  46         pass
  47
  48     def __len__(self):
  49         return self.count
  50
  51     def portable_data_hash(self):
  52         if self.count == 0:
  53             return arvados.config.EMPTY_BLOCK_LOCATOR
  54         else:
  55             return "99999999999999999999999999999996+99"
  56
  57
  58 class TestContainer(unittest.TestCase):
  59
  60     def setUp(self):
  61         cwltool.process._names = set()
  62         arv_docker_clear_cache()
  63
  64     def helper(self, runner, enable_reuse=True):
  65         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema(INTERNAL_VERSION)
  66
  67         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess,
  68                                          collection_cache=arvados_cwl.CollectionCache(runner.api, None, 0))
  69         loadingContext = arvados_cwl.context.ArvLoadingContext(
  70             {"avsc_names": avsc_names,
  71              "basedir": "",
  72              "make_fs_access": make_fs_access,
  73              "loader": Loader({}),
  74              "metadata": {"cwlVersion": INTERNAL_VERSION, "http://commonwl.org/cwltool#original_cwlVersion": "v1.0"}})
  75         runtimeContext = arvados_cwl.context.ArvRuntimeContext(
  76             {"work_api": "containers",
  77              "basedir": "",
  78              "name": "test_run_"+str(enable_reuse),
  79              "make_fs_access": make_fs_access,
  80              "tmpdir": "/tmp",
  81              "enable_reuse": enable_reuse,
  82              "priority": 500,
  83              "project_uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
  84             })
  85
  86         return loadingContext, runtimeContext
  87
  88     # Helper function to set up the ArvCwlExecutor to use the containers api
  89     # and test that the RuntimeStatusLoggingHandler is set up correctly
  90     def setup_and_test_container_executor_and_logging(self, gcc_mock) :
  91         api = mock.MagicMock()
  92         api._rootDesc = copy.deepcopy(get_rootDesc())
  93
  94         # Make sure ArvCwlExecutor thinks it's running inside a container so it
  95         # adds the logging handler that will call runtime_status_update() mock
  96         self.assertFalse(gcc_mock.called)
  97         runner = arvados_cwl.ArvCwlExecutor(api)
  98         self.assertEqual(runner.work_api, 'containers')
  99         root_logger = logging.getLogger('')
 100         handlerClasses = [h.__class__ for h in root_logger.handlers]
 101         self.assertTrue(arvados_cwl.RuntimeStatusLoggingHandler in handlerClasses)
 102         return runner
 103
 104     # The test passes no builder.resources
 105     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 106     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 107     def test_run(self, keepdocker):
 108         for enable_reuse in (True, False):
 109             arv_docker_clear_cache()
 110
 111             runner = mock.MagicMock()
 112             runner.ignore_docker_for_reuse = False
 113             runner.intermediate_output_ttl = 0
 114             runner.secret_store = cwltool.secrets.SecretStore()
 115
 116             keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 117             runner.api.collections().get().execute.return_value = {
 118                 "portable_data_hash": "99999999999999999999999999999993+99"}
 119
 120             tool = cmap({
 121                 "inputs": [],
 122                 "outputs": [],
 123                 "baseCommand": "ls",
 124                 "arguments": [{"valueFrom": "$(runtime.outdir)"}],
 125                 "id": "#",
 126                 "class": "CommandLineTool"
 127             })
 128
 129             loadingContext, runtimeContext = self.helper(runner, enable_reuse)
 130
 131             arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 132             arvtool.formatgraph = None
 133
 134             for j in arvtool.job({}, mock.MagicMock(), runtimeContext):
 135                 j.run(runtimeContext)
 136                 runner.api.container_requests().create.assert_called_with(
 137                     body=JsonDiffMatcher({
 138                         'environment': {
 139                             'HOME': '/var/spool/cwl',
 140                             'TMPDIR': '/tmp'
 141                         },
 142                         'name': 'test_run_'+str(enable_reuse),
 143                         'runtime_constraints': {
 144                             'vcpus': 1,
 145                             'ram': 1073741824
 146                         },
 147                         'use_existing': enable_reuse,
 148                         'priority': 500,
 149                         'mounts': {
 150                             '/tmp': {'kind': 'tmp',
 151                                      "capacity": 1073741824
 152                                  },
 153                             '/var/spool/cwl': {'kind': 'tmp',
 154                                                "capacity": 1073741824 }
 155                         },
 156                         'state': 'Committed',
 157                         'output_name': 'Output for step test_run_'+str(enable_reuse),
 158                         'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 159                         'output_path': '/var/spool/cwl',
 160                         'output_ttl': 0,
 161                         'container_image': '99999999999999999999999999999993+99',
 162                         'command': ['ls', '/var/spool/cwl'],
 163                         'cwd': '/var/spool/cwl',
 164                         'scheduling_parameters': {},
 165                         'properties': {},
 166                         'secret_mounts': {}
 167                     }))
 168
 169     # The test passes some fields in builder.resources
 170     # For the remaining fields, the defaults will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 171     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 172     def test_resource_requirements(self, keepdocker):
 173         runner = mock.MagicMock()
 174         runner.ignore_docker_for_reuse = False
 175         runner.intermediate_output_ttl = 3600
 176         runner.secret_store = cwltool.secrets.SecretStore()
 177
 178         keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 179         runner.api.collections().get().execute.return_value = {
 180             "portable_data_hash": "99999999999999999999999999999993+99"}
 181
 182         tool = cmap({
 183             "inputs": [],
 184             "outputs": [],
 185             "hints": [{
 186                 "class": "ResourceRequirement",
 187                 "coresMin": 3,
 188                 "ramMin": 3000,
 189                 "tmpdirMin": 4000,
 190                 "outdirMin": 5000
 191             }, {
 192                 "class": "http://arvados.org/cwl#RuntimeConstraints",
 193                 "keep_cache": 512
 194             }, {
 195                 "class": "http://arvados.org/cwl#APIRequirement",
 196             }, {
 197                 "class": "http://arvados.org/cwl#PartitionRequirement",
 198                 "partition": "blurb"
 199             }, {
 200                 "class": "http://arvados.org/cwl#IntermediateOutput",
 201                 "outputTTL": 7200
 202             }, {
 203                 "class": "http://arvados.org/cwl#ReuseRequirement",
 204                 "enableReuse": False
 205             }],
 206             "baseCommand": "ls",
 207             "id": "#",
 208             "class": "CommandLineTool"
 209         })
 210
 211         loadingContext, runtimeContext = self.helper(runner)
 212         runtimeContext.name = "test_resource_requirements"
 213
 214         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 215         arvtool.formatgraph = None
 216         for j in arvtool.job({}, mock.MagicMock(), runtimeContext):
 217             j.run(runtimeContext)
 218
 219         call_args, call_kwargs = runner.api.container_requests().create.call_args
 220
 221         call_body_expected = {
 222             'environment': {
 223                 'HOME': '/var/spool/cwl',
 224                 'TMPDIR': '/tmp'
 225             },
 226             'name': 'test_resource_requirements',
 227             'runtime_constraints': {
 228                 'vcpus': 3,
 229                 'ram': 3145728000,
 230                 'keep_cache_ram': 536870912,
 231                 'API': True
 232             },
 233             'use_existing': False,
 234             'priority': 500,
 235             'mounts': {
 236                 '/tmp': {'kind': 'tmp',
 237                          "capacity": 4194304000 },
 238                 '/var/spool/cwl': {'kind': 'tmp',
 239                                    "capacity": 5242880000 }
 240             },
 241             'state': 'Committed',
 242             'output_name': 'Output for step test_resource_requirements',
 243             'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 244             'output_path': '/var/spool/cwl',
 245             'output_ttl': 7200,
 246             'container_image': '99999999999999999999999999999993+99',
 247             'command': ['ls'],
 248             'cwd': '/var/spool/cwl',
 249             'scheduling_parameters': {
 250                 'partitions': ['blurb']
 251             },
 252             'properties': {},
 253             'secret_mounts': {}
 254         }
 255
 256         call_body = call_kwargs.get('body', None)
 257         self.assertNotEqual(None, call_body)
 258         for key in call_body:
 259             self.assertEqual(call_body_expected.get(key), call_body.get(key))
 260
 261
 262     # The test passes some fields in builder.resources
 263     # For the remaining fields, the defaults will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 264     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 265     @mock.patch("arvados.collection.Collection")
 266     def test_initial_work_dir(self, collection_mock, keepdocker):
 267         runner = mock.MagicMock()
 268         runner.ignore_docker_for_reuse = False
 269         runner.intermediate_output_ttl = 0
 270         runner.secret_store = cwltool.secrets.SecretStore()
 271
 272         keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 273         runner.api.collections().get().execute.return_value = {
 274             "portable_data_hash": "99999999999999999999999999999993+99"}
 275
 276         sourcemock = mock.MagicMock()
 277         def get_collection_mock(p):
 278             if "/" in p:
 279                 return (sourcemock, p.split("/", 1)[1])
 280             else:
 281                 return (sourcemock, "")
 282         runner.fs_access.get_collection.side_effect = get_collection_mock
 283
 284         vwdmock = mock.MagicMock()
 285         collection_mock.side_effect = lambda *args, **kwargs: CollectionMock(vwdmock, *args, **kwargs)
 286
 287         tool = cmap({
 288             "inputs": [],
 289             "outputs": [],
 290             "hints": [{
 291                 "class": "InitialWorkDirRequirement",
 292                 "listing": [{
 293                     "class": "File",
 294                     "basename": "foo",
 295                     "location": "keep:99999999999999999999999999999995+99/bar"
 296                 },
 297                 {
 298                     "class": "Directory",
 299                     "basename": "foo2",
 300                     "location": "keep:99999999999999999999999999999995+99"
 301                 },
 302                 {
 303                     "class": "File",
 304                     "basename": "filename",
 305                     "location": "keep:99999999999999999999999999999995+99/baz/filename"
 306                 },
 307                 {
 308                     "class": "Directory",
 309                     "basename": "subdir",
 310                     "location": "keep:99999999999999999999999999999995+99/subdir"
 311                 }                        ]
 312             }],
 313             "baseCommand": "ls",
 314             "id": "#",
 315             "class": "CommandLineTool"
 316         })
 317
 318         loadingContext, runtimeContext = self.helper(runner)
 319         runtimeContext.name = "test_initial_work_dir"
 320
 321         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 322         arvtool.formatgraph = None
 323         for j in arvtool.job({}, mock.MagicMock(), runtimeContext):
 324             j.run(runtimeContext)
 325
 326         call_args, call_kwargs = runner.api.container_requests().create.call_args
 327
 328         vwdmock.copy.assert_has_calls([mock.call('bar', 'foo', source_collection=sourcemock)])
 329         vwdmock.copy.assert_has_calls([mock.call('.', 'foo2', source_collection=sourcemock)])
 330         vwdmock.copy.assert_has_calls([mock.call('baz/filename', 'filename', source_collection=sourcemock)])
 331         vwdmock.copy.assert_has_calls([mock.call('subdir', 'subdir', source_collection=sourcemock)])
 332
 333         call_body_expected = {
 334             'environment': {
 335                 'HOME': '/var/spool/cwl',
 336                 'TMPDIR': '/tmp'
 337             },
 338             'name': 'test_initial_work_dir',
 339             'runtime_constraints': {
 340                 'vcpus': 1,
 341                 'ram': 1073741824
 342             },
 343             'use_existing': True,
 344             'priority': 500,
 345             'mounts': {
 346                 '/tmp': {'kind': 'tmp',
 347                          "capacity": 1073741824 },
 348                 '/var/spool/cwl': {'kind': 'tmp',
 349                                    "capacity": 1073741824 },
 350                 '/var/spool/cwl/foo': {
 351                     'kind': 'collection',
 352                     'path': 'foo',
 353                     'portable_data_hash': '99999999999999999999999999999996+99'
 354                 },
 355                 '/var/spool/cwl/foo2': {
 356                     'kind': 'collection',
 357                     'path': 'foo2',
 358                     'portable_data_hash': '99999999999999999999999999999996+99'
 359                 },
 360                 '/var/spool/cwl/filename': {
 361                     'kind': 'collection',
 362                     'path': 'filename',
 363                     'portable_data_hash': '99999999999999999999999999999996+99'
 364                 },
 365                 '/var/spool/cwl/subdir': {
 366                     'kind': 'collection',
 367                     'path': 'subdir',
 368                     'portable_data_hash': '99999999999999999999999999999996+99'
 369                 }
 370             },
 371             'state': 'Committed',
 372             'output_name': 'Output for step test_initial_work_dir',
 373             'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 374             'output_path': '/var/spool/cwl',
 375             'output_ttl': 0,
 376             'container_image': '99999999999999999999999999999993+99',
 377             'command': ['ls'],
 378             'cwd': '/var/spool/cwl',
 379             'scheduling_parameters': {
 380             },
 381             'properties': {},
 382             'secret_mounts': {}
 383         }
 384
 385         call_body = call_kwargs.get('body', None)
 386         self.assertNotEqual(None, call_body)
 387         for key in call_body:
 388             self.assertEqual(call_body_expected.get(key), call_body.get(key))
 389
 390
 391     # Test redirecting stdin/stdout/stderr
 392     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 393     def test_redirects(self, keepdocker):
 394         runner = mock.MagicMock()
 395         runner.ignore_docker_for_reuse = False
 396         runner.intermediate_output_ttl = 0
 397         runner.secret_store = cwltool.secrets.SecretStore()
 398
 399         keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 400         runner.api.collections().get().execute.return_value = {
 401             "portable_data_hash": "99999999999999999999999999999993+99"}
 402
 403         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema(INTERNAL_VERSION)
 404
 405         tool = cmap({
 406             "inputs": [],
 407             "outputs": [],
 408             "baseCommand": "ls",
 409             "stdout": "stdout.txt",
 410             "stderr": "stderr.txt",
 411             "stdin": "/keep/99999999999999999999999999999996+99/file.txt",
 412             "arguments": [{"valueFrom": "$(runtime.outdir)"}],
 413             "id": "#",
 414             "class": "CommandLineTool"
 415         })
 416
 417         loadingContext, runtimeContext = self.helper(runner)
 418         runtimeContext.name = "test_run_redirect"
 419
 420         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 421         arvtool.formatgraph = None
 422         for j in arvtool.job({}, mock.MagicMock(), runtimeContext):
 423             j.run(runtimeContext)
 424             runner.api.container_requests().create.assert_called_with(
 425                 body=JsonDiffMatcher({
 426                     'environment': {
 427                         'HOME': '/var/spool/cwl',
 428                         'TMPDIR': '/tmp'
 429                     },
 430                     'name': 'test_run_redirect',
 431                     'runtime_constraints': {
 432                         'vcpus': 1,
 433                         'ram': 1073741824
 434                     },
 435                     'use_existing': True,
 436                     'priority': 500,
 437                     'mounts': {
 438                         '/tmp': {'kind': 'tmp',
 439                                  "capacity": 1073741824 },
 440                         '/var/spool/cwl': {'kind': 'tmp',
 441                                            "capacity": 1073741824 },
 442                         "stderr": {
 443                             "kind": "file",
 444                             "path": "/var/spool/cwl/stderr.txt"
 445                         },
 446                         "stdin": {
 447                             "kind": "collection",
 448                             "path": "file.txt",
 449                             "portable_data_hash": "99999999999999999999999999999996+99"
 450                         },
 451                         "stdout": {
 452                             "kind": "file",
 453                             "path": "/var/spool/cwl/stdout.txt"
 454                         },
 455                     },
 456                     'state': 'Committed',
 457                     "output_name": "Output for step test_run_redirect",
 458                     'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 459                     'output_path': '/var/spool/cwl',
 460                     'output_ttl': 0,
 461                     'container_image': '99999999999999999999999999999993+99',
 462                     'command': ['ls', '/var/spool/cwl'],
 463                     'cwd': '/var/spool/cwl',
 464                     'scheduling_parameters': {},
 465                     'properties': {},
 466                     'secret_mounts': {}
 467                 }))
 468
 469     @mock.patch("arvados.collection.Collection")
 470     def test_done(self, col):
 471         api = mock.MagicMock()
 472
 473         runner = mock.MagicMock()
 474         runner.api = api
 475         runner.num_retries = 0
 476         runner.ignore_docker_for_reuse = False
 477         runner.intermediate_output_ttl = 0
 478         runner.secret_store = cwltool.secrets.SecretStore()
 479
 480         runner.api.containers().get().execute.return_value = {"state":"Complete",
 481                                                               "output": "abc+123",
 482                                                               "exit_code": 0}
 483
 484         col().open.return_value = []
 485
 486         loadingContext, runtimeContext = self.helper(runner)
 487
 488         arvjob = arvados_cwl.ArvadosContainer(runner,
 489                                               runtimeContext,
 490                                               mock.MagicMock(),
 491                                               {},
 492                                               None,
 493                                               [],
 494                                               [],
 495                                               "testjob")
 496         arvjob.output_callback = mock.MagicMock()
 497         arvjob.collect_outputs = mock.MagicMock()
 498         arvjob.successCodes = [0]
 499         arvjob.outdir = "/var/spool/cwl"
 500         arvjob.output_ttl = 3600
 501
 502         arvjob.collect_outputs.return_value = {"out": "stuff"}
 503
 504         arvjob.done({
 505             "state": "Final",
 506             "log_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz1",
 507             "output_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2",
 508             "uuid": "zzzzz-xvhdp-zzzzzzzzzzzzzzz",
 509             "container_uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz",
 510             "modified_at": "2017-05-26T12:01:22Z"
 511         })
 512
 513         self.assertFalse(api.collections().create.called)
 514         self.assertFalse(runner.runtime_status_error.called)
 515
 516         arvjob.collect_outputs.assert_called_with("keep:abc+123", 0)
 517         arvjob.output_callback.assert_called_with({"out": "stuff"}, "success")
 518         runner.add_intermediate_output.assert_called_with("zzzzz-4zz18-zzzzzzzzzzzzzz2")
 519
 520     # Test to make sure we dont call runtime_status_update if we already did
 521     # some where higher up in the call stack
 522     @mock.patch("arvados_cwl.util.get_current_container")
 523     def test_recursive_runtime_status_update(self, gcc_mock):
 524         self.setup_and_test_container_executor_and_logging(gcc_mock)
 525         root_logger = logging.getLogger('')
 526
 527         # get_current_container is invoked when we call runtime_status_update
 528         # so try and log again!
 529         gcc_mock.side_effect = lambda *args: root_logger.error("Second Error")
 530         try:
 531             root_logger.error("First Error")
 532         except RuntimeError:
 533             self.fail("RuntimeStatusLoggingHandler should not be called recursively")
 534
 535
 536     # Test to make sure that an exception raised from
 537     # get_current_container doesn't cause the logger to raise an
 538     # exception
 539     @mock.patch("arvados_cwl.util.get_current_container")
 540     def test_runtime_status_get_current_container_exception(self, gcc_mock):
 541         self.setup_and_test_container_executor_and_logging(gcc_mock)
 542         root_logger = logging.getLogger('')
 543
 544         # get_current_container is invoked when we call
 545         # runtime_status_update, it is going to also raise an
 546         # exception.
 547         gcc_mock.side_effect = Exception("Second Error")
 548         try:
 549             root_logger.error("First Error")
 550         except Exception:
 551             self.fail("Exception in logger should not propagate")
 552         self.assertTrue(gcc_mock.called)
 553
 554     @mock.patch("arvados_cwl.ArvCwlExecutor.runtime_status_update")
 555     @mock.patch("arvados_cwl.util.get_current_container")
 556     @mock.patch("arvados.collection.CollectionReader")
 557     @mock.patch("arvados.collection.Collection")
 558     def test_child_failure(self, col, reader, gcc_mock, rts_mock):
 559         runner = self.setup_and_test_container_executor_and_logging(gcc_mock)
 560
 561         gcc_mock.return_value = {"uuid" : "zzzzz-dz642-zzzzzzzzzzzzzzz"}
 562         self.assertTrue(gcc_mock.called)
 563
 564         runner.num_retries = 0
 565         runner.ignore_docker_for_reuse = False
 566         runner.intermediate_output_ttl = 0
 567         runner.secret_store = cwltool.secrets.SecretStore()
 568         runner.label = mock.MagicMock()
 569         runner.label.return_value = '[container testjob]'
 570
 571         runner.api.containers().get().execute.return_value = {
 572             "state":"Complete",
 573             "output": "abc+123",
 574             "exit_code": 1,
 575             "log": "def+234"
 576         }
 577
 578         col().open.return_value = []
 579
 580         loadingContext, runtimeContext = self.helper(runner)
 581
 582         arvjob = arvados_cwl.ArvadosContainer(runner,
 583                                               runtimeContext,
 584                                               mock.MagicMock(),
 585                                               {},
 586                                               None,
 587                                               [],
 588                                               [],
 589                                               "testjob")
 590         arvjob.output_callback = mock.MagicMock()
 591         arvjob.collect_outputs = mock.MagicMock()
 592         arvjob.successCodes = [0]
 593         arvjob.outdir = "/var/spool/cwl"
 594         arvjob.output_ttl = 3600
 595         arvjob.collect_outputs.return_value = {"out": "stuff"}
 596
 597         arvjob.done({
 598             "state": "Final",
 599             "log_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz1",
 600             "output_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2",
 601             "uuid": "zzzzz-xvhdp-zzzzzzzzzzzzzzz",
 602             "container_uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz",
 603             "modified_at": "2017-05-26T12:01:22Z"
 604         })
 605
 606         rts_mock.assert_called_with(
 607             'error',
 608             'arvados.cwl-runner: [container testjob] (zzzzz-xvhdp-zzzzzzzzzzzzzzz) error log:',
 609             '  ** log is empty **'
 610         )
 611         arvjob.output_callback.assert_called_with({"out": "stuff"}, "permanentFail")
 612
 613     # The test passes no builder.resources
 614     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 615     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 616     def test_mounts(self, keepdocker):
 617         runner = mock.MagicMock()
 618         runner.ignore_docker_for_reuse = False
 619         runner.intermediate_output_ttl = 0
 620         runner.secret_store = cwltool.secrets.SecretStore()
 621
 622         keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 623         runner.api.collections().get().execute.return_value = {
 624             "portable_data_hash": "99999999999999999999999999999994+99",
 625             "manifest_text": ". 99999999999999999999999999999994+99 0:0:file1 0:0:file2"}
 626
 627         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("v1.1")
 628
 629         tool = cmap({
 630             "inputs": [
 631                 {"id": "p1",
 632                  "type": "Directory"}
 633             ],
 634             "outputs": [],
 635             "baseCommand": "ls",
 636             "arguments": [{"valueFrom": "$(runtime.outdir)"}],
 637             "id": "#",
 638             "class": "CommandLineTool"
 639         })
 640
 641         loadingContext, runtimeContext = self.helper(runner)
 642         runtimeContext.name = "test_run_mounts"
 643
 644         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 645         arvtool.formatgraph = None
 646         job_order = {
 647             "p1": {
 648                 "class": "Directory",
 649                 "location": "keep:99999999999999999999999999999994+44",
 650                 "http://arvados.org/cwl#collectionUUID": "zzzzz-4zz18-zzzzzzzzzzzzzzz",
 651                 "listing": [
 652                     {
 653                         "class": "File",
 654                         "location": "keep:99999999999999999999999999999994+44/file1",
 655                     },
 656                     {
 657                         "class": "File",
 658                         "location": "keep:99999999999999999999999999999994+44/file2",
 659                     }
 660                 ]
 661             }
 662         }
 663         for j in arvtool.job(job_order, mock.MagicMock(), runtimeContext):
 664             j.run(runtimeContext)
 665             runner.api.container_requests().create.assert_called_with(
 666                 body=JsonDiffMatcher({
 667                     'environment': {
 668                         'HOME': '/var/spool/cwl',
 669                         'TMPDIR': '/tmp'
 670                     },
 671                     'name': 'test_run_mounts',
 672                     'runtime_constraints': {
 673                         'vcpus': 1,
 674                         'ram': 1073741824
 675                     },
 676                     'use_existing': True,
 677                     'priority': 500,
 678                     'mounts': {
 679                         "/keep/99999999999999999999999999999994+44": {
 680                             "kind": "collection",
 681                             "portable_data_hash": "99999999999999999999999999999994+44",
 682                             "uuid": "zzzzz-4zz18-zzzzzzzzzzzzzzz"
 683                         },
 684                         '/tmp': {'kind': 'tmp',
 685                                  "capacity": 1073741824 },
 686                         '/var/spool/cwl': {'kind': 'tmp',
 687                                            "capacity": 1073741824 }
 688                     },
 689                     'state': 'Committed',
 690                     'output_name': 'Output for step test_run_mounts',
 691                     'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 692                     'output_path': '/var/spool/cwl',
 693                     'output_ttl': 0,
 694                     'container_image': '99999999999999999999999999999994+99',
 695                     'command': ['ls', '/var/spool/cwl'],
 696                     'cwd': '/var/spool/cwl',
 697                     'scheduling_parameters': {},
 698                     'properties': {},
 699                     'secret_mounts': {}
 700                 }))
 701
 702     # The test passes no builder.resources
 703     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 704     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 705     def test_secrets(self, keepdocker):
 706         runner = mock.MagicMock()
 707         runner.ignore_docker_for_reuse = False
 708         runner.intermediate_output_ttl = 0
 709         runner.secret_store = cwltool.secrets.SecretStore()
 710
 711         keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 712         runner.api.collections().get().execute.return_value = {
 713             "portable_data_hash": "99999999999999999999999999999993+99"}
 714
 715         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("v1.1")
 716
 717         tool = cmap({"arguments": ["md5sum", "example.conf"],
 718                      "class": "CommandLineTool",
 719                      "hints": [
 720                          {
 721                              "class": "http://commonwl.org/cwltool#Secrets",
 722                              "secrets": [
 723                                  "#secret_job.cwl/pw"
 724                              ]
 725                          }
 726                      ],
 727                      "id": "#secret_job.cwl",
 728                      "inputs": [
 729                          {
 730                              "id": "#secret_job.cwl/pw",
 731                              "type": "string"
 732                          }
 733                      ],
 734                      "outputs": [
 735                      ],
 736                      "requirements": [
 737                          {
 738                              "class": "InitialWorkDirRequirement",
 739                              "listing": [
 740                                  {
 741                                      "entry": "username: user\npassword: $(inputs.pw)\n",
 742                                      "entryname": "example.conf"
 743                                  }
 744                              ]
 745                          }
 746                      ]})
 747
 748         loadingContext, runtimeContext = self.helper(runner)
 749         runtimeContext.name = "test_secrets"
 750
 751         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 752         arvtool.formatgraph = None
 753
 754         job_order = {"pw": "blorp"}
 755         runner.secret_store.store(["pw"], job_order)
 756
 757         for j in arvtool.job(job_order, mock.MagicMock(), runtimeContext):
 758             j.run(runtimeContext)
 759             runner.api.container_requests().create.assert_called_with(
 760                 body=JsonDiffMatcher({
 761                     'environment': {
 762                         'HOME': '/var/spool/cwl',
 763                         'TMPDIR': '/tmp'
 764                     },
 765                     'name': 'test_secrets',
 766                     'runtime_constraints': {
 767                         'vcpus': 1,
 768                         'ram': 1073741824
 769                     },
 770                     'use_existing': True,
 771                     'priority': 500,
 772                     'mounts': {
 773                         '/tmp': {'kind': 'tmp',
 774                                  "capacity": 1073741824
 775                              },
 776                         '/var/spool/cwl': {'kind': 'tmp',
 777                                            "capacity": 1073741824 }
 778                     },
 779                     'state': 'Committed',
 780                     'output_name': 'Output for step test_secrets',
 781                     'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 782                     'output_path': '/var/spool/cwl',
 783                     'output_ttl': 0,
 784                     'container_image': '99999999999999999999999999999993+99',
 785                     'command': ['md5sum', 'example.conf'],
 786                     'cwd': '/var/spool/cwl',
 787                     'scheduling_parameters': {},
 788                     'properties': {},
 789                     "secret_mounts": {
 790                         "/var/spool/cwl/example.conf": {
 791                             "content": "username: user\npassword: blorp\n",
 792                             "kind": "text"
 793                         }
 794                     }
 795                 }))
 796
 797     # The test passes no builder.resources
 798     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 799     @mock.patch("arvados.commands.keepdocker.list_images_in_arv")
 800     def test_timelimit(self, keepdocker):
 801         runner = mock.MagicMock()
 802         runner.ignore_docker_for_reuse = False
 803         runner.intermediate_output_ttl = 0
 804         runner.secret_store = cwltool.secrets.SecretStore()
 805
 806         keepdocker.return_value = [("zzzzz-4zz18-zzzzzzzzzzzzzz3", "")]
 807         runner.api.collections().get().execute.return_value = {
 808             "portable_data_hash": "99999999999999999999999999999993+99"}
 809
 810         tool = cmap({
 811             "inputs": [],
 812             "outputs": [],
 813             "baseCommand": "ls",
 814             "arguments": [{"valueFrom": "$(runtime.outdir)"}],
 815             "id": "#",
 816             "class": "CommandLineTool",
 817             "hints": [
 818                 {
 819                     "class": "ToolTimeLimit",
 820                     "timelimit": 42
 821                 }
 822             ]
 823         })
 824
 825         loadingContext, runtimeContext = self.helper(runner)
 826         runtimeContext.name = "test_timelimit"
 827
 828         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, loadingContext)
 829         arvtool.formatgraph = None
 830
 831         for j in arvtool.job({}, mock.MagicMock(), runtimeContext):
 832             j.run(runtimeContext)
 833
 834         _, kwargs = runner.api.container_requests().create.call_args
 835         self.assertEqual(42, kwargs['body']['scheduling_parameters'].get('max_run_time'))
 836
 837
 838 class TestWorkflow(unittest.TestCase):
 839     def setUp(self):
 840         cwltool.process._names = set()
 841         arv_docker_clear_cache()
 842
 843     def helper(self, runner, enable_reuse=True):
 844         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("v1.0")
 845
 846         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess,
 847                                          collection_cache=arvados_cwl.CollectionCache(runner.api, None, 0))
 848
 849         document_loader.fetcher_constructor = functools.partial(arvados_cwl.CollectionFetcher, api_client=runner.api, fs_access=make_fs_access(""))
 850         document_loader.fetcher = document_loader.fetcher_constructor(document_loader.cache, document_loader.session)
 851         document_loader.fetch_text = document_loader.fetcher.fetch_text
 852         document_loader.check_exists = document_loader.fetcher.check_exists
 853
 854         loadingContext = arvados_cwl.context.ArvLoadingContext(
 855             {"avsc_names": avsc_names,
 856              "basedir": "",
 857              "make_fs_access": make_fs_access,
 858              "loader": document_loader,
 859              "metadata": {"cwlVersion": INTERNAL_VERSION, "http://commonwl.org/cwltool#original_cwlVersion": "v1.0"},
 860              "construct_tool_object": runner.arv_make_tool})
 861         runtimeContext = arvados_cwl.context.ArvRuntimeContext(
 862             {"work_api": "containers",
 863              "basedir": "",
 864              "name": "test_run_wf_"+str(enable_reuse),
 865              "make_fs_access": make_fs_access,
 866              "tmpdir": "/tmp",
 867              "enable_reuse": enable_reuse,
 868              "priority": 500})
 869
 870         return loadingContext, runtimeContext
 871
 872     # The test passes no builder.resources
 873     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 874     @mock.patch("arvados.collection.CollectionReader")
 875     @mock.patch("arvados.collection.Collection")
 876     @mock.patch('arvados.commands.keepdocker.list_images_in_arv')
 877     def test_run(self, list_images_in_arv, mockcollection, mockcollectionreader):
 878         arvados_cwl.add_arv_hints()
 879
 880         api = mock.MagicMock()
 881         api._rootDesc = get_rootDesc()
 882
 883         runner = arvados_cwl.executor.ArvCwlExecutor(api)
 884         self.assertEqual(runner.work_api, 'containers')
 885
 886         list_images_in_arv.return_value = [["zzzzz-4zz18-zzzzzzzzzzzzzzz"]]
 887         runner.api.collections().get().execute.return_value = {"portable_data_hash": "99999999999999999999999999999993+99"}
 888         runner.api.collections().list().execute.return_value = {"items": [{"uuid": "zzzzz-4zz18-zzzzzzzzzzzzzzz",
 889                                                                            "portable_data_hash": "99999999999999999999999999999993+99"}]}
 890
 891         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 892         runner.ignore_docker_for_reuse = False
 893         runner.num_retries = 0
 894         runner.secret_store = cwltool.secrets.SecretStore()
 895
 896         loadingContext, runtimeContext = self.helper(runner)
 897         runner.fs_access = runtimeContext.make_fs_access(runtimeContext.basedir)
 898
 899         mockcollectionreader().exists.return_value = True
 900
 901         tool, metadata = loadingContext.loader.resolve_ref("tests/wf/scatter2.cwl")
 902         metadata["cwlVersion"] = tool["cwlVersion"]
 903
 904         mockc = mock.MagicMock()
 905         mockcollection.side_effect = lambda *args, **kwargs: CollectionMock(mockc, *args, **kwargs)
 906         mockcollectionreader().find.return_value = arvados.arvfile.ArvadosFile(mock.MagicMock(), "token.txt")
 907
 908         arvtool = arvados_cwl.ArvadosWorkflow(runner, tool, loadingContext)
 909         arvtool.formatgraph = None
 910         it = arvtool.job({}, mock.MagicMock(), runtimeContext)
 911
 912         next(it).run(runtimeContext)
 913         next(it).run(runtimeContext)
 914
 915         with open("tests/wf/scatter2_subwf.cwl") as f:
 916             subwf = StripYAMLComments(f.read()).rstrip()
 917
 918         runner.api.container_requests().create.assert_called_with(
 919             body=JsonDiffMatcher({
 920                 "command": [
 921                     "cwltool",
 922                     "--no-container",
 923                     "--move-outputs",
 924                     "--preserve-entire-environment",
 925                     "workflow.cwl",
 926                     "cwl.input.yml"
 927                 ],
 928                 "container_image": "99999999999999999999999999999993+99",
 929                 "cwd": "/var/spool/cwl",
 930                 "environment": {
 931                     "HOME": "/var/spool/cwl",
 932                     "TMPDIR": "/tmp"
 933                 },
 934                 "mounts": {
 935                     "/keep/99999999999999999999999999999999+118": {
 936                         "kind": "collection",
 937                         "portable_data_hash": "99999999999999999999999999999999+118"
 938                     },
 939                     "/tmp": {
 940                         "capacity": 1073741824,
 941                         "kind": "tmp"
 942                     },
 943                     "/var/spool/cwl": {
 944                         "capacity": 1073741824,
 945                         "kind": "tmp"
 946                     },
 947                     "/var/spool/cwl/cwl.input.yml": {
 948                         "kind": "collection",
 949                         "path": "cwl.input.yml",
 950                         "portable_data_hash": "99999999999999999999999999999996+99"
 951                     },
 952                     "/var/spool/cwl/workflow.cwl": {
 953                         "kind": "collection",
 954                         "path": "workflow.cwl",
 955                         "portable_data_hash": "99999999999999999999999999999996+99"
 956                     },
 957                     "stdout": {
 958                         "kind": "file",
 959                         "path": "/var/spool/cwl/cwl.output.json"
 960                     }
 961                 },
 962                 "name": "scatterstep",
 963                 "output_name": "Output for step scatterstep",
 964                 "output_path": "/var/spool/cwl",
 965                 "output_ttl": 0,
 966                 "priority": 500,
 967                 "properties": {},
 968                 "runtime_constraints": {
 969                     "ram": 1073741824,
 970                     "vcpus": 1
 971                 },
 972                 "scheduling_parameters": {},
 973                 "secret_mounts": {},
 974                 "state": "Committed",
 975                 "use_existing": True
 976             }))
 977         mockc.open().__enter__().write.assert_has_calls([mock.call(subwf)])
 978         mockc.open().__enter__().write.assert_has_calls([mock.call(
 979 '''{
 980   "fileblub": {
 981     "basename": "token.txt",
 982     "class": "File",
 983     "location": "/keep/99999999999999999999999999999999+118/token.txt",
 984     "size": 0
 985   },
 986   "sleeptime": 5
 987 }''')])
 988
 989     # The test passes no builder.resources
 990     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 991     @mock.patch("arvados.collection.CollectionReader")
 992     @mock.patch("arvados.collection.Collection")
 993     @mock.patch('arvados.commands.keepdocker.list_images_in_arv')
 994     def test_overall_resource_singlecontainer(self, list_images_in_arv, mockcollection, mockcollectionreader):
 995         arvados_cwl.add_arv_hints()
 996
 997         api = mock.MagicMock()
 998         api._rootDesc = get_rootDesc()
 999
1000         runner = arvados_cwl.executor.ArvCwlExecutor(api)
1001         self.assertEqual(runner.work_api, 'containers')
1002
1003         list_images_in_arv.return_value = [["zzzzz-4zz18-zzzzzzzzzzzzzzz"]]
1004         runner.api.collections().get().execute.return_value = {"uuid": "zzzzz-4zz18-zzzzzzzzzzzzzzz",
1005                                                                "portable_data_hash": "99999999999999999999999999999993+99"}
1006         runner.api.collections().list().execute.return_value = {"items": [{"uuid": "zzzzz-4zz18-zzzzzzzzzzzzzzz",
1007                                                                            "portable_data_hash": "99999999999999999999999999999993+99"}]}
1008
1009         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
1010         runner.ignore_docker_for_reuse = False
1011         runner.num_retries = 0
1012         runner.secret_store = cwltool.secrets.SecretStore()
1013
1014         loadingContext, runtimeContext = self.helper(runner)
1015         runner.fs_access = runtimeContext.make_fs_access(runtimeContext.basedir)
1016         loadingContext.do_update = True
1017         tool, metadata = loadingContext.loader.resolve_ref("tests/wf/echo-wf.cwl")
1018
1019         mockcollection.side_effect = lambda *args, **kwargs: CollectionMock(mock.MagicMock(), *args, **kwargs)
1020
1021         arvtool = arvados_cwl.ArvadosWorkflow(runner, tool, loadingContext)
1022         arvtool.formatgraph = None
1023         it = arvtool.job({}, mock.MagicMock(), runtimeContext)
1024
1025         next(it).run(runtimeContext)
1026         next(it).run(runtimeContext)
1027
1028         with open("tests/wf/echo-subwf.cwl") as f:
1029             subwf = StripYAMLComments(f.read())
1030
1031         runner.api.container_requests().create.assert_called_with(
1032             body=JsonDiffMatcher({
1033                 'output_ttl': 0,
1034                 'environment': {'HOME': '/var/spool/cwl', 'TMPDIR': '/tmp'},
1035                 'scheduling_parameters': {},
1036                 'name': u'echo-subwf',
1037                 'secret_mounts': {},
1038                 'runtime_constraints': {'API': True, 'vcpus': 3, 'ram': 1073741824},
1039                 'properties': {},
1040                 'priority': 500,
1041                 'mounts': {
1042                     '/var/spool/cwl/cwl.input.yml': {
1043                         'portable_data_hash': '99999999999999999999999999999996+99',
1044                         'kind': 'collection',
1045                         'path': 'cwl.input.yml'
1046                     },
1047                     '/var/spool/cwl/workflow.cwl': {
1048                         'portable_data_hash': '99999999999999999999999999999996+99',
1049                         'kind': 'collection',
1050                         'path': 'workflow.cwl'
1051                     },
1052                     'stdout': {
1053                         'path': '/var/spool/cwl/cwl.output.json',
1054                         'kind': 'file'
1055                     },
1056                     '/tmp': {
1057                         'kind': 'tmp',
1058                         'capacity': 1073741824
1059                     }, '/var/spool/cwl': {
1060                         'kind': 'tmp',
1061                         'capacity': 3221225472
1062                     }
1063                 },
1064                 'state': 'Committed',
1065                 'output_path': '/var/spool/cwl',
1066                 'container_image': '99999999999999999999999999999993+99',
1067                 'command': [
1068                     u'cwltool',
1069                     u'--no-container',
1070                     u'--move-outputs',
1071                     u'--preserve-entire-environment',
1072                     u'workflow.cwl',
1073                     u'cwl.input.yml'
1074                 ],
1075                 'use_existing': True,
1076                 'output_name': u'Output for step echo-subwf',
1077                 'cwd': '/var/spool/cwl'
1078             }))
1079
1080     def test_default_work_api(self):
1081         arvados_cwl.add_arv_hints()
1082
1083         api = mock.MagicMock()
1084         api._rootDesc = copy.deepcopy(get_rootDesc())
1085         runner = arvados_cwl.executor.ArvCwlExecutor(api)
1086         self.assertEqual(runner.work_api, 'containers')