How to use cannot_encode method in hypothesis

Best Python code snippet using hypothesis

test_from_dtype.py

Source:test_from_dtype.py

...86    # See https://github.com/numpy/numpy/issues/1536387    pass88@pytest.mark.xfail(strict=False, reason="mitigation for issue above")89def test_unicode_string_dtypes_need_not_be_utf8():90    def cannot_encode(string):91        try:92            string.encode("utf-8")93            return False94        except UnicodeEncodeError:95            return True96    find_any(nps.from_dtype(np.dtype("U")), cannot_encode)97@given(st.data())98def test_byte_string_dtypes_generate_unicode_strings(data):99    dt = data.draw(nps.byte_string_dtypes())100    result = data.draw(nps.from_dtype(dt))101    assert isinstance(result, bytes)102@pytest.mark.parametrize("dtype", ["U", "S", "a"])103def test_unsized_strings_length_gt_one(dtype):104    # See https://github.com/HypothesisWorks/hypothesis/issues/2229...

app.py

Source:app.py

1import logging2from io import BytesIO3from aiogram import Bot, Dispatcher, executor, types4from aiogram.utils.exceptions import MessageError, MessageTextIsEmpty5import morse6from config import Config7CANNOT_ENCODE = r"""ÐÐ·Ð²Ð¸Ð½Ð¸ÑÐµ, Ð½Ðµ Ð¿Ð¾Ð»ÑÑÐ¸Ð»Ð¾ÑÑ Ð·Ð°ÐºÐ¾Ð´Ð¸ÑÐ¾Ð²Ð°ÑÑ ÑÑÐ¾ ÑÐ¾Ð¾Ð±ÑÐµÐ½Ð¸Ðµ.8ÐÐ·Ð±ÑÐºÐ° ÐÐ¾ÑÐ·Ðµ ÑÐ¾ÑÑÐ¾Ð¸Ñ ÑÐ¾Ð»ÑÐºÐ¾ Ð¸Ð· Ð±ÑÐºÐ² (Ð°Ð½Ð³Ð»Ð¸Ð¹ÑÐºÐ¸Ñ Ð¸ ÑÑÑÑÐºÐ¸Ñ), ÑÐ¸ÑÑ Ð¸ ÑÐ»ÐµÐ´ÑÑÑÐ¸Ñ Ð·Ð½Ð°ÐºÐ¾Ð² Ð¿ÑÐµÐ¿Ð¸Ð½Ð°Ð½Ð¸Ñ:9`. , : ; ( ) ' " - \ _ ? ! + @`10Ð¡ÐºÐ¾ÑÐµÐµ Ð²ÑÐµÐ³Ð¾, Ð²Ñ ÑÐ¾ÑÐ¸ÑÐµ Ð·Ð°ÐºÐ¾Ð´Ð¸ÑÐ¾Ð²Ð°ÑÑ ÑÐ¾Ð¾Ð±ÑÐµÐ½Ð¸Ðµ, ÑÐ¾ÑÑÐ¾ÑÑÐµÐµ ÑÐ¾Ð»ÑÐºÐ¾ Ð¸Ð· ÑÐ¸Ð¼Ð²Ð¾Ð»Ð¾Ð², ÐºÐ¾ÑÐ¾ÑÑÑ Ð½ÐµÑ Ð² Ð°Ð·Ð±ÑÐºÐµ ÐÐ¾ÑÐ·Ðµ11"""12MSG_TOO_LONG = "ÐÐ·Ð²Ð¸Ð½Ð¸ÑÐµ, ÑÑÐ¾ ÑÐ¾Ð¾Ð±ÑÐµÐ½Ð¸Ðµ Ð¿Ð¾Ð»ÑÑÐ¸Ð»Ð¾ÑÑ ÑÐ»Ð¸ÑÐºÐ¾Ð¼ Ð´Ð»Ð¸Ð½Ð½ÑÐ¼"13config = Config()14logging.basicConfig(15    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s',16    level=config.loglevel,17)18logger = logging.getLogger(__name__)19bot = Bot(token=config.token)20dispatcher = Dispatcher(bot)21def format_error(message: str) -> str:22    return f"ð¤· {message}"23@dispatcher.errors_handler()24async def global_error_handler(update, exception):25    logger.exception(f"Update: {update}\n{exception}")26@dispatcher.message_handler()27async def echo(message: types.Message):28    message_translated = morse.text_to_morse(message.text)29    if not message_translated.strip():30        await message.answer(format_error(CANNOT_ENCODE), parse_mode='markdown')31    elif len(message_translated) > 1024:32        # TODO Split long messages into smaller ones33        await message.answer(format_error(MSG_TOO_LONG))34    else:35        bot_info = await bot.get_me()36        await types.ChatActions.record_audio()37        sound = morse.morse_to_sound(message_translated)38        await types.ChatActions.upload_audio()39        await message.answer_audio(40            audio=BytesIO(sound),41            caption=message_translated.replace("-", "â").replace(".", "Â·"),42            performer=bot_info["username"],43            # TODO filename=f"{performer} - {title}.mp3",44            title=message.text,45        )46async def on_startup(dp):47    await bot.set_webhook(config.webhook_url + config.webhook_path)48async def on_shutdown(dp):49    pass50def main():51    if config.use_webhook:52        logger.debug("Using webhook")53        executor.start_webhook(54            dispatcher=dispatcher,55            webhook_path=config.webhook_path,56            skip_updates=True,57            on_startup=on_startup, on_shutdown=on_shutdown,58            host=config.host, port=config.port59        )60    else:61        logger.debug("Using polling")62        executor.start_polling(dispatcher, skip_updates=True)63if __name__ == "__main__":...

tweet_kentt.py

Source:tweet_kentt.py

1import json2import tweepy3import socket4import sys5def initialize():6    with open("/home/kentt/pems/twitter/config.json") as config_data:7        config = json.load(config_data)8    auth = tweepy.OAuthHandler(config['consumer_key'], config['consumer_secret'])9    auth.set_access_token(config['access_token'], config['access_token_secret'])10    api = tweepy.API(auth)11    stream = TwitterStreamListener()12    twitter_stream = tweepy.Stream(auth = api.auth, listener=stream)13    twitter_stream.filter(track=['python'], async=True)14class TwitterStreamListener(tweepy.StreamListener):15    def on_data(self, data):16        17        data_dict=json.loads(data)18        19        try:20            text = str(data_dict["lang"])+' : '+str(data_dict["text"].decode('utf_8'))21        except:22            text = "Cannot_encode"23        print text24        #except KeyboardInterrupt:25        #    print('Aborting on Ctrl-c, goodbye!')26        #    pass27    #def on_error(self, status_code):28    #    try:29    #        if status_code == 420:30    #            return False31    #    except KeyboardInterrupt:32    #        print('Aborting on Ctrl-c, goodbye!')33    #        pass34        35if __name__ == "__main__":36    initialize()...

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.