How to use force_release method in autotest

Best Python code snippet using autotest_python

requests.py

Source:requests.py

1from concurrent import futures2from requests_futures.sessions import FuturesSession3from gremlinclient.connection import Connection, Session4from gremlinclient.graph import GraphDatabase5from gremlinclient.log import pool_logger6from gremlinclient.pool import Pool7from gremlinclient.response import Response8class Response(Response):9    """10    Wrapper for Tornado websocket client connection.11    :param tornado.websocket.WebSocketClientConnection conn: The websocket12        connection13    """14    HEADERS = {'content-type': 'application/json'}15    def __init__(self, url, session, future_class):16        super().__init__(session, future_class)17        self._url = url18        self._closed = False19        self._future = None20    @property21    def conn(self):22        """23        :returns: Underlying connection.24        """25        return self._session26    @property27    def closed(self):28        """29        :returns: Connection protocol. None if conn is closed30        """31        return self._closed32    def close(self):33        self._closed = True34    def send(self, msg):35        """36        Send a message37        :param msg: The message to be sent.38        """39        self._future = self._session.post(self._url, data=msg, self.HEADERS)40    def receive(self, callback=None):41        """42        Read a message off the websocket.43        :param callback: To be called on message read.44        :returns: :py:class:`tornado.concurrent.Future`45        """46        self._closed = True47        return self._future48class Connection(Connection):49    def send(self, gremlin, bindings=None, lang="gremlin-groovy",50               aliases=None, op="eval", processor="", session=None,51               timeout=None, handler=None):52        """53        Send a script to the Gremlin Server.54        :param str gremlin: Gremlin script to submit to server.55        :param dict bindings: A mapping of bindings for Gremlin script.56        :param str lang: Language of scripts submitted to the server.57            "gremlin-groovy" by default58        :param dict aliases: Rebind ``Graph`` and ``TraversalSource``59            objects to different variable names in the current request60        :param str op: Gremlin Server op argument. "eval" by default.61        :param str processor: Gremlin Server processor argument. "" by default.62        :param float timeout: timeout for establishing connection (optional).63            Values ``0`` or ``None`` mean no timeout64        :param str session: Session id (optional). Typically a uuid65        :param loop: If param is ``None``, `tornado.ioloop.IOLoop.current`66            is used for getting default event loop (optional)67        :returns: :py:class:`gremlinclient.connection.Stream` object68        """69        if session is None:70            session = self._session71        if timeout is None:72            timeout = self._timeout73        if aliases is None:74            aliases = {}75        message = json.dumps({76            "gremlin": gremlin,77            "bindings": bindings,78            "language": lang})79        self.conn.send(message)80        return Stream(self,81                      session,82                      processor,83                      handler,84                      self._loop,85                      self._username,86                      self._password,87                      self._force_close,88                      self._force_release,89                      self._future_class)90class GraphDatabase(GraphDatabase):91    def __init__(self, url, timeout=None, username="", password="",92                 loop=None, verify_ssl=False, future_class=None):93        super().__init__(url, timeout=timeout, username=username,94                         password=password, loop=loop,95                         validate_cert=verify_ssl, future_class=futures.Future)96        self._session = FuturesSession()97    def session(self):98        raise NotImplementedError99    def connect(self,100                session=None,101                force_close=False,102                force_release=False,103                pool=None):104        """105        Get a connection to the graph database.106        :param str session: Session id (optional). Typically a uuid107        :param bool force_close: force connection to close after read.108        :param bool force_release: If possible, force release to pool after109            read.110        :param gremlinclient.pool.Pool pool: Associated connection pool.111        :returns: :py:class:`gremlinclient.connection.Connection`112        """113        return self._connect(114            Connection, session, force_close, force_release, pool)115    def _connect(self,116                 conn_type,117                 session,118                 force_close,119                 force_release,120                 pool):121        future = self._future_class()122        resp = Response(url, self._session, self._future_class,123                        loop=self._loop)124        gc = conn_type(resp, self._future_class, self._timeout,125                       self._username, self._password, self._loop,126                       self._validate_cert, False, pool, force_release,127                       session)128        future.set_result(gc)129        return future130    def close(self):131        self._session.close()132class Pool(Pool):133    def __init__(self, url, timeout=None, username="", password="",134                 maxsize=256, loop=None, force_release=False,135                 future_class=None):136        super().__init__(url, timeout=timeout, username=username,137                         password=password, graph_class=GraphDatabase,138                         maxsize=maxsize, loop=loop)139    def close(self):...

graph.py

Source:graph.py

