5
5
6
6
import os
7
7
from functools import partial
8
+ import tempfile
8
9
9
10
import pytest
10
11
from six import StringIO
@@ -47,7 +48,7 @@ def test_pack_missing_cwlVersion(self):
47
48
# Testing single tool workflow
48
49
document_loader , workflowobj , uri = fetch_document (
49
50
get_data ("tests/wf/hello_single_tool.cwl" ))
50
- document_loader , avsc_names , processobj , metadata , uri = validate_document (
51
+ document_loader , _ , processobj , metadata , uri = validate_document (
51
52
document_loader , workflowobj , uri )
52
53
# generate pack output dict
53
54
packed = json .loads (print_pack (document_loader , processobj , uri , metadata ))
@@ -57,7 +58,7 @@ def test_pack_missing_cwlVersion(self):
57
58
# Testing single step workflow
58
59
document_loader , workflowobj , uri = fetch_document (
59
60
get_data ("tests/wf/hello-workflow.cwl" ))
60
- document_loader , avsc_names , processobj , metadata , uri = validate_document (
61
+ document_loader , _ , processobj , metadata , uri = validate_document (
61
62
document_loader , workflowobj , uri )
62
63
# generate pack output dict
63
64
packed = json .loads (print_pack (document_loader , processobj , uri , metadata ))
@@ -103,7 +104,7 @@ def test_packed_workflow_execution(self):
103
104
document_loader , avsc_names , processobj , metadata , uri = validate_document (
104
105
document_loader , workflowobj , uri )
105
106
packed = json .loads (print_pack (document_loader , processobj , uri , metadata ))
106
- temp_packed_path = "/tmp/packedwf"
107
+ temp_packed_path = tempfile . mkstemp ()[ 1 ]
107
108
with open (temp_packed_path , 'w' ) as f :
108
109
json .dump (packed , f )
109
110
normal_output = StringIO ()
@@ -115,16 +116,30 @@ def test_packed_workflow_execution(self):
115
116
get_data (test_wf_job )],
116
117
stdout = normal_output ), 0 )
117
118
self .assertEquals (json .loads (packed_output .getvalue ()), json .loads (normal_output .getvalue ()))
119
+ os .remove (temp_packed_path )
118
120
119
121
@pytest .mark .skipif (onWindows (),
120
122
reason = "Instance of cwltool is used, on Windows it invokes a default docker container"
121
123
"which is not supported on AppVeyor" )
122
124
def test_preserving_namespaces (self ):
123
125
test_wf = "tests/wf/formattest.cwl"
126
+ test_wf_job = "tests/wf/formattest-job.json"
124
127
document_loader , workflowobj , uri = fetch_document (
125
128
get_data (test_wf ))
126
129
document_loader , avsc_names , processobj , metadata , uri = validate_document (
127
130
document_loader , workflowobj , uri )
128
131
packed = json .loads (print_pack (document_loader , processobj , uri , metadata ))
129
132
assert "$namespaces" in packed
130
-
133
+ temp_packed_path = tempfile .mkstemp ()[1 ]
134
+ with open (temp_packed_path , 'w' ) as f :
135
+ json .dump (packed , f )
136
+ normal_output = StringIO ()
137
+ packed_output = StringIO ()
138
+ self .assertEquals (main (['--debug' , get_data (temp_packed_path ),
139
+ get_data (test_wf_job )],
140
+ stdout = packed_output ), 0 )
141
+ self .assertEquals (main ([get_data (test_wf ),
142
+ get_data (test_wf_job )],
143
+ stdout = normal_output ), 0 )
144
+ self .assertEquals (json .loads (packed_output .getvalue ()), json .loads (normal_output .getvalue ()))
145
+ os .remove (temp_packed_path )
0 commit comments