1
- # from parser import load_document_by_uri, save
1
+ import importlib
2
2
from pathlib import Path
3
- from typing import Any , MutableSequence , Optional , cast
4
- from urllib .parse import unquote_plus , urlparse
3
+ from typing import Any , Dict , List , MutableSequence , Optional , Union , cast
4
+ from urllib .parse import urlparse
5
5
6
6
from ruamel .yaml .comments import CommentedMap
7
7
8
- import schema_salad .tests .cwl_v1_2 as cwl_v1_2
9
8
from schema_salad .utils import yaml_no_ts
9
+ from schema_salad import codegen
10
+ from schema_salad .avro .schema import Names
11
+ from schema_salad .exceptions import ValidationException
12
+ from schema_salad .schema import load_schema
10
13
11
- from .util import get_data
14
+ from .util import get_data , cwl_file_uri
12
15
13
-
14
- def test_secondary_files_dsl () -> None :
16
+ def test_secondary_files_dsl (tmp_path : Path ) -> None :
15
17
"""
16
18
Checks object is properly saving when dsl is used
17
19
"""
18
20
t = "test_secondary_files_dsl.cwl"
19
21
path = get_data ("tests/" + t )
20
- obj = load_document_by_uri (str (path ))
22
+ obj = load_document_by_uri (tmp_path , str (path ))
21
23
saved_obj = obj .save ()
22
24
assert isinstance (saved_obj , CommentedMap )
23
25
assert saved_obj .lc .data == {
@@ -49,13 +51,13 @@ def test_secondary_files_dsl() -> None:
49
51
}
50
52
51
53
52
- def test_outputs_before_inputs () -> None :
54
+ def test_outputs_before_inputs (tmp_path : Path ) -> None :
53
55
"""
54
56
Tests when output comes in cwl file before inputs
55
57
"""
56
58
t = "test_outputs_before_inputs.cwl"
57
59
path = get_data ("tests/" + t )
58
- obj = load_document_by_uri (str (path ))
60
+ obj = load_document_by_uri (tmp_path , str (path ))
59
61
saved_obj = obj .save ()
60
62
assert isinstance (saved_obj , CommentedMap )
61
63
assert {
@@ -80,15 +82,15 @@ def test_outputs_before_inputs() -> None:
80
82
}
81
83
82
84
83
- def test_type_dsl () -> None :
85
+ def test_type_dsl (tmp_path : Path ) -> None :
84
86
"""
85
87
Checks object is properly saving when type DSL is used.
86
88
In this example, type for the input is File? which should expand to
87
89
null, File.
88
90
"""
89
91
t = "test_type_dsl.cwl"
90
92
path = get_data ("tests/" + t )
91
- obj = load_document_by_uri (str (path ))
93
+ obj = load_document_by_uri (tmp_path , str (path ))
92
94
saved_obj = obj .save ()
93
95
assert isinstance (saved_obj , CommentedMap )
94
96
assert {
@@ -114,28 +116,37 @@ def test_type_dsl() -> None:
114
116
assert saved_obj ["outputs" ][0 ]["outputBinding" ].lc .data == {"glob" : [15 , 6 , 15 , 12 ]}
115
117
116
118
117
- def load_document_by_uri (path : str ) -> Any :
118
- """
119
- Takes in a path and loads it via the python codegen.
120
- """
121
- uri = urlparse (path )
122
- if not uri .scheme or uri .scheme == "file" :
123
- real_path = Path (unquote_plus (uri .path )).resolve ().as_uri ()
119
+ def load_document_by_uri (tmp_path : Path , path : Union [str , Path ]) -> Any :
120
+ src_target = tmp_path / "cwl_v1_0.py"
121
+ python_codegen (cwl_file_uri , src_target )
122
+ spec = importlib .util .spec_from_file_location ("cwl_v1_0" , src_target )
123
+ assert isinstance (spec , importlib .machinery .ModuleSpec )
124
+ assert isinstance (spec .loader , importlib .abc .Loader )
125
+ temp_cwl_v1_0 = importlib .util .module_from_spec (spec )
126
+ spec .loader .exec_module (temp_cwl_v1_0 )
127
+ cwl_v1_0 : Any = temp_cwl_v1_0
128
+
129
+ if isinstance (path , str ):
130
+ uri = urlparse (path )
131
+ if not uri .scheme or uri .scheme == "file" :
132
+ real_path = Path (uri .path ).resolve ().as_uri ()
133
+ else :
134
+ real_path = path
124
135
else :
125
- real_path = path
136
+ real_path = path . resolve (). as_uri ()
126
137
127
138
baseuri = str (real_path )
128
139
129
- loadingOptions = cwl_v1_2 .LoadingOptions (fileuri = baseuri )
130
- # doc = loadingOptions.fetcher.fetch_text(real_path)
131
- with open (path , 'r' ) as file :
132
- doc = file .read ()
140
+ loadingOptions = cwl_v1_0 .LoadingOptions (fileuri = baseuri )
133
141
142
+ with open (path , "r" ) as file :
143
+ doc = file .read ()
144
+ # doc = loadingOptions.fetcher.fetch_text(urllib.parse.unquote(str(real_path)))
134
145
yaml = yaml_no_ts ()
135
146
doc = yaml .load (doc )
136
147
137
- result = cwl_v1_2 .load_document_by_yaml (
138
- doc , baseuri , cast (Optional [cwl_v1_2 .LoadingOptions ], loadingOptions )
148
+ result = cwl_v1_0 .load_document_by_yaml (
149
+ doc , baseuri , cast (Optional [cwl_v1_0 .LoadingOptions ], loadingOptions )
139
150
)
140
151
141
152
if isinstance (result , MutableSequence ):
@@ -144,3 +155,25 @@ def load_document_by_uri(path: str) -> Any:
144
155
lst .append (r )
145
156
return lst
146
157
return result
158
+
159
+
160
+
161
+ def python_codegen (
162
+ file_uri : str ,
163
+ target : Path ,
164
+ parser_info : Optional [str ] = None ,
165
+ package : Optional [str ] = None ,
166
+ ) -> None :
167
+ document_loader , avsc_names , schema_metadata , metaschema_loader = load_schema (file_uri )
168
+ assert isinstance (avsc_names , Names )
169
+ schema_raw_doc = metaschema_loader .fetch (file_uri )
170
+ schema_doc , schema_metadata = metaschema_loader .resolve_all (schema_raw_doc , file_uri )
171
+ codegen .codegen (
172
+ "python" ,
173
+ cast (List [Dict [str , Any ]], schema_doc ),
174
+ schema_metadata ,
175
+ document_loader ,
176
+ target = str (target ),
177
+ parser_info = parser_info ,
178
+ package = package ,
179
+ )
0 commit comments