Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _flush_buffer method in autotest

Best Python code snippet using autotest_python

rst2ipynb.py

Source:rst2ipynb.py

...136                if self._state['need_hanging_indent']:137                    self._state['need_hanging_indent'] = False138                else:139                    self._add2buffer('', newline=True, paragraph=True)140                self._flush_buffer()141                self._parse(child)142            # FIXME: literal_block likely needs better handling143            elif tag == 'literal_block':144                self._add_markdowncell()145                if self._state['need_hanging_indent']:146                    self._state['need_hanging_indent'] = False147                else:148                    self._add2buffer('', newline=True, paragraph=True)149                self._flush_buffer()150                self._parse(child)151            elif tag == 'inline':152                print("warning, no idea how to handle ``inline``")153                # FIXME:154            elif tag == 'raw':155                self._add_markdowncell()156                self._flush_buffer()157                self._currcell['source'].insert(0, child.astext())158                self._store_currcell()159            elif tag == 'doctest_block':160                self._add_codecell()161                needs_new_codecell = False162                for ex in self._doctest_parser.get_examples(child.rawsource):163                    if needs_new_codecell:164                        self._add_codecell()165                    self._add2buffer('%s%s' % (' ' * ex.indent, ex.source),166                                     newline=False)167                    self._flush_buffer(startnew=True)168                    needs_new_codecell = len(ex.want) > 0169            elif tag == 'section':170                self._state['sec_depth'] += 1171                self._parse(child)172                self._state['sec_depth'] -= 1173            elif tag == 'note':174                self._add_markdowncell(force=True)175                self._parse(child)176                self._flush_buffer()177                self._currcell['source'].insert(0, '- - -\n*Note*')178                self._currcell['source'].append('- - -\n')179                self._store_currcell()180            elif tag == 'title_reference':181                self._flush_buffer()182                self._parse(child)183                if self._buffer.startswith(':term:'):184                    # link to glossary185                    term = re.match('.*<(.*)>', self._buffer)186                    if term is None:187                        term = re.match(':term:(.*)', self._buffer).groups()[0]188                        term_text = term189                    else:190                        term = term.groups()[0]191                        term_text = re.match(':term:(.*) <', self._buffer).groups()[0]192                    self._buffer = '[%s](%s#term-%s)' % (term_text,193                                                 self._glossary_baseurl,194                                                 term.lower().replace(' ', '-'))195                elif self._buffer.startswith('~mvpa') \196                     or self._buffer.startswith('mvpa') \197                     or self._buffer.startswith(':meth:') \198                     or self._buffer.startswith(':mod:') \199                     or self._buffer.startswith(':class:') \200                     or self._buffer.startswith(':func:'):201                    # various API reference link variants202                    self._buffer = self._ref2apiref(self._buffer)203                # XXX for the rest I have no idea how to link them without huge204                # effort205                elif self._buffer.startswith(':ref:'):206                    self._buffer = '*%s*' \207                            % [m for m in re.match(':ref:(.*) <|:ref:(.*)',208                                                   self._buffer).groups()209                                                        if not m is None][0]210                elif self._buffer.startswith(':math:'):211                    self._buffer = '$$%s$$' % self._buffer212                elif re.match(':([a-z]+):', self._buffer):213                    # catch other ref type we should handle, but do not yet214                    raise RuntimeError("unhandled reference type '%s'" % self._buffer)215                else:216                    # plain refs seems to be mostly used for external API217                    self._buffer = '`%s`' % self._buffer218            elif tag == 'emphasis':219                self._flush_buffer()220                self._parse(child)221                self._buffer = '*%s*' % self._buffer222            elif tag == 'strong':223                self._flush_buffer()224                self._parse(child)225                self._buffer = '**%s**' % self._buffer226            elif tag == 'literal':227                # strip one layer of backticks228                self._add2buffer(child.rawsource[1:-1])229            elif tag == 'problematic':230                print('PROBLEMATIC: %s' % child)231                self._parse(child)232            elif tag == 'reference':233                self._flush_buffer()234                self._parse(child)235                self._buffer = '[%s][%s]' % (self._buffer,236                                             child.attributes['name'])237            elif tag in ['comment', 'target']:238                pass239            elif tag == 'definition_list':240                self._add_markdowncell()241                for item in child.children:242                    self._flush_buffer()243                    self._parse(item.children[0])244                    term = self._buffer245                    self._buffer = ''246                    self._parse(item.children[1])247                    self._buffer = '\n%s: %s' % (term, self._buffer)248            elif tag in ['enumerated_list', 'bullet_list']:249                self._add_markdowncell()250                for i, item in enumerate(child.children):251                    if tag == 'enumerated_list':252                        prefix = '%i.' % (i + 1,)253                    else:254                        prefix = '*'255                    self._flush_buffer()256                    self._add2buffer('%s ' % prefix,257                                     newline=True, paragraph=True)258                    self._state['indent'] += 4259                    self._state['need_hanging_indent'] = True260                    self._parse(item)261                    self._state['indent'] -= 4262                self._flush_buffer()263            elif tag == 'list_item':264                for c in child.children:265                    self._parse(c)266            elif tag == 'term':267                self._parse(child.children[0])268            elif tag == 'figure':269                # this can't be expressed in markdown270                self._flush_buffer()271                file_url = '%s/%s.html' % (self._baseurl,272                                           os.path.splitext(os.path.basename(self._filename))[0])273                self._add2buffer('\[Visit [%s](%s) to view this figure\]' % (file_url, file_url),274                                 newline=True, paragraph=True)275            elif tag == 'block_quote':276                self._flush_buffer()277                first_line = len(self._currcell['source'])278                # skip the wrapping paragraph279                self._parse(child.children[0])280                self._flush_buffer()281                self._currcell['source'][first_line] = \282                        '\n\n> %s' % self._currcell['source'][first_line]283            elif tag == 'system_message':284                if child.attributes['type'] == 'INFO':285                    pass286                elif child.children[0].astext() == 'Unknown directive type "exercise".':287                    exercise_text = \288                        '\n'.join([l.strip()289                            for l in child.children[1][0].astext().split('\n')][2:])290                    self._add_markdowncell(force=True)291                    self._parse(publish_doctree(292                                    exercise_text,293                                    settings_overrides=self._doctree_parser_settings))294                    self._flush_buffer()295                    self._currcell['source'].insert(0, '- - -\n*Exercise*')296                    self._add_codecell()297                    self._add2buffer('# you can use this cell to for this exercise\n')298                    self._add_markdowncell()299                    self._currcell['source'].append('- - -\n')300                elif child.children[0].astext() == 'Unknown directive type "todo".':301                    pass302                elif child.children[0].astext() == 'Unknown directive type "tikz".':303                    pass304                elif child.children[0].astext() == 'Unknown directive type "ipython".':305                    python_code = \306                        '\n'.join([l.strip()307                            for l in child.children[1][0].astext().split('\n')][2:])308                    self._flush_buffer()309                    self._add_codecell()310                    self._currcell['input'].insert(0, python_code)311                    self._store_currcell()312                else:313                    raise RuntimeError("cannot handle system message '%s'"314                                       % child.astext())315            else:316                if hasattr(child, 'line') and child.line:317                    line = ' on line %i' % child.line318                else:319                    line = ''320                raise RuntimeError("Unknown tag '%s'%s" % (tag, line))321    def _store_currcell(self):322        if not self._currcell is None:323            self._flush_buffer()324            if self._currcell['cell_type'] == 'code':325                # remove last newline to save on vertical space326                self._currcell['input'][-1] = self._currcell['input'][-1].rstrip('\n')327            self._currcells.append(self._currcell)328        self._currcell = None329    def _add_headingcell(self, level):330        self._store_currcell()331        self._currcell = deepcopy(ReST2IPyNB.headingcell_template)332        self._currcell['level'] = level333    def _add_codecell(self):334        self._store_currcell()335        self._currcell = deepcopy(ReST2IPyNB.codecell_template)336    def _add_markdowncell(self, force=False):337        if self._currcell is None \338           or not self._currcell['cell_type'] == 'markdown' \339           or force:340            # we need a new cell341            self._store_currcell()342            self._currcell = deepcopy(ReST2IPyNB.markdowncell_template)343    def _add2buffer(self, value, newline=False, paragraph=False):344        if paragraph:345            nl = '\n\n'346        else:347            nl = '\n'348        if newline:349            self._buffer += '%s%s%s' % (nl, ' ' * self._state['indent'], value)350        else:351            self._buffer += value352    def _flush_buffer(self, startnew=True):353        if not len(self._buffer):354            return355        if self._currcell['cell_type'] == 'code':356            target_field = 'input'357        else:358            target_field = 'source'359        if startnew or not len(self._currcell[target_field]):360            self._currcell[target_field].append(self._buffer)361        else:362            self._currcell[target_field][-1] += self._buffer363        self._buffer = ''364def main():365    from optparse import OptionParser366    parser = OptionParser( \...

