How to use the trio.current_time function in trio

To help you get started, we’ve selected a few trio examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github groove-x / trio-util / tests / test_async_bool.py View on Github external
async def listener(event: AsyncBool):
        assert not event.value  # condition already true
        t0 = trio.current_time()
        await event.wait_value(False, held_for=1)
        assert trio.current_time() - t0 == 1
        test1_done.set()

        assert not event.value  # condition not yet true
        t0 = trio.current_time()
        await event.wait_value(True, held_for=1)
        assert trio.current_time() - t0 == 1.5
        test2_done.set()
github ethereum / trinity / tests-trio / p2p-trio / test_trio_utils.py View on Github external
async def test_every_late(autojump_clock):
    start_time = trio.current_time()

    every_generator = every(2, initial_delay=1)

    first_time = await every_generator.__anext__()
    await trio.sleep(3)

    second_time = await every_generator.__anext__()
    assert second_time == pytest.approx(first_time + 2)
    assert trio.current_time() == pytest.approx(start_time + 1 + 3)

    third_time = await every_generator.__anext__()
    assert third_time == pytest.approx(second_time + 2)
    assert trio.current_time() == pytest.approx(third_time)
github HyperionGray / starbelly / tests / __init__.py View on Github external
def assert_min_elapsed(seconds):
    '''
    Fail the test if the execution of a block takes less than ``seconds``.
    '''
    start = trio.current_time()
    yield
    elapsed = trio.current_time() - start
    assert elapsed >= seconds, 'Completed in under {} seconds'.format(seconds)
github groove-x / trio-util / tests / test_async_value.py View on Github external
async def listener(event: AsyncValue):
        assert event.value == 10  # condition already true
        t0 = trio.current_time()
        assert await event.wait_value(lambda x: x == 10, held_for=1) == 10
        assert trio.current_time() - t0 == 1
        test1_done.set()

        assert event.value < 20  # condition not yet true
        t0 = trio.current_time()
        assert await event.wait_value(lambda x: x >= 20, held_for=1) == 22
        assert trio.current_time() - t0 == 1.5
        test2_done.set()
github HyperionGray / starbelly / starbelly / server / __init__.py View on Github external
async def _handle_request(self, request_data):
        '''
        Handle a single API request.

        :param request: A protobuf request object.
        '''
        start = trio.current_time()
        message = ServerMessage()
        message.response.is_success = False
        request = None

        try:
            # Prepare response.
            request = Request.FromString(request_data)
            message.response.request_id = request.request_id

            # Find an appropriate handler.
            command_name = request.WhichOneof('Command')
            if command_name is None:
                raise InvalidRequestException('No command specified')
            command = getattr(request, command_name)
            try:
                handler = _handlers[command_name]
github ethereum / trinity / p2p / trio_utils.py View on Github external
initial_delay: float = 0,
                ) -> AsyncGenerator[float, Optional[float]]:
    """Generator used to perform a task in regular intervals.

    The generator will attempt to yield at a sequence of target times, defined as
    `start_time + initial_delay + N * interval` seconds where `start_time` is trio's current time
    at instantiation of the generator and `N` starts at `0`. The target time is also the value that
    is yielded.

    If at a certain iteration the target time has already passed, the generator will yield
    immediately (with a checkpoint in between). The yield value is still the target time.

    The generator accepts an optional send value which will delay the next and all future
    iterations of the generator by that amount.
    """
    start_time = trio.current_time()
    undelayed_yield_times = (
        start_time + interval * iteration for iteration in itertools.count()
    )
    delay = initial_delay

    for undelayed_yield_time in undelayed_yield_times:
        yield_time = undelayed_yield_time + delay
        await trio.sleep_until(yield_time)

        additional_delay = yield yield_time
        if additional_delay is not None:
            delay += additional_delay
github ziirish / burp-ui / burpui / engines / monitor.py View on Github external
async def get_mon(self, ident) -> Monitor:
        self.logger.info(f'{ident} - Waiting for a monitor...')
        t1 = trio.current_time()
        mon = await self.pool.get()  # type: Monitor
        t2 = trio.current_time()
        t = t2 - t1
        self.logger.info(f'{ident} - Waited {t:.3f}s')
        yield mon
        self.logger.info(f'{ident} - Releasing monitor')
        await self.pool.put(mon)
github M-o-a-T / qbroker / trio_qbroker / conn.py View on Github external
await chan.basic_qos(prefetch_count=1, prefetch_size=0, connection_global=False)
            logger.debug("Chan %s: read %s", ch, q['queue'])
            await chan.basic_consume(
                queue_name=q['queue'],
                callback=self._on_rpc_in if ep.type == "rpc" else self._on_alert_in
            )

            ep._c_channel = chan
            ep._c_queue = q

        except BaseException:  # pragma: no cover
            del self.rpcs[ep.tag]
            if chan is not None:
                del ep._c_channel
                with trio.open_cancel_scope(shield=True, deadline=trio.current_time() + 1):
                    await chan.close()
            raise
github ziirish / burp-ui / burpui / engines / monitor.py View on Github external
async def get_mon(self, ident) -> Monitor:
        self.logger.info(f'{ident} - Waiting for a monitor...')
        t1 = trio.current_time()
        mon = await self.pool.get()  # type: Monitor
        t2 = trio.current_time()
        t = t2 - t1
        self.logger.info(f'{ident} - Waited {t:.3f}s')
        yield mon
        self.logger.info(f'{ident} - Releasing monitor')
        await self.pool.put(mon)
github HyperionGray / starbelly / starbelly / server / __init__.py View on Github external
args.append(self._scheduler)
                elif var == 'server_db':
                    args.append(self._server_db)
                elif var == 'subscription_manager':
                    args.append(self._subscription_manager)
                elif var == 'stats_tracker':
                    args.append(self._stats_tracker)
                elif var == 'websocket':
                    args.append(self._ws)
                else:
                    raise Exception('Unknown dependency "{}" in handler {}()'
                        .format(var, command_name))

            await handler(*args)
            message.response.is_success = True
            elapsed = trio.current_time() - start
            logger.info('Request OK %s %s %0.3fs', self._client,
                command_name, elapsed)
        except DecodeError:
            # Failure to decode a protobuf message means that the connection
            # is severely damaged; raise to the nursery so we can close the
            # entire connection.
            raise
        except InvalidRequestException as ire:
            error_message = str(ire)
            logger.error('Request ERROR %s %s (%s)', command_name,
                self._client, error_message)
            message.response.error_message = error_message
        except:
            logger.exception('Exception while handling request:\n%r',
                request)
            message.response.error_message = 'A server exception occurred'