1import socket2import sys3import textwrap4from gremlinclient.connection import Connection, Session5from gremlinclient.response import Response6PY_33 = sys.version_info >= (3, 3)7PY_35 = sys.version_info >= (3, 5)8class GraphDatabase(object):9    """This class generates connections to the Gremlin Server.10    :param str url: url for Gremlin Server.11    :param float timeout: timeout for establishing connection (optional).12        Values ``0`` or ``None`` mean no timeout13    :param str username: Username for SASL auth14    :param str password: Password for SASL auth15    :param loop: If param is ``None``, `tornado.ioloop.IOLoop.current`16        is used for getting default event loop (optional)17    :param bool validate_cert: validate ssl certificate. False by default18    :param class future_class: type of Future -19        :py:class:`asyncio.Future`, :py:class:`trollius.Future`, or20        :py:class:`tornado.concurrent.Future`21    """22    def __init__(self, url, timeout=None, username="",23                 password="", loop=None, validate_cert=False,24                 future_class=None, session_class=Session):25        self._url = url26        self._timeout = timeout27        self._username = username28        self._password = password29        self._loop = loop30        self._validate_cert = validate_cert31        # Hmmm32        self._future_class = future_class33        self._session_class = session_class34    @property35    def future_class(self):36        return self._future_class37    def connect(self,38                session=None,39                force_close=False,40                force_release=False,41                pool=None):42        """43        Get a connection to the graph database.44        :param str session: Session id (optional). Typically a uuid45        :param bool force_close: force connection to close after read.46        :param bool force_release: If possible, force release to pool after47            read.48        :param gremlinclient.pool.Pool pool: Associated connection pool.49        :returns: :py:class:`gremlinclient.connection.Connection`50        """51        return self._connect(52            Connection, session, force_close, force_release, pool)53    def session(self,54                connector=None,55                session=None,56                force_close=False,57                force_release=False,58                pool=None):59        """60        Get a session connection to the graph database.61        :param str session: Session id (optional). Typically a uuid62        :param bool force_close: force connection to close after read.63        :param bool force_release: If possible, force release to pool after64            read.65        :param gremlinclient.pool.Pool pool: Associated connection pool.66        :returns: :py:class:`gremlinclient.connection.Session`67        """68        return self._connect(69            Session, session, force_close, force_release, pool)70    def _connect(self,71                 conn_type,72                 session,73                 force_close,74                 force_release,75                 pool):...

envelope.py

Source:envelope.py

1# Not an enum for performance reasons2class EnvelopeStage:3    DELAY = 04    ATTACK = 15    HOLD = 26    DECAY = 37    SUSTAIN = 48    RELEASE = 59    FINISHED = 610class Envelope:11    def __init__(self, delay, attack, hold, decay, sustain, release):12        self.phases = [delay, attack, hold, decay, sustain, release, -1]13        self.current_phase = EnvelopeStage.DELAY14        self.position = 015        self.start_val = 016        self.current_val = 017        self.target_val = 018        self.total_time = self.phases[0]19        self.force_release = False20    def get_init_vals(self):21        return self.current_phase, self.position, self.start_val, self.current_val, self.target_val, self.total_time22    def update_vals(self, vals):23        if not self.force_release:24            self.current_phase, self.position, self.start_val, self.current_val, self.target_val, self.total_time = vals25        else:26            # Make sure we stay in release phase if it has been activated during a collection.27            _, self.position, self.start_val, self.current_val, self.target_val, self.total_time = vals28            self.force_release = False29    def next_phase(self):30        self.position = 031        self.current_phase += 132        self.total_time = self.phases[self.current_phase]      # Invalid if SUSTAIN or FINISHED33        if self.current_phase == EnvelopeStage.SUSTAIN:34            self.start_val = self.target_val35            self.current_val = self.target_val  # This will be the sustain value36        elif self.current_phase == EnvelopeStage.FINISHED:37            self.start_val = 038            self.current_val = 039            self.target_val = 040        elif self.current_phase == EnvelopeStage.ATTACK:41            self.start_val = 042            self.current_val = 043            self.target_val = 144        elif self.current_phase == EnvelopeStage.HOLD:45            self.start_val = 146            self.current_val = 147            self.target_val = 148        elif self.current_phase == EnvelopeStage.DECAY:49            self.start_val = 150            self.current_val = 151            self.target_val = self.phases[EnvelopeStage.SUSTAIN]52        # EnvelopeStage.RELEASE is managed in `release`53        return self.start_val, self.target_val, self.total_time, self.current_phase54    @property55    def finished(self):56        return self.current_phase == EnvelopeStage.FINISHED57    def release(self):58        self.current_phase = EnvelopeStage.RELEASE59        self.position = 060        self.start_val = self.current_val61        self.target_val = 062        self.total_time = self.phases[self.current_phase]...

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.