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

How to use _connection_lost method in fMBT

Best Python code snippet using fMBT_python

base_protocol.py

Source:base_protocol.py

1import asyncio2from typing import Optional, cast3from .tcp_helpers import tcp_nodelay4class BaseProtocol(asyncio.Protocol):5    __slots__ = (6        "_loop",7        "_paused",8        "_drain_waiter",9        "_connection_lost",10        "_reading_paused",11        "transport",12    )13    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:14        self._loop = loop  # type: asyncio.AbstractEventLoop15        self._paused = False16        self._drain_waiter = None  # type: Optional[asyncio.Future[None]]17        self._connection_lost = False18        self._reading_paused = False19        self.transport = None  # type: Optional[asyncio.Transport]20    def pause_writing(self) -> None:21        assert not self._paused22        self._paused = True23    def resume_writing(self) -> None:24        assert self._paused25        self._paused = False26        waiter = self._drain_waiter27        if waiter is not None:28            self._drain_waiter = None29            if not waiter.done():30                waiter.set_result(None)31    def pause_reading(self) -> None:32        if not self._reading_paused and self.transport is not None:33            try:34                self.transport.pause_reading()35            except (AttributeError, NotImplementedError, RuntimeError):36                pass37            self._reading_paused = True38    def resume_reading(self) -> None:39        if self._reading_paused and self.transport is not None:40            try:41                self.transport.resume_reading()42            except (AttributeError, NotImplementedError, RuntimeError):43                pass44            self._reading_paused = False45    def connection_made(self, transport: asyncio.BaseTransport) -> None:46        tr = cast(asyncio.Transport, transport)47        tcp_nodelay(tr, True)48        self.transport = tr49    def connection_lost(self, exc: Optional[BaseException]) -> None:50        self._connection_lost = True51        # Wake up the writer if currently paused.52        self.transport = None53        if not self._paused:54            return55        waiter = self._drain_waiter56        if waiter is None:57            return58        self._drain_waiter = None59        if waiter.done():60            return61        if exc is None:62            waiter.set_result(None)63        else:64            waiter.set_exception(exc)65    async def _drain_helper(self) -> None:66        if self._connection_lost:67            raise ConnectionResetError("Connection lost")68        if not self._paused:69            return70        waiter = self._drain_waiter71        assert waiter is None or waiter.cancelled()72        waiter = self._loop.create_future()73        self._drain_waiter = waiter...

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.