sdk/cwl/tests/test_job.py

   1 import arvados_cwl
   2 import logging
   3 import mock
   4 import unittest
   5 import os
   6 import functools
   7 import cwltool.process
   8
   9 from schema_salad.ref_resolver import Loader
  10
  11 if not os.getenv('ARVADOS_DEBUG'):
  12     logging.getLogger('arvados.cwl-runner').setLevel(logging.WARN)
  13     logging.getLogger('arvados.arv-run').setLevel(logging.WARN)
  14
  15
  16 class TestJob(unittest.TestCase):
  17
  18     # The test passes no builder.resources
  19     # Hence the default resources will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
  20     def test_run(self):
  21         runner = mock.MagicMock()
  22         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
  23         runner.ignore_docker_for_reuse = False
  24         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("draft-3")
  25
  26         tool = {
  27             "inputs": [],
  28             "outputs": [],
  29             "baseCommand": "ls",
  30             "arguments": [{"valueFrom": "$(runtime.outdir)"}]
  31         }
  32         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess, api_client=runner.api)
  33         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="jobs", avsc_names=avsc_names,
  34                                                  basedir="", make_fs_access=make_fs_access, loader=Loader({}))
  35         arvtool.formatgraph = None
  36         for j in arvtool.job({}, mock.MagicMock(), basedir="", make_fs_access=make_fs_access):
  37             j.run()
  38             runner.api.jobs().create.assert_called_with(
  39                 body={
  40                     'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
  41                     'runtime_constraints': {},
  42                     'script_parameters': {
  43                         'tasks': [{
  44                             'task.env': {'HOME': '$(task.outdir)', 'TMPDIR': '$(task.tmpdir)'},
  45                             'command': ['ls', '$(task.outdir)']
  46                         }],
  47                     },
  48                     'script_version': 'master',
  49                     'minimum_script_version': '9e5b98e8f5f4727856b53447191f9c06e3da2ba6',
  50                     'repository': 'arvados',
  51                     'script': 'crunchrunner',
  52                     'runtime_constraints': {
  53                         'docker_image': 'arvados/jobs',
  54                         'min_cores_per_node': 1,
  55                         'min_ram_mb_per_node': 1024,
  56                         'min_scratch_mb_per_node': 2048 # tmpdirSize + outdirSize
  57                     }
  58                 },
  59                 find_or_create=True,
  60                 filters=[['repository', '=', 'arvados'],
  61                          ['script', '=', 'crunchrunner'],
  62                          ['script_version', 'in git', '9e5b98e8f5f4727856b53447191f9c06e3da2ba6'],
  63                          ['docker_image_locator', 'in docker', 'arvados/jobs']]
  64             )
  65
  66     # The test passes some fields in builder.resources
  67     # For the remaining fields, the defaults will apply: {'cores': 1, 'ram': 1024, 'outdirSize': 1024, 'tmpdirSize': 1024}
  68     def test_resource_requirements(self):
  69         runner = mock.MagicMock()
  70         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
  71         runner.ignore_docker_for_reuse = False
  72         document_loader, avsc_names, schema_metadata, metaschema_loader = cwltool.process.get_schema("draft-3")
  73
  74         tool = {
  75             "inputs": [],
  76             "outputs": [],
  77             "hints": [{
  78                 "class": "ResourceRequirement",
  79                 "coresMin": 3,
  80                 "ramMin": 3000,
  81                 "tmpdirMin": 4000
  82             }],
  83             "baseCommand": "ls"
  84         }
  85         make_fs_access=functools.partial(arvados_cwl.CollectionFsAccess, api_client=runner.api)
  86         arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="jobs", avsc_names=avsc_names,
  87                                                  make_fs_access=make_fs_access, loader=Loader({}))
  88         arvtool.formatgraph = None
  89         for j in arvtool.job({}, mock.MagicMock(), basedir="", make_fs_access=make_fs_access):
  90             j.run()
  91         runner.api.jobs().create.assert_called_with(
  92             body={
  93                 'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
  94                 'runtime_constraints': {},
  95                 'script_parameters': {
  96                     'tasks': [{
  97                         'task.env': {'HOME': '$(task.outdir)', 'TMPDIR': '$(task.tmpdir)'},
  98                         'command': ['ls']
  99                     }]
 100             },
 101             'script_version': 'master',
 102                 'minimum_script_version': '9e5b98e8f5f4727856b53447191f9c06e3da2ba6',
 103                 'repository': 'arvados',
 104                 'script': 'crunchrunner',
 105                 'runtime_constraints': {
 106                     'docker_image': 'arvados/jobs',
 107                     'min_cores_per_node': 3,
 108                     'min_ram_mb_per_node': 3000,
 109                     'min_scratch_mb_per_node': 5024 # tmpdirSize + outdirSize
 110                 }
 111             },
 112             find_or_create=True,
 113             filters=[['repository', '=', 'arvados'],
 114                      ['script', '=', 'crunchrunner'],
 115                      ['script_version', 'in git', '9e5b98e8f5f4727856b53447191f9c06e3da2ba6'],
 116                      ['docker_image_locator', 'in docker', 'arvados/jobs']])
 117
 118     @mock.patch("arvados.collection.Collection")
 119     def test_done(self, col):
 120         api = mock.MagicMock()
 121
 122         runner = mock.MagicMock()
 123         runner.api = api
 124         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 125         runner.num_retries = 0
 126         runner.ignore_docker_for_reuse = False
 127
 128         col().open.return_value = []
 129         api.collections().list().execute.side_effect = ({"items": []},
 130                                                         {"items": [{"manifest_text": "XYZ"}]})
 131
 132         arvjob = arvados_cwl.ArvadosJob(runner)
 133         arvjob.name = "testjob"
 134         arvjob.builder = mock.MagicMock()
 135         arvjob.output_callback = mock.MagicMock()
 136         arvjob.collect_outputs = mock.MagicMock()
 137
 138         arvjob.done({
 139             "state": "Complete",
 140             "output": "99999999999999999999999999999993+99",
 141             "log": "99999999999999999999999999999994+99",
 142             "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 143         })
 144
 145         api.collections().list.assert_has_calls([
 146             mock.call(),
 147             mock.call(filters=[['owner_uuid', '=', 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'],
 148                           ['portable_data_hash', '=', '99999999999999999999999999999993+99'],
 149                           ['name', '=', 'Output 9999999 of testjob']]),
 150             mock.call().execute(num_retries=0),
 151             mock.call(limit=1, filters=[['portable_data_hash', '=', '99999999999999999999999999999993+99']],
 152                  select=['manifest_text']),
 153             mock.call().execute(num_retries=0)])
 154
 155         api.collections().create.assert_called_with(
 156             ensure_unique_name=True,
 157             body={'portable_data_hash': '99999999999999999999999999999993+99',
 158                   'manifest_text': 'XYZ',
 159                   'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz',
 160                   'name': 'Output 9999999 of testjob'})
 161
 162     @mock.patch("arvados.collection.Collection")
 163     def test_done_use_existing_collection(self, col):
 164         api = mock.MagicMock()
 165
 166         runner = mock.MagicMock()
 167         runner.api = api
 168         runner.project_uuid = "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 169         runner.num_retries = 0
 170
 171         col().open.return_value = []
 172         api.collections().list().execute.side_effect = ({"items": [{"uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2"}]},)
 173
 174         arvjob = arvados_cwl.ArvadosJob(runner)
 175         arvjob.name = "testjob"
 176         arvjob.builder = mock.MagicMock()
 177         arvjob.output_callback = mock.MagicMock()
 178         arvjob.collect_outputs = mock.MagicMock()
 179
 180         arvjob.done({
 181             "state": "Complete",
 182             "output": "99999999999999999999999999999993+99",
 183             "log": "99999999999999999999999999999994+99",
 184             "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
 185         })
 186
 187         api.collections().list.assert_has_calls([
 188             mock.call(),
 189             mock.call(filters=[['owner_uuid', '=', 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'],
 190                                ['portable_data_hash', '=', '99999999999999999999999999999993+99'],
 191                                ['name', '=', 'Output 9999999 of testjob']]),
 192             mock.call().execute(num_retries=0)])
 193
 194         self.assertFalse(api.collections().create.called)