import unittest
import copy
import StringIO
-import datetime
import arvados
import arvados_cwl
logging.getLogger('arvados.cwl-runner').setLevel(logging.WARN)
logging.getLogger('arvados.arv-run').setLevel(logging.WARN)
-class MockDateTime(datetime.datetime):
- @classmethod
- def now(cls):
- return datetime.datetime(2018, 1, 1, 0, 0, 0, 0)
-
-datetime.datetime = MockDateTime
-
class TestJob(unittest.TestCase):
def helper(self, runner, enable_reuse=True):
arvjob.output_callback.assert_called_with({"out": "stuff"}, "success")
- def test_get_intermediate_collection_info(self):
- arvrunner = mock.MagicMock()
- arvrunner.intermediate_output_ttl = 60
- arvrunner.api.containers().current().execute.return_value = {"uuid" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"}
-
- job = arvados_cwl.ArvadosJob(arvrunner)
-
- info = job._get_intermediate_collection_info()
-
- self.assertEqual(info["name"], "Intermediate collection")
- self.assertEqual(info["trash_at"], datetime.datetime(2018, 1, 1, 0, 1))
- self.assertEqual(info["properties"], {"type" : "Intermediate", "container" : "zzzzz-8i9sb-zzzzzzzzzzzzzzz"})
-
-
class TestWorkflow(unittest.TestCase):
def helper(self, runner, enable_reuse=True):
metadata["cwlVersion"] = tool["cwlVersion"]
mockcollection().portable_data_hash.return_value = "99999999999999999999999999999999+118"
+ mockcollectionreader().find.return_value = arvados.arvfile.ArvadosFile(mock.MagicMock(), "token.txt")
arvtool = arvados_cwl.ArvadosWorkflow(runner, tool, loadingContext)
arvtool.formatgraph = None
"fileblub": {
"basename": "token.txt",
"class": "File",
- "location": "/keep/99999999999999999999999999999999+118/token.txt"
+ "location": "/keep/99999999999999999999999999999999+118/token.txt",
+ "size": 0
},
"sleeptime": 5
}''')])