Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/validate_js.py @ 0:26e78fe6e8c4 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author shellac
date Sat, 02 May 2020 07:14:21 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:26e78fe6e8c4
1 import copy
2 import itertools
3 import json
4 import logging
5 from collections import namedtuple
6 from typing import (cast, Any, Dict, List, MutableMapping, MutableSequence,
7 Optional, Tuple, Union)
8
9 from pkg_resources import resource_stream
10 from ruamel.yaml.comments import CommentedMap # pylint: disable=unused-import
11 from six import string_types
12 from typing_extensions import Text # pylint: disable=unused-import
13 from schema_salad import avro
14 from schema_salad.sourceline import SourceLine
15 from schema_salad.validate import Schema # pylint: disable=unused-import
16 from schema_salad.validate import validate_ex
17
18 from .errors import WorkflowException
19 from .expression import scanner as scan_expression
20 from .expression import SubstitutionError
21 from .loghandler import _logger
22 from .sandboxjs import code_fragment_to_js, exec_js_process
23 from .utils import json_dumps
24
25
26 def is_expression(tool, schema):
27 # type: (Any, Optional[Schema]) -> bool
28 return isinstance(schema, avro.schema.EnumSchema) \
29 and schema.name == "Expression" and isinstance(tool, string_types)
30
31 class SuppressLog(logging.Filter):
32 def __init__(self, name): # type: (Text) -> None
33 """Initialize this log suppressor."""
34 name = str(name)
35 super(SuppressLog, self).__init__(name)
36
37 def filter(self, record): # type: (logging.LogRecord) -> bool
38 return False
39
40
41 _logger_validation_warnings = logging.getLogger("cwltool.validation_warnings")
42 _logger_validation_warnings.addFilter(SuppressLog("cwltool.validation_warnings"))
43
44 def get_expressions(tool, # type: Union[CommentedMap, Text]
45 schema, # type: Optional[avro.schema.Schema]
46 source_line=None # type: Optional[SourceLine]
47 ): # type: (...) -> List[Tuple[Text, Optional[SourceLine]]]
48 if is_expression(tool, schema):
49 return [(cast(Text, tool), source_line)]
50 elif isinstance(schema, avro.schema.UnionSchema):
51 valid_schema = None
52
53 for possible_schema in schema.schemas:
54 if is_expression(tool, possible_schema):
55 return [(cast(Text, tool), source_line)]
56 elif validate_ex(possible_schema, tool, raise_ex=False,
57 logger=_logger_validation_warnings):
58 valid_schema = possible_schema
59
60 return get_expressions(tool, valid_schema, source_line)
61 elif isinstance(schema, avro.schema.ArraySchema):
62 if not isinstance(tool, MutableSequence):
63 return []
64
65 return list(itertools.chain(
66 *map(lambda x: get_expressions(x[1], schema.items, SourceLine(tool, x[0])), enumerate(tool))
67 ))
68
69 elif isinstance(schema, avro.schema.RecordSchema):
70 if not isinstance(tool, MutableMapping):
71 return []
72
73 expression_nodes = []
74
75 for schema_field in schema.fields:
76 if schema_field.name in tool:
77 expression_nodes.extend(get_expressions(
78 tool[schema_field.name],
79 schema_field.type,
80 SourceLine(tool, schema_field.name)
81 ))
82
83 return expression_nodes
84 else:
85 return []
86
87
88 JSHintJSReturn = namedtuple("jshint_return", ["errors", "globals"])
89
90 def jshint_js(js_text, # type: Text
91 globals=None, # type: Optional[List[Text]]
92 options=None # type: Optional[Dict[Text, Union[List[Text], Text, int]]]
93 ): # type: (...) -> Tuple[List[Text], List[Text]]
94 if globals is None:
95 globals = []
96 if options is None:
97 options = {
98 "includewarnings": [
99 "W117", # <VARIABLE> not defined
100 "W104", "W119" # using ES6 features
101 ],
102 "strict": "implied",
103 "esversion": 5
104 }
105
106 with resource_stream(__name__, "jshint/jshint.js") as file:
107 # NOTE: we need a global variable for lodash (which jshint depends on)
108 jshint_functions_text = "var global = this;" + file.read().decode('utf-8')
109
110 with resource_stream(__name__, "jshint/jshint_wrapper.js") as file:
111 # NOTE: we need to assign to ob, as the expression {validateJS: validateJS} as an expression
112 # is interpreted as a block with a label `validateJS`
113 jshint_functions_text += "\n" + file.read().decode('utf-8') + "\nvar ob = {validateJS: validateJS}; ob"
114
115 returncode, stdout, stderr = exec_js_process(
116 "validateJS(%s)" % json_dumps({
117 "code": js_text,
118 "options": options,
119 "globals": globals
120 }),
121 timeout=30,
122 context=jshint_functions_text
123 )
124
125 def dump_jshint_error():
126 # type: () -> None
127 raise RuntimeError("jshint failed to run succesfully\nreturncode: %d\nstdout: \"%s\"\nstderr: \"%s\"" % (
128 returncode,
129 stdout,
130 stderr
131 ))
132
133 if returncode == -1:
134 _logger.warning("jshint process timed out")
135
136 if returncode != 0:
137 dump_jshint_error()
138
139 try:
140 jshint_json = json.loads(stdout)
141 except ValueError:
142 dump_jshint_error()
143
144 jshint_errors = [] # type: List[Text]
145
146 js_text_lines = js_text.split("\n")
147
148 for jshint_error_obj in jshint_json.get("errors", []):
149 text = u"JSHINT: " + js_text_lines[jshint_error_obj["line"] - 1] + "\n"
150 text += u"JSHINT: " + " " * (jshint_error_obj["character"] - 1) + "^\n"
151 text += u"JSHINT: %s: %s" % (jshint_error_obj["code"], jshint_error_obj["reason"])
152 jshint_errors.append(text)
153
154 return JSHintJSReturn(jshint_errors, jshint_json.get("globals", []))
155
156
157 def print_js_hint_messages(js_hint_messages, source_line):
158 # type: (List[Text], Optional[SourceLine]) -> None
159 if source_line is not None:
160 for js_hint_message in js_hint_messages:
161 _logger.warning(source_line.makeError(js_hint_message))
162
163 def validate_js_expressions(tool, # type: CommentedMap
164 schema, # type: Schema
165 jshint_options=None # type: Optional[Dict[Text, Union[List[Text], Text, int]]]
166 ): # type: (...) -> None
167
168 if tool.get("requirements") is None:
169 return
170
171 requirements = tool["requirements"]
172
173 default_globals = [u"self", u"inputs", u"runtime", u"console"]
174
175 for prop in reversed(requirements):
176 if prop["class"] == "InlineJavascriptRequirement":
177 expression_lib = prop.get("expressionLib", [])
178 break
179 else:
180 return
181
182 js_globals = copy.deepcopy(default_globals)
183
184 for i, expression_lib_line in enumerate(expression_lib):
185 expression_lib_line_errors, expression_lib_line_globals = jshint_js(expression_lib_line, js_globals, jshint_options)
186 js_globals.extend(expression_lib_line_globals)
187 print_js_hint_messages(expression_lib_line_errors, SourceLine(expression_lib, i))
188
189 expressions = get_expressions(tool, schema)
190
191 for expression, source_line in expressions:
192 unscanned_str = expression.strip()
193 try:
194 scan_slice = scan_expression(unscanned_str)
195 except SubstitutionError as se:
196 if source_line:
197 source_line.raise_type = WorkflowException
198 raise source_line.makeError(str(se))
199 else:
200 raise se
201
202 while scan_slice:
203 if unscanned_str[scan_slice[0]] == '$':
204 code_fragment = unscanned_str[scan_slice[0] + 1:scan_slice[1]]
205 code_fragment_js = code_fragment_to_js(code_fragment, "")
206 expression_errors, _ = jshint_js(code_fragment_js, js_globals, jshint_options)
207 print_js_hint_messages(expression_errors, source_line)
208
209 unscanned_str = unscanned_str[scan_slice[1]:]
210 scan_slice = scan_expression(unscanned_str)