sdk/cwl/tests/test_job.py

   1 import logging
   2 import mock
   3 import unittest
   4 import os
   5 import functools
   6 import json
   7
   8 import arvados_cwl
   9 import cwltool.process
  10 from schema_salad.ref_resolver import Loader
  11
  12 if not os.getenv('ARVADOS_DEBUG'):
  13     logging.getLogger('arvados.cwl-runner').setLevel(logging.WARN)
  14     logging.getLogger('arvados.arv-run').setLevel(logging.WARN)
  15
  16
  17 class TestJob(unittest.TestCase):
  18
  19     # The test passes no builder.resources
  20     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
  21     def test_run(self):
  22         runner = mock.MagicMock()
  23         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
  24         runner.ignore_docker_for_reuse = False
  25         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("draft-3")
  26
  27         tool = {
  28             "inputs": [],
  29             "outputs": [],
  30             "baseCommand": "ls",
  31             "arguments": [{"valueFrom": "$(runtime.outdir)"}]
  32         }
  33         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess, api_client=runner.api)
  34         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="jobs", avsc_names=avsc_names,
  35                                                  basedir="", make_fs_access=make_fs_access, loader=Loader({}))
  36         arvtool.formatgraph = None
  37         for j in arvtool.job({}, mock.MagicMock(), basedir="", make_fs_access=make_fs_access):
  38             j.run()
  39             runner.api.jobs().create.assert_called_with(
  40                 body={
  41                     'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
  42                     'runtime_constraints': {},
  43                     'script_parameters': {
  44                         'tasks': [{
  45                             'task.env': {'HOME': '$(task.outdir)', 'TMPDIR': '$(task.tmpdir)'},
  46                             'command': ['ls', '$(task.outdir)']
  47                         }],
  48                     },
  49                     'script_version': 'master',
  50                     'minimum_script_version': '9e5b98e8f5f4727856b53447191f9c06e3da2ba6',
  51                     'repository': 'arvados',
  52                     'script': 'crunchrunner',
  53                     'runtime_constraints': {
  54                         'docker_image': 'arvados/jobs',
  55                         'min_cores_per_node': 1,
  56                         'min_ram_mb_per_node': 1024,
  57                         'min_scratch_mb_per_node': 2048 # tmpdirSize + outdirSize
  58                     }
  59                 },
  60                 find_or_create=True,
  61                 filters=[['repository', '=', 'arvados'],
  62                          ['script', '=', 'crunchrunner'],
  63                          ['script_version', 'in git', '9e5b98e8f5f4727856b53447191f9c06e3da2ba6'],
  64                          ['docker_image_locator', 'in docker', 'arvados/jobs']]
  65             )
  66
  67     # The test passes some fields in builder.resources
  68     # For the remaining fields, the defaults will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
  69     def test_resource_requirements(self):
  70         runner = mock.MagicMock()
  71         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
  72         runner.ignore_docker_for_reuse = False
  73         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("draft-3")
  74
  75         tool = {
  76             "inputs": [],
  77             "outputs": [],
  78             "hints": [{
  79                 "class": "ResourceRequirement",
  80                 "coresMin": 3,
  81                 "ramMin": 3000,
  82                 "tmpdirMin": 4000
  83             }, {
  84                 "class": "http://arvados.org/cwl#RuntimeConstraints",
  85                 "keep_cache": 512
  86             }, {
  87                 "class": "http://arvados.org/cwl#APIRequirement",
  88             }],
  89             "baseCommand": "ls"
  90         }
  91         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess, api_client=runner.api)
  92         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="jobs", avsc_names=avsc_names,
  93                                                  make_fs_access=make_fs_access, loader=Loader({}))
  94         arvtool.formatgraph = None
  95         for j in arvtool.job({}, mock.MagicMock(), basedir="", make_fs_access=make_fs_access):
  96             j.run()
  97         runner.api.jobs().create.assert_called_with(
  98             body={
  99                 'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 100                 'runtime_constraints': {},
 101                 'script_parameters': {
 102                     'tasks': [{
 103                         'task.env': {'HOME': '$(task.outdir)', 'TMPDIR': '$(task.tmpdir)'},
 104                         'command': ['ls']
 105                     }]
 106             },
 107             'script_version': 'master',
 108                 'minimum_script_version': '9e5b98e8f5f4727856b53447191f9c06e3da2ba6',
 109                 'repository': 'arvados',
 110                 'script': 'crunchrunner',
 111                 'runtime_constraints': {
 112                     'docker_image': 'arvados/jobs',
 113                     'min_cores_per_node': 3,
 114                     'min_ram_mb_per_node': 3000,
 115                     'min_scratch_mb_per_node': 5024, # tmpdirSize + outdirSize
 116                     'keep_cache_mb_per_task': 512
 117                 }
 118             },
 119             find_or_create=True,
 120             filters=[['repository', '=', 'arvados'],
 121                      ['script', '=', 'crunchrunner'],
 122                      ['script_version', 'in git', '9e5b98e8f5f4727856b53447191f9c06e3da2ba6'],
 123                      ['docker_image_locator', 'in docker', 'arvados/jobs']])
 124
 125     @mock.patch("arvados.collection.Collection")
 126     def test_done(self, col):
 127         api = mock.MagicMock()
 128
 129         runner = mock.MagicMock()
 130         runner.api = api
 131         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 132         runner.num_retries = 0
 133         runner.ignore_docker_for_reuse = False
 134
 135         col().open.return_value = []
 136         api.collections().list().execute.side_effect = ({"items": []},
 137                                                         {"items": [{"manifest_text": "XYZ"}]})
 138
 139         arvjob = arvados_cwl.ArvadosJob(runner)
 140         arvjob.name = "testjob"
 141         arvjob.builder = mock.MagicMock()
 142         arvjob.output_callback = mock.MagicMock()
 143         arvjob.collect_outputs = mock.MagicMock()
 144
 145         arvjob.done({
 146             "state": "Complete",
 147             "output": "99999999999999999999999999999993+99",
 148             "log": "99999999999999999999999999999994+99",
 149             "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 150         })
 151
 152         api.collections().list.assert_has_calls([
 153             mock.call(),
 154             mock.call(filters=[['owner_uuid', '=', 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'],
 155                           ['portable_data_hash', '=', '99999999999999999999999999999993+99'],
 156                           ['name', '=', 'Output 9999999 of testjob']]),
 157             mock.call().execute(num_retries=0),
 158             mock.call(limit=1, filters=[['portable_data_hash', '=', '99999999999999999999999999999993+99']],
 159                  select=['manifest_text']),
 160             mock.call().execute(num_retries=0)])
 161
 162         api.collections().create.assert_called_with(
 163             ensure_unique_name=True,
 164             body={'portable_data_hash': '99999999999999999999999999999993+99',
 165                   'manifest_text': 'XYZ',
 166                   'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 167                   'name': 'Output 9999999 of testjob'})
 168
 169     @mock.patch("arvados.collection.Collection")
 170     def test_done_use_existing_collection(self, col):
 171         api = mock.MagicMock()
 172
 173         runner = mock.MagicMock()
 174         runner.api = api
 175         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 176         runner.num_retries = 0
 177
 178         col().open.return_value = []
 179         api.collections().list().execute.side_effect = ({"items": [{"uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2"}]},)
 180
 181         arvjob = arvados_cwl.ArvadosJob(runner)
 182         arvjob.name = "testjob"
 183         arvjob.builder = mock.MagicMock()
 184         arvjob.output_callback = mock.MagicMock()
 185         arvjob.collect_outputs = mock.MagicMock()
 186
 187         arvjob.done({
 188             "state": "Complete",
 189             "output": "99999999999999999999999999999993+99",
 190             "log": "99999999999999999999999999999994+99",
 191             "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 192         })
 193
 194         api.collections().list.assert_has_calls([
 195             mock.call(),
 196             mock.call(filters=[['owner_uuid', '=', 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'],
 197                                ['portable_data_hash', '=', '99999999999999999999999999999993+99'],
 198                                ['name', '=', 'Output 9999999 of testjob']]),
 199             mock.call().execute(num_retries=0)])
 200
 201         self.assertFalse(api.collections().create.called)
 202
 203
 204 class TestWorkflow(unittest.TestCase):
 205     # The test passes no builder.resources
 206     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
 207     @mock.patch("arvados.collection.Collection")
 208     def test_run(self, mockcollection):
 209         arvados_cwl.add_arv_hints()
 210
 211         runner = arvados_cwl.ArvCwlRunner(mock.MagicMock())
 212         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 213         runner.ignore_docker_for_reuse = False
 214         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("v1.0")
 215
 216         tool, metadata = document_loader.resolve_ref("tests/wf/scatter2.cwl")
 217         metadata["cwlVersion"] = tool["cwlVersion"]
 218
 219         mockcollection().portable_data_hash.return_value = "99999999999999999999999999999999+118"
 220
 221         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess, api_client=runner.api)
 222         arvtool = arvados_cwl.ArvadosWorkflow(runner, tool, work_api="jobs", avsc_names=avsc_names,
 223                                               basedir="", make_fs_access=make_fs_access, loader=document_loader,
 224                                               makeTool=runner.arv_make_tool, metadata=metadata)
 225         arvtool.formatgraph = None
 226         it = arvtool.job({}, mock.MagicMock(), basedir="", make_fs_access=make_fs_access)
 227         it.next().run()
 228         it.next().run()
 229
 230         with open("tests/wf/scatter2_subwf.cwl") as f:
 231             subwf = f.read()
 232
 233         mockcollection().open().__enter__().write.assert_has_calls([mock.call(subwf)])
 234         mockcollection().open().__enter__().write.assert_has_calls([mock.call('{sleeptime: 5}')])
 235
 236         runner.api.jobs().create.assert_called_with(
 237             body={
 238                 'minimum_script_version': '9e5b98e8f5f4727856b53447191f9c06e3da2ba6',
 239                 'repository': 'arvados',
 240                 'script_version': 'master',
 241                 'script': 'crunchrunner',
 242                 'script_parameters': {
 243                     'tasks': [{'task.env': {
 244                         'HOME': '$(task.outdir)',
 245                         'TMPDIR': '$(task.tmpdir)'},
 246                                'task.vwd': {
 247                                    'workflow.cwl': '$(task.keep)/99999999999999999999999999999999+118/workflow.cwl',
 248                                    'cwl.input.yml': '$(task.keep)/99999999999999999999999999999999+118/cwl.input.yml'
 249                                },
 250                     'command': [u'cwltool', u'--no-container', u'--move-outputs', u'--preserve-entire-environment', u'workflow.cwl#main', u'cwl.input.yml'],
 251                     'task.stdout': 'cwl.output.json'}]},
 252                 'runtime_constraints': {
 253                     'min_scratch_mb_per_node': 2048,
 254                     'min_cores_per_node': 1,
 255                     'docker_image': 'arvados/jobs',
 256                     'min_ram_mb_per_node': 1024
 257                 },
 258                 'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'},
 259             filters=[['repository', '=', 'arvados'],
 260                      ['script', '=', 'crunchrunner'],
 261                      ['script_version', 'in git', '9e5b98e8f5f4727856b53447191f9c06e3da2ba6'],
 262                      ['docker_image_locator', 'in docker', 'arvados/jobs']],
 263             find_or_create=True)