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

How to use _close_stream method in lisa

Best Python code snippet using lisa_python

console_logger.py

Source:console_logger.py

...74                try:75                    data = stream.recv(libvirt.virStorageVol.streamBufSize)76                except libvirt.libvirtError:77                    # An error occured. So, close the stream.78                    self._close_stream(True)79                    break80                if data == -2:81                    # No more data available at the moment.82                    assert self._log_file83                    self._log_file.flush()84                    break85                if len(data) == 0:86                    # EOF reached.87                    self._close_stream(False)88                    break89                assert self._log_file90                self._log_file.write(data)91        if (92            events & libvirt.VIR_STREAM_EVENT_ERROR93            or events & libvirt.VIR_STREAM_EVENT_HANGUP94        ):95            # Stream is shutting down. So, close it.96            self._close_stream(True)97    # Close the stream resource.98    # Threading: Must only be called on libvirt events thread.99    def _close_stream(self, abort: bool) -> None:100        if self._stream_completed.is_set():101            # Already closed. Nothing to do.102            return103        try:104            # Close the log file105            assert self._log_file106            self._log_file.close()107            # Close the stream108            assert self._console_stream109            if self._console_stream_callback_added:110                self._console_stream.eventRemoveCallback()111            if abort:112                self._console_stream.abort()113            else:...

io.py

Source:io.py

...37            self.stream.write(np.ascontiguousarray(val, dtype=np.float32))38        except sd.PortAudioError:39            pass40    def on_error(self, err):41        self._close_stream()42    def on_completed(self):43        self._close_stream()44    def _close_stream(self):45        self.stream.close()46    def start(self):47        self.stream.start()48    def stop(self):49        self.stream.stop()50class WavFileOutput(rx.Observer):51    def __init__(self, filename: str, sample_rate: int = 8000):52        super().__init__()53        self.filename = filename54        self.cache = None55        self.sample_rate = sample_rate56    def on_next(self, value):57        if self.cache is None:58            self.cache = value...

base_parser.py

Source:base_parser.py

1import gzip2class AbstractParser(object):3    def __init__(self, stream, delimiter="\t"):4        self._close_stream = False5        if isinstance(stream, str):6            if stream.endswith(".gz"):7                self._stream = gzip.open(stream, mode="rt")8            else:9                self._stream = open(stream, "r")10            self._close_stream = True11        else:12            self._stream = stream13        self._delimiter = delimiter14    def __del__(self):15        if self._close_stream:16            self._stream.close()17    def parse_line(self, line):18        return list(map(str.strip, line.split(self.delimiter)))19    def __iter__(self):20        self._iter = iter(self._stream)21        return self22    def __next__(self):23        return self.next()24    def next(self):25        line = self._iter.__next__()26        while self.comment and line.startswith(self.comment):27            line = self._iter.__next__()...

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.