/usr/lib/python3/dist-packages/behave/formatter/base.py is in python3-behave 1.2.5-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 | # -*- coding: utf-8 -*-
import codecs
import os.path
import six
import sys
class StreamOpener(object):
"""
Provides a transport vehicle to open the formatter output stream
when the formatter needs it.
In addition, it provides the formatter with more control:
* when a stream is opened
* if a stream is opened at all
* the name (filename/dirname) of the output stream
* let it decide if directory mode is used instead of file mode
"""
default_encoding = "UTF-8"
def __init__(self, filename=None, stream=None, encoding=None):
if not encoding:
encoding = self.default_encoding
if stream:
stream = self.ensure_stream_with_encoder(stream, encoding)
self.name = filename
self.stream = stream
self.encoding = encoding
self.should_close_stream = not stream # Only for not pre-opened ones.
@staticmethod
def ensure_dir_exists(directory):
if directory and not os.path.isdir(directory):
os.makedirs(directory)
@classmethod
def ensure_stream_with_encoder(cls, stream, encoding=None):
if not encoding:
encoding = cls.default_encoding
if six.PY3:
return stream
elif hasattr(stream, "stream"):
return stream # Already wrapped with a codecs.StreamWriter
else:
assert six.PY2
# py2 does, however, sometimes declare an encoding on sys.stdout,
# even if it doesn't use it (or it might be explicitly None)
stream = codecs.getwriter(encoding)(stream)
# elif not getattr(stream, 'encoding', None):
# # -- TODO: POTENTIAL DEAD-CODE: Inspect and cleanup later.
# # ok, so the stream doesn't have an encoding at all so add one
# stream = codecs.getwriter(encoding)(stream)
return stream
def open(self):
if not self.stream or self.stream.closed:
self.ensure_dir_exists(os.path.dirname(self.name))
stream = open(self.name, "w")
# stream = codecs.open(self.name, "w", encoding=self.encoding)
stream = self.ensure_stream_with_encoder(stream, self.encoding)
self.stream = stream # -- Keep stream for house-keeping.
self.should_close_stream = True
assert self.should_close_stream
return self.stream
def close(self):
"""
Close the stream, if it was opened by this stream_opener.
Skip closing for sys.stdout and pre-opened streams.
:return: True, if stream was closed.
"""
closed = False
if self.stream and self.should_close_stream:
closed = getattr(self.stream, "closed", False)
if not closed:
self.stream.close()
closed = True
self.stream = None
return closed
class Formatter(object):
"""
Base class for all formatter classes.
A formatter is an extension point (variation point) for the runner logic.
A formatter is called while processing model elements.
Processing Logic (simplified, without ScenarioOutline and skip logic)::
for feature in runner.features:
formatter = make_formatters(...)
formatter.uri(feature.filename)
formatter.feature(feature)
for scenario in feature.scenarios:
formatter.scenario(scenario)
for step in scenario.all_steps:
formatter.step(step)
step_match = step_registry.find_match(step)
formatter.match(step_match)
if step_match:
step_match.run()
else:
step.status = "undefined"
formatter.result(step.status)
formatter.eof() # -- FEATURE-END
formatter.close()
"""
name = None
description = None
def __init__(self, stream_opener, config):
self.stream_opener = stream_opener
self.stream = stream_opener.stream
self.config = config
@property
def stdout_mode(self):
return not self.stream_opener.name
def open(self):
"""
Ensure that the output stream is open.
Triggers the stream opener protocol (if necessary).
:return: Output stream to use (just opened or already open).
"""
if not self.stream:
self.stream = self.stream_opener.open()
return self.stream
def uri(self, uri):
"""
Called before processing a file (normally a feature file).
:param uri: URI or filename (as string).
"""
pass
def feature(self, feature):
"""
Called before a feature is executed.
:param feature: Feature object (as :class:`behave.model.Feature`)
"""
pass
def background(self, background):
"""
Called when a (Feature) Background is provided.
Called after :method:`feature()` is called.
Called before processing any scenarios or scenario outlines.
:param background: Background object (as :class:`behave.model.Background`)
"""
pass
def scenario(self, scenario):
"""
Called before a scenario is executed (or an example of ScenarioOutline).
:param scenario: Scenario object (as :class:`behave.model.Scenario`)
"""
pass
def scenario_outline(self, outline):
pass
def examples(self, examples):
pass
def step(self, step):
"""
Called before a step is executed (and matched).
:param step: Step object (as :class:`behave.model.Step`)
"""
def match(self, match):
"""
Called when a step was matched against its step implementation.
:param match: Registered step (as Match), undefined step (as NoMatch).
"""
pass
def result(self, step_result):
"""
Called after processing a step (when the step result is known).
:param step_result: Step result (as string-enum).
"""
pass
def eof(self):
"""
Called after processing a feature (or a feature file).
"""
pass
def close(self):
"""
Called before the formatter is no longer used (stream/io compatibility).
"""
self.close_stream()
def close_stream(self):
"""
Close the stream, but only if this is needed.
This step is skipped if the stream is sys.stdout.
"""
if self.stream:
# -- DELEGATE STREAM-CLOSING: To stream_opener
assert self.stream is self.stream_opener.stream
self.stream_opener.close()
self.stream = None # -- MARK CLOSED.
|