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

How to use coro_decorator method in hypothesis

Best Python code snippet using hypothesis

decorators.py

Source:decorators.py

1import asyncio2import enum3import functools4import logging5import time6import irc.client7from common.utils import coro_decorator, throttle_base, DEFAULT_THROTTLE8from lrrbot.docstring import encode_docstring, add_header, parse_docstring9from common import twitch10log = logging.getLogger("lrrbot.decorators")11@coro_decorator12def mod_only(func):13	"""Prevent an event-handler function from being called by non-moderators14	Usage:15	@mod_only16	def on_event(self, conn, event, ...):17		...18	"""19	@functools.wraps(func)20	async def wrapper(self, conn, event, *args, **kwargs):21		if self.is_mod(event):22			return await func(self, conn, event, *args, **kwargs)23		else:24			log.info("Refusing %s due to not-a-mod" % func.__name__)25			source = irc.client.NickMask(event.source)26			conn.privmsg(source.nick, "That is a mod-only command.")27			return None28	wrapper.__doc__ = encode_docstring(add_header(parse_docstring(wrapper.__doc__),29		"Mod-Only", "true"))30	return wrapper31@coro_decorator32def sub_only(func):33	"""Prevent an event-handler function from being called by non-subscribers34	Usage:35	@sub_only36	def on_event(self, conn, event, ...):37		...38	"""39	@functools.wraps(func)40	async def wrapper(self, conn, event, *args, **kwargs):41		if self.is_sub(event) or self.is_mod(event):42			return await func(self, conn, event, *args, **kwargs)43		else:44			log.info("Refusing %s due to not-a-sub" % func.__name__)45			source = irc.client.NickMask(event.source)46			conn.privmsg(source.nick, "That is a sub-only command.")47			return None48	wrapper.__doc__ = encode_docstring(add_header(parse_docstring(wrapper.__doc__),49		"Sub-Only", "true"))50	return wrapper51@coro_decorator52def public_only(func):53	"""Prevent an event-handler function from being called via private message54	Usage:55	@public_only56	def on_event(self, conn, event, ...):57		...58	"""59	@functools.wraps(func)60	async def wrapper(self, conn, event, *args, **kwargs):61		if event.type == "pubmsg" or self.is_mod(event):62			return await func(self, conn, event, *args, **kwargs)63		else:64			source = irc.client.NickMask(event.source)65			conn.privmsg(source.nick, "That command cannot be used via private message.")66	wrapper.__doc__ = encode_docstring(add_header(parse_docstring(wrapper.__doc__),67		"Public-Only", "true"))68	return wrapper69class twitch_throttle:70	def __init__(self, count=20, period=30):71		self.count = count72		self.period = period73		self.timestamps = []74	def __call__(self, f):75		@functools.wraps(f, assigned=functools.WRAPPER_ASSIGNMENTS + ("is_logged",))76		def wrapper(*args, **kwargs):77			now = time.time()78			self.timestamps = [t for t in self.timestamps if now-t <= self.period]79			if len(self.timestamps) >= self.count:80				log.info("Ignoring {}(*{}, **{})".format(f.__name__, args, kwargs))81			else:82				self.timestamps.append(now)83				return f(*args, **kwargs)84		wrapper.is_throttled = True85		return wrapper86class Visibility(enum.Enum):87	SILENT = 088	PRIVATE = 189	PUBLIC = 290class throttle(throttle_base):91	"""Prevent an event function from being called more often than once per period92	Usage:93	@throttle([period])94	def func(lrrbot, conn, event, ...):95		...96	count allows the function to be called a given number of times during the period,97	but no more.98	"""99	def __init__(self, period=DEFAULT_THROTTLE, notify=Visibility.PRIVATE, modoverride=True, params=[], log=True, count=1, allowprivate=True):100		super().__init__(period=period, params=params, log=log, count=count)101		self.notify = notify102		self.modoverride = modoverride103		self.allowprivate = allowprivate104	def bypass(self, func, args, kwargs):105		if self.modoverride:106			lrrbot = args[0]107			event = args[2]108			if lrrbot.is_mod(event):109				return True110		if self.allowprivate:111			event = args[2]112			if event.type == "privmsg":113				return True114		return super().bypass(func, args, kwargs)115	def cache_hit(self, func, args, kwargs):116		if self.notify is not Visibility.SILENT:117			conn = args[1]118			event = args[2]119			source = irc.client.NickMask(event.source)120			conn.privmsg(source.nick, "A similar command has been registered recently.")121		return super().cache_hit(func, args, kwargs)122	def decorate(self, func):123		wrapper = super().decorate(func)124		wrapper.__doc__ = encode_docstring(add_header(add_header(parse_docstring(wrapper.__doc__),125			"Throttled", str(self.period)), "Throttle-Count", str(self.count)))126		return wrapper127@coro_decorator128def private_reply_when_live(func):129	"""Cause command handler to respond with private message when the stream is live.130	Usage:131	@private_reply_when_live132	def on_event(self, conn, event, respond_to, ...):133		...134	"""135	@functools.wraps(func)136	async def wrapper(self, conn, event, respond_to, *args, **kwargs):137		if event.type == "pubmsg" and twitch.get_info()["live"]:138			source = irc.client.NickMask(event.source)139			respond_to = source.nick140		return await func(self, conn, event, respond_to, *args, **kwargs)...

test_asyncio.py

Source:test_asyncio.py

1# This file is part of Hypothesis, which may be found at2# https://github.com/HypothesisWorks/hypothesis/3#4# Most of this work is copyright (C) 2013-2020 David R. MacIver5# (david@drmaciver.com), but it contains contributions by others. See6# CONTRIBUTING.rst for a full list of people who may hold copyright, and7# consult the git log if you need to determine who owns an individual8# contribution.9#10# This Source Code Form is subject to the terms of the Mozilla Public License,11# v. 2.0. If a copy of the MPL was not distributed with this file, You can12# obtain one at https://mozilla.org/MPL/2.0/.13#14# END HEADER15import asyncio16import sys17from unittest import TestCase18import pytest19from hypothesis import assume, given, strategies as st20from hypothesis.internal.compat import PYPY21if sys.version_info < (3, 8):22    coro_decorator = asyncio.coroutine23else:24    coro_decorator = pytest.mark.skip25class TestAsyncio(TestCase):26    timeout = 527    def setUp(self):28        self.loop = asyncio.new_event_loop()29        asyncio.set_event_loop(self.loop)30    def tearDown(self):31        self.loop.close()32    def execute_example(self, f):33        error = None34        def g():35            nonlocal error36            try:37                x = f()38                if x is not None:39                    yield from x40            except BaseException as e:41                error = e42        coro = asyncio.coroutine(g)43        future = asyncio.wait_for(coro(), timeout=self.timeout)44        self.loop.run_until_complete(future)45        if error is not None:46            raise error47    @pytest.mark.skipif(PYPY, reason="Error in asyncio.new_event_loop()")48    @given(st.text())49    @coro_decorator50    def test_foo(self, x):51        assume(x)52        yield from asyncio.sleep(0.001)53        assert x54class TestAsyncioRun(TestCase):55    timeout = 556    def execute_example(self, f):57        asyncio.run(f())58    @pytest.mark.skipif(sys.version_info[:2] < (3, 7), reason="asyncio.run() is new")59    @given(st.text())60    @coro_decorator61    def test_foo(self, x):62        assume(x)63        yield from asyncio.sleep(0.001)...

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.