sdk/cwl/tests/test_job.py

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