How to use _next_stream method in stestr

Best Python code snippet using stestr_python

support.py

Source:support.py Github

copy

Full Screen

1#LICENCE2'''3Support classes shared by various engines.4''' 5from operator import xor6from functools import reduce7from lepl.rxpy.parser.support import GroupState8from lepl.stream.core import s_next, s_empty, s_len9class Fail(Exception):10 '''11 Raised on failure.12 '''13 pass14class Match(Exception):15 '''16 Raised on success17 '''18 pass19class Loops(object):20 '''21 The state needed to track explicit repeats. This assumes that loops are 22 nested (as they must be).23 '''24 25 def __init__(self, counts=None, order=None):26 # counts for loops in nested order27 self.__counts = counts if counts else []28 # map from node to position in counts list29 self.__order = order if order else {}30 31 def increment(self, node):32 if node not in self.__order:33 order = len(self.__counts)34 self.__order[node] = order35 self.__counts.append(0)36 else:37 order = self.__order[node]38 self.__counts = self.__counts[0:order+1]39 self.__counts[order] += 140 return self.__counts[order]41 42 def drop(self, node):43 self.__counts = self.__counts[0:self.__order[node]]44 del self.__order[node]45 46 def clone(self):47 return Loops(list(self.__counts), dict(self.__order))48 49 def __eq__(self, other):50 return self.__counts == other.__counts and self.__order == other.__order51 52 def __hash__(self):53 return reduce(xor, map(hash, self.__counts), 0) ^ \54 reduce(xor, [hash(node) ^ hash(self.__order[node])55 for node in self.__order], 0)56class Groups(object):57 58 def __init__(self, group_state=None, stream=None,59 groups=None, offsets=None, last_index=None):60 '''61 `group_state` - The group definitions (GroupState)62 63 `text` - The text being matched64 65 Other arguments are internal for cloning.66 '''67 self.__state = group_state if group_state else GroupState()68 self.__stream = stream69 # map from index to (text, start, end)70 self.__groups = groups if groups else {}71 # map from index to start for pending groups72 self.__offsets = offsets if offsets else {}73 # last index matched74 self.__last_index = last_index75 # cache for str76 self.__str = None77 78 def start_group(self, number, offset):79 assert isinstance(number, int)80 self.__str = None81 self.__offsets[number] = offset82 83 def end_group(self, number, offset):84 assert isinstance(number, int)85 assert number in self.__offsets, 'Unopened group: ' + str(number)86 self.__str = None87 (_, stream) = s_next(self.__stream, self.__offsets[number])88 (text, _) = s_next(stream, offset - self.__offsets[number])89 self.__groups[number] = (text, self.__offsets[number], offset)90 del self.__offsets[number]91 if number: # avoid group 092 self.__last_index = number93 def __len__(self):94 return self.__state.count95 96 def __bool__(self):97 return bool(self.__groups)98 99 def __nonzero__(self):100 return self.__bool__()101 102 def __eq__(self, other):103 '''104 Ignores values from context (so does not work for comparison across 105 matches).106 '''107 return type(self) == type(other) and str(self) == str(other)108 109 def __hash__(self):110 '''111 Ignores values from context (so does not work for comparison across 112 matches).113 '''114 return hash(self.__str__())115 116 def __str__(self):117 '''118 Unique representation, used for eq and hash. Ignores values from 119 context (so does not work for comparison across matches).120 '''121 if not self.__str:122 def fmt_group(index):123 group = self.__groups[index]124 return str(group[1]) + ':' + str(group[2]) + ':' + repr(group[0])125 self.__str = ';'.join(str(index) + '=' + fmt_group(index)126 for index in sorted(self.__groups.keys())) + ' ' + \127 ';'.join(str(index) + ':' + str(self.__offsets[index])128 for index in sorted(self.__offsets.keys()))129 return self.__str130 def clone(self):131 return Groups(group_state=self.__state, stream=self.__stream,132 groups=dict(self.__groups), offsets=dict(self.__offsets), 133 last_index=self.__last_index)134 135 def data(self, number):136 if number in self.__state.names:137 index = self.__state.names[number]138 else:139 index = number140 try:141 return self.__groups[index]142 except KeyError:143 if isinstance(index, int) and index <= self.__state.count:144 return [None, -1, -1]145 else:146 raise IndexError(number)147 148 def group(self, number, default=None):149 group = self.data(number)[0]150 return default if group is None else group151 152 def start(self, number):153 return self.data(number)[1]154 155 def end(self, number):156 return self.data(number)[2]157 @property158 def last_index(self):159 return self.__last_index160 161 @property162 def last_group(self):163 return self.__state.indices.get(self.__last_index, None)164 165 @property166 def indices(self):167 return self.__state.indices.keys()168 169 @property170 def groups(self):171 return self.__groups172class StreamTargetMixin(object):173 '''174 Some common operations from `BaseMatchTarget` that are common across175 multiple engines (not necessarily in the engine class itself).176 Assumes self._parser_state is present in subclass.177 '''178 def _reset(self, offset, stream, previous):179 self._previous = previous180 self._stream = stream181 self._offset = offset182 self._excess = 0183 try:184 (self._current, self._next_stream) = s_next(stream)185 except StopIteration:186 self._current = None187 self._next_stream = None188 self._excess = 1189 def _advance(self, delta=1):190 '''191 Move forwards in the stream.192 I've tried to optimise for the common (delta=1) case.193 The following conventions are followed:194 - `offset` is the offset from the initial input195 - `stream` is the stream starting at the current location196 - `next_stream` is the stream after current197 - `current` is the character at the current location198 - `previous` is the character just before the current location199 - `excess` is the amount by which we advanced past the end200 If `excess` is set, streams should not be used.201 '''202 assert delta >= 0203 self._offset += delta204 if self._excess:205 self._excess += delta206 self._previous = None207 elif delta == 1:208 self._stream = self._next_stream209 self._previous = self._current210 try:211 (self._current, self._next_stream) = s_next(self._next_stream)212 except StopIteration:213 self._current = None214 self._next_stream = None215 self._excess = 1216 elif delta:217 old_stream = self._stream218 try:219 (advanced, self._stream) = s_next(old_stream, delta)220 self._previous = advanced[-1:]221 try:222 (self._current, self._next_stream) = s_next(self._stream)223 except StopIteration:224 self._current = None225 self._next_stream = None226 self._excess = 1227 except StopIteration:228 self._stream = None229 self._next_stream = None230 self._current = None231 self._previous = None232 self._excess = delta - s_len(old_stream) + 1233 return True234 @property235 def _final(self):236 '''Current character is last?'''237 if not self._excess:238 try:239 (_, stream) = s_next(self._stream)240 try:241 s_next(stream)242 except StopIteration:243 return True244 except StopIteration:245 pass246 return False247 def character(self, charset):248 if self._current is not None and self._current in charset:249 return True250 else:251 raise Fail252 def dot(self, multiline):253 current_str = self._parser_state.alphabet.letter_to_str(self._current)254 if self._current is not None and (multiline or current_str != '\\n'):255 return True256 else:257 raise Fail258 def start_of_line(self, multiline):259 previous_str = self._parser_state.alphabet.letter_to_str(self._previous)260 if self._offset == 0 or (multiline and previous_str == '\\n'):261 return False262 else:263 raise Fail264 def end_of_line(self, multiline):265 if self._current is None:266 return False267 current_str = self._parser_state.alphabet.letter_to_str(self._current)268 if ((multiline or self._final) and current_str == '\\n'):269 return False270 raise Fail271 def word_boundary(self, inverted):272 word = self._parser_state.alphabet.word273 flags = self._parser_state.flags274 boundary = word(self._current, flags) != word(self._previous, flags)275 if boundary != inverted:276 return False277 else:278 raise Fail279 def digit(self, inverted):280 if self._current is not None and \281 self._parser_state.alphabet.digit(self._current,282 self._parser_state.flags) != inverted:283 return True284 else:285 raise Fail286 def space(self, inverted):287 if self._current is not None and \288 self._parser_state.alphabet.space(self._current,289 self._parser_state.flags) != inverted:290 return True291 else:292 raise Fail293 def word(self, inverted):294 if self._current is not None and \295 self._parser_state.alphabet.word(self._current,296 self._parser_state.flags) != inverted:297 return True298 else:...

Full Screen

Full Screen

basic_stream.py

Source:basic_stream.py Github

copy

Full Screen

1__all__ = ["BasicStream"]2class BasicStream:3 def __init__(self, stream_name: str = None) -> None:4 self.name = stream_name5 self._next_stream = None6 def input(self, data, *args, **kwargs):7 """8 read data from buffer9 :param data: data that move in the stream10 :return data: None if there is nothing to return11 """12 return None13 def connect(self, next_stream, *args, **kwargs) -> None:14 self._next_stream = next_stream15 def next_stream(self):...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run stestr automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful