KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _next_stream method in stestr

Best Python code snippet using stestr_python

support.py

Source:support.py

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:...

basic_stream.py

Source:basic_stream.py

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):...

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.