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