Robot Framework
resultbuilder.py
Go to the documentation of this file.
1 # Copyright 2008-2015 Nokia Networks
2 # Copyright 2016- Robot Framework Foundation
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 # http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 from robot.errors import DataError
17 from robot.model import SuiteVisitor
18 from robot.utils import ET, ETSource, get_error_message
19 
20 from .executionresult import Result, CombinedResult
21 from .flattenkeywordmatcher import (FlattenByNameMatcher, FlattenByTypeMatcher,
22  FlattenByTagMatcher)
23 from .merger import Merger
24 from .xmlelementhandlers import XmlElementHandler
25 
26 
27 
45 def ExecutionResult(*sources, **options):
46  if not sources:
47  raise DataError('One or more data source needed.')
48  if options.pop('merge', False):
49  return _merge_results(sources[0], sources[1:], options)
50  if len(sources) > 1:
51  return _combine_results(sources, options)
52  return _single_result(sources[0], options)
53 
54 
55 def _merge_results(original, merged, options):
56  result = ExecutionResult(original, **options)
57  merger = Merger(result, rpa=result.rpa)
58  for path in merged:
59  merged = ExecutionResult(path, **options)
60  merger.merge(merged)
61  return result
62 
63 
64 def _combine_results(sources, options):
65  return CombinedResult(ExecutionResult(src, **options) for src in sources)
66 
67 
68 def _single_result(source, options):
69  ets = ETSource(source)
70  result = Result(source, rpa=options.pop('rpa', None))
71  try:
72  return ExecutionResultBuilder(ets, **options).build(result)
73  except IOError as err:
74  error = err.strerror
75  except:
76  error = get_error_message()
77  raise DataError(f"Reading XML source '{ets}' failed: {error}")
78 
79 
80 
86 
87 
98  def __init__(self, source, include_keywords=True, flattened_keywords=None):
99  self._source_source = source \
100  if isinstance(source, ETSource) else ETSource(source)
101  self._include_keywords_include_keywords = include_keywords
102  self._flattened_keywords_flattened_keywords = flattened_keywords
103 
104  def build(self, result):
105  # Parsing is performance optimized. Do not change without profiling!
106  handler = XmlElementHandler(result)
107  with self._source_source as source:
108  self._parse_parse(source, handler.start, handler.end)
109  result.handle_suite_teardown_failures()
110  if not self._include_keywords_include_keywords:
111  result.suite.visit(RemoveKeywords())
112  return result
113 
114  def _parse(self, source, start, end):
115  context = ET.iterparse(source, events=('start', 'end'))
116  if not self._include_keywords_include_keywords:
117  context = self._omit_keywords_omit_keywords(context)
118  elif self._flattened_keywords_flattened_keywords:
119  context = self._flatten_keywords_flatten_keywords(context, self._flattened_keywords_flattened_keywords)
120  for event, elem in context:
121  if event == 'start':
122  start(elem)
123  else:
124  end(elem)
125  elem.clear()
126 
127  def _omit_keywords(self, context):
128  omitted_kws = 0
129  for event, elem in context:
130  # Teardowns aren't omitted yet to allow checking suite teardown status.
131  # They'll be removed later when not needed in `build()`.
132  omit = elem.tag in ('kw', 'for', 'if') and elem.get('type') != 'TEARDOWN'
133  start = event == 'start'
134  if omit and start:
135  omitted_kws += 1
136  if not omitted_kws:
137  yield event, elem
138  elif not start:
139  elem.clear()
140  if omit and not start:
141  omitted_kws -= 1
142 
143  def _flatten_keywords(self, context, flattened):
144  # Performance optimized. Do not change without profiling!
145  name_match, by_name = self._get_matcher_get_matcher(FlattenByNameMatcher, flattened)
146  type_match, by_type = self._get_matcher_get_matcher(FlattenByTypeMatcher, flattened)
147  tags_match, by_tags = self._get_matcher_get_matcher(FlattenByTagMatcher, flattened)
148  started = -1 # if 0 or more, we are flattening
149  tags = []
150  containers = {'kw', 'for', 'while', 'iter', 'if', 'try'}
151  inside_kw = 0 # to make sure we don't read tags from a test
152  seen_doc = False
153  for event, elem in context:
154  tag = elem.tag
155  start = event == 'start'
156  end = not start
157  if start:
158  if tag in containers:
159  inside_kw += 1
160  if started >= 0:
161  started += 1
162  elif by_name and name_match(elem.get('name', ''), elem.get('library')):
163  started = 0
164  seen_doc = False
165  elif by_type and type_match(tag):
166  started = 0
167  seen_doc = False
168  tags = []
169  else:
170  if tag in containers:
171  inside_kw -= 1
172  if started == 0 and not seen_doc:
173  doc = ET.Element('doc')
174  doc.text = '_*Content flattened.*_'
175  yield 'start', doc
176  yield 'end', doc
177  elif by_tags and inside_kw and started < 0 and tag == 'tag':
178  tags.append(elem.text or '')
179  if tags_match(tags):
180  started = 0
181  seen_doc = False
182  elif started == 0 and tag == 'doc':
183  seen_doc = True
184  elem.text = f"{elem.text or ''}\n\n_*Content flattened.*_".strip()
185  if started <= 0 or tag == 'msg':
186  yield event, elem
187  else:
188  elem.clear()
189  if started >= 0 and end and tag in containers:
190  started -= 1
191 
192  def _get_matcher(self, matcher_class, flattened):
193  matcher = matcher_class(flattened)
194  return matcher.match, bool(matcher)
195 
196 
198 
199  def start_suite(self, suite):
200  suite.setup = None
201  suite.teardown = None
202 
203  def visit_test(self, test):
204  test.body = []
Interface to ease traversing through a test suite structure.
Definition: visitor.py:85
Combined results of multiple test executions.
Builds :class:~.executionresult.Result objects based on output files.
def __init__(self, source, include_keywords=True, flattened_keywords=None)
:param source: Path to the XML output file to build :class:~.executionresult.Result objects from.
def _get_matcher(self, matcher_class, flattened)
def _flatten_keywords(self, context, flattened)
def visit_test(self, test)
Implements traversing through tests.
def start_suite(self, suite)
Called when a suite starts.
def _merge_results(original, merged, options)
def _single_result(source, options)
def _combine_results(sources, options)
def ExecutionResult(*sources, **options)
Factory method to constructs :class:~.executionresult.Result objects.
def get_error_message()
Returns error message of the last occurred exception.
Definition: error.py:34