rst2ipnbpy

Source:rst2ipnbpy

...141                if self._state['need_hanging_indent']:142                    self._state['need_hanging_indent'] = False143                else:144                    self._add2buffer('', newline=True, paragraph=True)145                self._flush_buffer()146                self._parse(child)147            elif tag == 'doctest_block':148                self._add_codecell()149                needs_new_codecell = False150                for ex in self._doctest_parser.get_examples(child.rawsource):151                    if needs_new_codecell:152                        self._add_codecell()153                    # Filter out possible doctest directives154                    source = re.sub('#  *doctest:[^\n]*', '', ex.source)155                    self._add2buffer('%s%s' % (' ' * ex.indent, source),156                                     newline=False)157                    self._flush_buffer(startnew=True)158                    needs_new_codecell = len(ex.want) > 0159            elif tag == 'literal_block':160                self._add_codecell()161                self._add2buffer(child.rawsource)162                self._flush_buffer(startnew=True)163            elif tag == 'section':164                self._state['sec_depth'] += 1165                self._parse(child)166                self._state['sec_depth'] -= 1167            elif tag == 'note':168                self._add_markdowncell(force=True)169                self._parse(child)170                self._flush_buffer()171                self._currcell['source'].insert(0, '- - -\n*Note*')172                self._currcell['source'].append('- - -\n')173                self._store_currcell()174            elif tag == 'title_reference':175                self._flush_buffer()176                self._parse(child)177                if self._buffer.startswith(':term:'):178                    # link to glossary179                    term = re.match('.*<(.*)>', self._buffer)180                    if term is None:181                        term = re.match(':term:(.*)', self._buffer).groups()[0]182                        term_text = term183                    else:184                        term = term.groups()[0]185                        term_text = re.match(':term:(.*) <', self._buffer).groups()[0]186                    self._buffer = '[%s](%s#term-%s)' % (term_text,187                                                 self._glossary_baseurl,188                                                 term.lower().replace(' ', '-'))189                elif self._buffer.startswith('~mvpa') \190                     or self._buffer.startswith('mvpa') \191                     or self._buffer.startswith(':meth:') \192                     or self._buffer.startswith(':mod:') \193                     or self._buffer.startswith(':class:') \194                     or self._buffer.startswith(':func:'):195                    # various API reference link variants196                    self._buffer = self._ref2apiref(self._buffer)197                # XXX for the rest I have no idea how to link them without huge198                # effort199                elif self._buffer.startswith(':ref:'):200                    self._buffer = '*%s*' \201                            % [m for m in re.match(':ref:(.*) <|:ref:(.*)',202                                                   self._buffer).groups()203                                                        if m is not None][0]204                elif self._buffer.startswith(':math:'):205                    self._buffer = '$%s$' \206                            % [m for m in re.match(':math:(.*) <|:math:(.*)',207                                                   self._buffer).groups()208                                                        if m is not None][0]209                elif re.match(':([a-z]+):', self._buffer):210                    # catch other ref type we should handle, but do not yet211                    raise RuntimeError("unhandled reference type '%s'" % self._buffer)212                else:213                    # plain refs seems to be mostly used for external API214                    self._buffer = '`%s`' % self._buffer215            elif tag == 'emphasis':216                self._flush_buffer()217                self._parse(child)218                self._buffer = '*%s*' % self._buffer219            elif tag == 'strong':220                self._flush_buffer()221                self._parse(child)222                self._buffer = '**%s**' % self._buffer223            elif tag == 'literal':224                # strip one layer of backticks225                self._add2buffer(child.rawsource[1:-1])226            elif tag == 'math_block':227                # wrap in $$ $$228                for ch in child.children:229                    self._add2buffer('\n$$%s$$\n' % ch)230            elif tag == 'problematic':231                print 'PROBLEMATIC: %s' % child232                self._parse(child)233            elif tag == 'reference':234                self._flush_buffer()235                self._parse(child)236                if 'name' in child.attributes:237                    self._buffer = '[%(name)s](%(refuri)s)' % child.attributes238                else:239                    self._buffer = '%(refuri)s' % child.attributes240            elif tag in ['comment', 'target']:241                pass242            elif tag == 'definition_list':243                self._add_markdowncell()244                for item in child.children:245                    self._flush_buffer()246                    self._parse(item.children[0])247                    term = self._buffer248                    self._buffer = ''249                    self._parse(item.children[1])250                    self._buffer = '\n%s: %s' % (term, self._buffer)251            elif tag in ['enumerated_list', 'bullet_list']:252                # TODO: has problems with nested lists formatting253                self._add_markdowncell()254                for i, item in enumerate(child.children):255                    if tag == 'enumerated_list':256                        prefix = '%i.' % (i + 1,)257                    else:258                        prefix = '*'259                    self._flush_buffer()260                    self._add2buffer('%s ' % prefix,261                                     newline=True, paragraph=True)262                    self._state['indent'] += 4263                    self._state['need_hanging_indent'] = True264                    self._parse(item)265                    self._state['indent'] -= 4266                self._flush_buffer()267            elif tag == 'list_item':268                for c in child.children:269                    self._parse(c)270            elif tag == 'term':271                self._parse(child.children[0])272            elif tag in ['figure', 'image'] :273                # this can't be expressed in markdown274                self._add_markdowncell()275                file_url = '%s/%s.html' % (self._baseurl,276                                           os.path.splitext(os.path.basename(self._filename))[0])277                self._add2buffer('\[Visit [%s](%s) to view this figure\]' % (file_url, file_url),278                                 newline=True, paragraph=True)279            elif tag == 'block_quote':280                self._flush_buffer()281                first_line = len(self._currcell['source']) - 1282                # skip the wrapping paragraph283                self._parse(child.children[0])284                self._flush_buffer()285                self._currcell['source'][first_line] = \286                        '\n\n> %s' % self._currcell['source'][first_line]287            elif tag == 'system_message':288                if child.attributes['type'] == 'INFO':289                    pass290                elif child.children[0].astext() == 'Unknown directive type "exercise".':291                    exercise_text = \292                        '\n'.join([l.strip()293                            for l in child.children[1][0].astext().split('\n')][2:])294                    self._add_markdowncell(force=True)295                    self._parse(publish_doctree(296                                    exercise_text,297                                    settings_overrides=self._doctree_parser_settings))298                    self._flush_buffer()299                    self._currcell['source'].insert(0, '- - -\n**Exercise**')300                    self._add_codecell()301                    self._add2buffer('# you can use this cell for this exercise\n')302                    self._add_markdowncell()303                    self._currcell['source'].append('- - -\n')304                elif re.match('Unknown directive type "(seealso|style|todo|layout)".',305                              child.children[0].astext()):306                    pass307                else:308                    raise RuntimeError("cannot handle system message '%s'"309                                       % child.astext())310            else:311                if hasattr(child, 'line') and child.line:312                    line = ' on line %i' % child.line313                else:314                    line = ''315                raise RuntimeError("Unknown tag '%s'%s" % (tag, line))316    def _store_currcell(self):317        if self._currcell is not None:318            self._flush_buffer()319            if self._currcell['cell_type'] == 'code':320                # remove last newline to save on vertical space321                self._currcell['input'][-1] = self._currcell['input'][-1].rstrip('\n')322            self._currcells.append(self._currcell)323        self._currcell = None324    def _add_headingcell(self, level):325        self._store_currcell()326        self._currcell = deepcopy(ReST2IPyNB.headingcell_template)327        self._currcell['level'] = level328    def _add_codecell(self):329        self._store_currcell()330        self._currcell = deepcopy(ReST2IPyNB.codecell_template)331    def _add_markdowncell(self, force=False):332        if self._currcell is None \333           or not self._currcell['cell_type'] == 'markdown' \334           or force:335            # we need a new cell336            self._store_currcell()337            self._currcell = deepcopy(ReST2IPyNB.markdowncell_template)338    def _add2buffer(self, value, newline=False, paragraph=False):339        if paragraph:340            nl = '\n\n'341        else:342            nl = '\n'343        if newline:344            self._buffer += '%s%s%s' % (nl, ' ' * self._state['indent'], value)345        else:346            self._buffer += value347    def _flush_buffer(self, startnew=True):348        if not len(self._buffer):349            return350        if self._currcell['cell_type'] == 'code':351            target_field = 'input'352        else:353            target_field = 'source'354        if startnew or not len(self._currcell[target_field]):355            self._currcell[target_field].append(self._buffer)356        else:357            self._currcell[target_field][-1] += self._buffer358        self._buffer = ''359def main():360    parser = OptionParser( \361        usage="%prog [options] <filename|directory> [...]", \...

ovsdb_client.py

Source:ovsdb_client.py

...54    # ===============55    # PUBLIC METHODS56    # ===============57    def receive(self):58        self._flush_buffer()59        message = ""60        start = time.time()61        while len(message) == 0 or message.count("}") < message.count("{"):62            message = self.buffer63            duration = time.time() - start64            if duration > self._request_timeout:65                raise OVSDBClient.ReceiveTimeoutError(duration, message)66        return json.loads(message)67    def request(self, dict):68        message = json.dumps(dict)69        self._flush_buffer()70        self._socket.send(message)71        response = ""72        start = time.time()73        while len(response) == 0 or \74                response.count("}") < response.count("{"):75            response = self.buffer76            duration = time.time() - start77            if duration > self._request_timeout:78                raise OVSDBClient.RequestTimeoutError(79                    message, duration, response)80        return json.loads(response)81    def send(self, dict):82        self._flush_buffer()83        self._socket.send(json.dumps(dict))84    # ================85    # PRIVATE METHODS86    # ================87    def _flush_buffer(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.