KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use run_func method in localstack
Best Python code snippet using localstack_python
CallTouchBot.py
Source:CallTouchBot.py 
1from aiogram import Bot, Dispatcher, types2from aiogram.utils import executor3from aiogram.contrib.fsm_storage.memory import MemoryStorage4from states import Actions5from aiogram.dispatcher import FSMContext6# ипоÑÑиÑÑем клавиаÑÑÑÑ7from tg_keyboard import main_keyboard, calltouch_keyboard8# импоÑÑиÑÑем внÑÑÑенни ÑÑнкÑии9from calltouch_functions_async import CallTouchFunctions10from db_functions_async import DbFunctions11# импоÑÑ Ð´ÑÑгиÑ
библиоÑек12from datetime import datetime13import itertools14import asyncio15import nest_asyncio16token = '1974255060:AAFosMbvKycoKRydyPdi4yjflv8031SW4mE' # боевой17#token = '1938283222:AAEe7C80RbtpAjW7BVBzt6qISW8VnzIpg0A' # ÑеÑÑовÑй18bot = Bot(token=token)19dp = Dispatcher(bot, storage=MemoryStorage())20nest_asyncio.apply() # Ð´Ð»Ñ Ð´Ð¾Ð±Ð°Ð²Ð»ÐµÐ½Ð¸Ñ Ð·Ð°Ð´Ð°Ñ Ð² аÑÑинÑ
Ñон21# глобалÑнÑе пеÑеменнÑе22ct_creds = dict() # ÑÑда кладем кÑденÑÐ°Ð»Ñ Ð´Ð»Ñ ÑоÑ
ÑÐ°Ð½ÐµÐ½Ð¸Ñ ÐºÐ¾Ð»ÑаÑа23# ÐÑивеÑÑÑвеннÑй блок24@dp.message_handler(commands=['start'], state='*') # пÑивеÑÑÑвÑем и показÑваем клавиаÑÑÑÑ25async def start_message(message: types.Message, state: FSMContext):26 await state.reset_state()27 message_text = 'ÐÑивеÑ!\n' \28 'Я вÑÑÐ»Ñ Ðам ÑÑаÑиÑÑÐ¸ÐºÑ Ð¸Ð· CallTouch в лÑбое вÑемÑ!\n'\29 'ÐÑбеÑеÑе нÑжнÑй пÑÐ½ÐºÑ Ð¼ÐµÐ½Ñ'30 await bot.send_message(message.from_user.id, message_text, reply_markup=main_keyboard)31# блок ÑÐ¾Ð·Ð´Ð°Ð½Ð¸Ñ CT32@dp.callback_query_handler(lambda c: c.data == 'create_creds')33async def create_creds_get_id(callback_query: types.CallbackQuery):34 message_text = 'ÐÑе даннÑе Ð´Ð»Ñ Ð¿Ð¾Ð´ÐºÐ»ÑÑениÑ, можно найÑи в лиÑном кабинеÑе Calltouch:\n'\35 'ÐнÑегÑаÑÐ¸Ñ => API и Webhooks\n' \36 'ЧÑо Ð±Ñ Ð·Ð°Ð²ÐµÑÑи кабинеÑ, вам понадобиÑÑÑ:\n'\37 '1. ID Ñвоего кабинеÑа Calltouch,\n'\38 '2. СÑÑÐ»ÐºÑ Ð½Ð° ÑеÑвеÑ,\n'\39 '3. Токен доÑÑÑпа\n' \40 'Ðли /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð½Ð°Ð·Ð°Ð´\n'41 await bot.answer_callback_query(callback_query.id)42 await bot.send_message(callback_query.from_user.id, message_text)43 await bot.send_message(callback_query.from_user.id, 'ÐведиÑе ID кабинеÑа CallTouch')44 await Actions.create_ct_get_id.set()45# полÑÑение ID кабинеÑа calltouch46@dp.message_handler(state=Actions.create_ct_get_id)47async def get_ct_id(message: types.Message, state: FSMContext):48 # ÑоÑ
ÑанÑем ID введенного кабинеÑа49 async with state.proxy() as data:50 data['cabinet_id'] = message.text51 await Actions.create_ct_get_server.set()52 await bot.send_message(message.from_user.id, 'ÐведиÑе ÑÑÑÐ»ÐºÑ Ð½Ð° ÑеÑвеÑ')53# полÑÑение Ð½Ð¾Ð´Ñ calltouch54@dp.message_handler(state=Actions.create_ct_get_server)55async def get_ct_server(message: types.Message, state: FSMContext):56 # ÑоÑ
ÑанÑем Ð½Ð¾Ð´Ñ Ð²Ð²ÐµÐ´ÐµÐ½Ð½Ð¾Ð³Ð¾ кабинеÑа57 async with state.proxy() as data:58 data['server'] = message.text59 await Actions.create_ct_get_token.set()60 await bot.send_message(message.from_user.id, 'ÐведиÑе Ñокен')61# полÑÑение Ñокена calltouch, валидаÑÐ¸Ñ Ð¸ запиÑÑ Ð´Ð°Ð½Ð½ÑÑ
в ÐÐ62@dp.message_handler(state=Actions.create_ct_get_token)63async def get_ct_token(message: types.Message, state: FSMContext):64 async with state.proxy() as data:65 data['token'] = message.text66 data['create_date'] = datetime.now().strftime('%Y-%m-%d %H:%M')67 data['user_id'] = message.from_user.id68 try:69 async with state.proxy() as data:70 data['user_name'] = message.from_user.first_name71 except ValueError:72 async with state.proxy() as data:73 data['user_name'] = 'unknown'74 try:75 async with state.proxy() as data:76 data['user_surname'] = message.from_user.last_name77 except ValueError:78 async with state.proxy() as data:79 data['user_surname'] = 'unknown'80 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно81 loop = asyncio.get_running_loop()82 run_func = loop.create_task(CallTouchFunctions.validate_creds(data['cabinet_id'], data['server'], data['token']))83 loop.run_until_complete(run_func)84 result = run_func.result()85 if result != 'ok':86 # оÑиÑаем кÑÐµÐ´Ñ ÐºÐ¾Ð»ÑаÑа в ÑлÑÑае оÑибки87 async with state.proxy() as data:88 data.clear()89 message_text = 'ÐведеннÑе даннÑе некоÑÑекÑнÑ.\n'\90 'ÐопÑобÑйÑе еÑе Ñаз'91 await bot.send_message(message.from_user.id, message_text, reply_markup=main_keyboard)92 await state.reset_state()93 else:94 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно95 loop = asyncio.get_running_loop()96 run_func = loop.create_task(DbFunctions.create_new_ct(data))97 loop.run_until_complete(run_func)98 result = run_func.result()99 if result == 'already exists':100 message_text = 'ÐаннÑй ÐºÐ°Ð±Ð¸Ð½ÐµÑ Ñже подклÑÑен'101 await bot.send_message(message.from_user.id, message_text, reply_markup=main_keyboard)102 await state.finish()103 else:104 message_text = 'ÐаннÑе ÑÑпеÑно ÑоÑ
ÑаненÑ!\n'\105 'ТепеÑÑ Ð²Ñ Ð¼Ð¾Ð¶ÐµÑе полÑÑаÑÑ ÑÑаÑиÑÑикÑ'106 await bot.send_message(message.from_user.id, message_text, reply_markup=main_keyboard)107 await state.finish()108# запÑаÑиваем ÑпиÑок аккаÑнÑов CT Ð´Ð»Ñ ÑдалениÑ109@dp.callback_query_handler(lambda c: c.data == 'remove_creds')110async def show_creds_to_remove(callback_query: types.CallbackQuery, state: FSMContext):111 await bot.answer_callback_query(callback_query.id)112 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно113 loop = asyncio.get_running_loop()114 run_func = loop.create_task(DbFunctions.list_of_accounts(callback_query.from_user.id))115 loop.run_until_complete(run_func)116 calltouch_accounts = run_func.result()117 if len(calltouch_accounts) == 0:118 message_text = 'У Ð²Ð°Ñ Ð½ÐµÑ Ð°ÐºÐºÐ°ÑнÑов CallTouch\n' \119 'ÐажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð² менÑ'120 await bot.send_message(callback_query.from_user.id, message_text)121 await state.reset_state()122 else:123 message_text = 'ÐведиÑе ID кабинеÑа коÑоÑÑй необÑ
одимо ÑдалиÑÑ\n' \124 'или нажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð½Ð°Ð·Ð°Ð´\n' \125 'ID ваÑиÑ
аккаÑнÑов:\n'126 for account in calltouch_accounts:127 message_text += account[0] + '\n'128 await bot.send_message(callback_query.from_user.id, message_text)129 await Actions.remove_ct_creds.set()130# ÑдалÑем ÑпиÑок131@dp.message_handler(state=Actions.remove_ct_creds)132async def remove_ct_creds(message: types.Message, state: FSMContext):133 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно134 loop = asyncio.get_running_loop()135 run_func = loop.create_task(DbFunctions.list_of_accounts(message.from_user.id))136 loop.run_until_complete(run_func)137 calltouch_accounts = list(itertools.chain.from_iterable(run_func.result())) # делаем плоÑкий ÑпиÑок138 if message.text not in calltouch_accounts:139 message_text = 'Такого аккаÑнÑа Ð½ÐµÑ Ð² ÑпиÑке.\n'\140 'ÐведиÑе аккаÑÐ½Ñ Ð¸Ð· ÑпиÑка вÑÑе или нажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð½Ð°Ð·Ð°Ð´'141 await bot.send_message(message.from_user.id, message_text)142 else:143 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно144 loop = asyncio.get_running_loop()145 run_func = loop.create_task(DbFunctions.remove_account(message.from_user.id, message.text))146 loop.run_until_complete(run_func)147 message_text = f'ÐккаÑÐ½Ñ {message.text} Ñдален!\n' \148 f'ÐажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð² менÑ'149 await bot.send_message(message.from_user.id, message_text)150 await state.finish()151# вÑводим ÑпиÑок кÑеденÑалов152@dp.callback_query_handler(lambda c: c.data == 'show_creds')153async def show_my_creds(callback_query: types.CallbackQuery):154 await bot.answer_callback_query(callback_query.id)155 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно156 loop = asyncio.get_running_loop()157 run_func = loop.create_task(DbFunctions.list_of_accounts(callback_query.from_user.id))158 loop.run_until_complete(run_func)159 calltouch_accounts = list(itertools.chain.from_iterable(run_func.result()))160 if len(calltouch_accounts) == 0:161 message_text = 'У Ð²Ð°Ñ Ð½ÐµÑ Ð°ÐºÐºÐ°ÑнÑов CallTouch\n' \162 'ÐажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð² менÑ'163 await bot.send_message(callback_query.from_user.id, message_text)164 else:165 message_text = 'ID ваÑиÑ
кабинеÑов:\n'166 for account in calltouch_accounts:167 message_text += account + '\n'168 message_text += 'ÐажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð½Ð°Ð·Ð°Ð´'169 await bot.send_message(callback_query.from_user.id, message_text)170# вÑводим ÑпиÑок кабинеÑов пÑи запÑоÑе ÑÑаÑиÑÑики171@dp.callback_query_handler(lambda c: c.data == 'get_call_stat')172async def show_my_creds(callback_query: types.CallbackQuery, state: FSMContext):173 await bot.answer_callback_query(callback_query.id)174 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно175 loop = asyncio.get_running_loop()176 run_func = loop.create_task(DbFunctions.list_of_accounts(callback_query.from_user.id))177 loop.run_until_complete(run_func)178 calltouch_accounts = list(itertools.chain.from_iterable(run_func.result()))179 if len(calltouch_accounts) == 0:180 message_text = 'У Ð²Ð°Ñ Ð½ÐµÑ Ð°ÐºÐºÐ°ÑнÑов CallTouch\n' \181 'ÐажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð² менÑ'182 await bot.send_message(callback_query.from_user.id, message_text)183 elif len(calltouch_accounts) == 1:184 async with state.proxy() as data:185 data['cabinet_id'] = calltouch_accounts[0]186 message_text = 'ÐÑбеÑеÑе пÑÐ½ÐºÑ Ð¼ÐµÐ½Ñ:'187 await bot.send_message(callback_query.from_user.id, message_text, reply_markup=calltouch_keyboard)188 await Actions.get_statistic.set()189 else:190 message_text = 'ÐведиÑе ID кабинеÑа по коÑоÑÐ¾Ð¼Ñ Ñ
оÑели Ð±Ñ Ð¿Ð¾Ð»ÑÑаÑÑ ÑÑаÑиÑÑикÑ:\n'191 for account in calltouch_accounts:192 message_text += account + '\n'193 message_text += 'Ðли нажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð½Ð°Ð·Ð°Ð´'194 await bot.send_message(callback_query.from_user.id, message_text)195 await Actions.choose_cabinet_id.set()196# вÑÐ±Ð¾Ñ ÐºÐ°Ð±Ð¸Ð½ÐµÑа еÑли иÑ
болÑÑе одного197@dp.message_handler(state=Actions.choose_cabinet_id)198async def choose_cabinet_id(message: types.Message, state: FSMContext):199 cabinet_id = message.text200 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно201 loop = asyncio.get_running_loop()202 run_func = loop.create_task(DbFunctions.validate_cabinet_id(message.from_user.id, cabinet_id))203 loop.run_until_complete(run_func)204 account_list = list(itertools.chain.from_iterable(run_func.result()))205 if cabinet_id not in account_list:206 message_text = 'ÐккаÑнÑов CallTouch Ñ Ñаким ID не найдено.\n' \207 'ÐажмиÑе /start ÑÑо Ð±Ñ Ð²ÐµÑнÑÑÑÑÑ Ð½Ð°Ð·Ð°Ð´'208 await bot.send_message(message.from_user.id, message_text)209 else:210 async with state.proxy() as data:211 data['cabinet_id'] = cabinet_id212 message_text = 'ÐÑбеÑеÑе пÑÐ½ÐºÑ Ð¼ÐµÐ½Ñ:'213 await bot.send_message(message.from_user.id, message_text, reply_markup=calltouch_keyboard)214 await Actions.get_statistic.set()215# СÑаÑиÑÑика за ÑÐµÐ³Ð¾Ð´Ð½Ñ Ð¿Ð¾ вÑбÑÐ°Ð½Ð½Ð¾Ð¼Ñ ÐÐ216@dp.callback_query_handler(lambda c: c.data == 'get_today_stat', state=Actions.get_statistic)217async def get_today_calls(callback_query: types.CallbackQuery, state: FSMContext):218 await bot.answer_callback_query(callback_query.id)219 async with state.proxy() as data:220 data['days_ago'] = 0221 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно222 loop = asyncio.get_running_loop()223 run_func = loop.create_task(224 CallTouchFunctions.get_n_days_ago_data(callback_query.from_user.id, data['cabinet_id'], data['days_ago']))225 loop.run_until_complete(run_func) # оÑибка но поÑÐµÐ¼Ñ - Ñ
ÐµÑ Ð·Ð½Ð°ÐµÑ226 result = run_func.result()227 if result == 'error':228 message_text = f'Ðозникла оÑибка.\nÐопÑобÑйÑе ÑдалиÑÑ ÐºÐ°Ð±Ð¸Ð½ÐµÑ {data["cabinet_id"]} и попÑобÑйÑе Ñнова'229 await bot.send_message(callback_query.from_user.id, message_text, reply_markup=calltouch_keyboard)230 else:231 message_text = f'СÑаÑиÑÑика за {result["date"]}:\n' \232 f'ÐÑего звонков: {result["total_calls"]}\n' \233 f'УникалÑнÑÑ
звонков: {result["unique_calls"]}\n' \234 f'ЦелевÑÑ
звонков: {result["target_calls"]}\n' \235 f'УникалÑно-ÑелевÑÑ
звонков: {result["unique_target_calls"]}'236 await bot.send_message(callback_query.from_user.id, message_text, reply_markup=calltouch_keyboard)237# СÑаÑиÑÑика за вÑеÑа по вÑбÑÐ°Ð½Ð½Ð¾Ð¼Ñ ÐÐ238@dp.callback_query_handler(lambda c: c.data == 'get_yesterday_stat', state=Actions.get_statistic)239async def get_today_calls(callback_query: types.CallbackQuery, state: FSMContext):240 await bot.answer_callback_query(callback_query.id)241 async with state.proxy() as data:242 data['days_ago'] = 1243 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно244 loop = asyncio.get_running_loop()245 run_func = loop.create_task(246 CallTouchFunctions.get_n_days_ago_data(callback_query.from_user.id, data['cabinet_id'], data['days_ago']))247 loop.run_until_complete(run_func) # оÑибка но поÑÐµÐ¼Ñ - Ñ
ÐµÑ Ð·Ð½Ð°ÐµÑ248 result = run_func.result()249 if result == 'error':250 message_text = f'Ðозникла оÑибка.\nÐопÑобÑйÑе ÑдалиÑÑ ÐºÐ°Ð±Ð¸Ð½ÐµÑ {data["cabinet_id"]} и попÑобÑйÑе Ñнова'251 await bot.send_message(callback_query.from_user.id, message_text, reply_markup=calltouch_keyboard)252 else:253 message_text = f'СÑаÑиÑÑика за {result["date"]}:\n' \254 f'ÐÑего звонков: {result["total_calls"]}\n' \255 f'УникалÑнÑÑ
звонков: {result["unique_calls"]}\n' \256 f'ЦелевÑÑ
звонков: {result["target_calls"]}\n' \257 f'УникалÑно-ÑелевÑÑ
звонков: {result["unique_target_calls"]}'258 await bot.send_message(callback_query.from_user.id, message_text, reply_markup=calltouch_keyboard)259# ÑÑнкÑÐ¸Ñ Ð²Ð¾Ð·Ð²ÑаÑа в главное менÑ260@dp.callback_query_handler(lambda c: c.data == 'get_back', state='*')261async def get_back_to_upper_menu(callback_query: types.CallbackQuery, state: FSMContext):262 await bot.answer_callback_query(callback_query.id)263 message_text = 'ÐÑбеÑеÑе нÑжнÑй пÑÐ½ÐºÑ Ð¼ÐµÐ½Ñ:'264 await bot.send_message(callback_query.from_user.id, message_text, reply_markup=main_keyboard)265 await state.reset_state()266# вÑгÑÑзка ÐÐ Ð´Ð»Ñ "ÑвоиÑ
"267@dp.message_handler(commands=['list'], state='*')268async def start_message(message: types.Message, state: FSMContext):269 message_text = 'ФоÑмиÑÑÑ CSV-Ñайл'270 await bot.send_message(message.from_user.id, message_text)271 # добавлÑем задаÑÑ Ð°ÑÑинÑ
Ñонно272 loop = asyncio.get_running_loop()273 run_func = loop.create_task(DbFunctions.download_db(message.from_user.id))274 loop.run_until_complete(run_func)275 result = run_func.result()276 if result != 'ok':277 pass278 else:279 file_name = open('db_file.csv', 'rb')280 await bot.send_document(message.from_user.id, file_name)281executor.start_polling(dp, skip_updates=True)282# збÑ, вÑе ÑеÑенÑко пеÑеведено в аÑÑинÑ
Ñон...
test_stapler.py
Source:test_stapler.py
1import os2import unittest3from unittest import mock4from mglib import stapler5from mglib.conf import settings6BASE_DIR = os.path.dirname(os.path.abspath(__file__))7DATA_DIR = os.path.join(BASE_DIR, "data")8class TestPdfLib(unittest.TestCase):9 def test_ranges_for_reorder(self):10 actual = stapler.cat_ranges_for_reorder(4, [11 {"page_order": 1, "page_num": 4},12 {"page_order": 2, "page_num": 3},13 {"page_order": 3, "page_num": 2},14 {"page_order": 4, "page_num": 1}15 ])16 expected = [4, 3, 2, 1]17 assert expected == actual18 self.assertRaises(ValueError, stapler.cat_ranges_for_reorder, 2, [])19 self.assertRaises(KeyError, stapler.cat_ranges_for_reorder, 2, [20 {"page_order": 3, "page_num": 4},21 {"page_order": 5, "page_num": 6}22 ])23 def test_delete_pages(self):24 input_file = os.path.join(DATA_DIR, "berlin.pdf")25 output_file = os.path.join(DATA_DIR, "berlin2.pdf")26 with mock.patch("mglib.stapler.run") as run_func:27 stapler.delete_pages(input_file, output_file, [1])28 run_func.assert_called()29 run_func.assert_called_with(30 [settings.BINARY_STAPLER, "del", input_file, "1", output_file]31 )32 def test_split_ranges(self):33 page_count = 934 self.assertRaises(35 ValueError,36 stapler.split_ranges,37 9,38 after="a",39 before=False40 )41 self.assertRaises(42 ValueError,43 stapler.split_ranges,44 9, after=False,45 before=True46 )47 actual1, actual2 = stapler.split_ranges(page_count, 1, False)48 expected1 = [1]49 expected2 = [2, 3, 4, 5, 6, 7, 8, 9]50 assert actual1 == expected151 assert actual2 == expected252 actual1, actual2 = stapler.split_ranges(page_count, False, 2)53 expected1 = [1]54 expected2 = [2, 3, 4, 5, 6, 7, 8, 9]55 assert actual1 == expected156 assert actual2 == expected257 actual1, actual2 = stapler.split_ranges(page_count)58 expected1 = list(range(1, page_count + 1))59 expected2 = []60 assert actual1 == expected161 assert actual2 == expected262 def test_reorder_pages(self):63 input_file = os.path.join(DATA_DIR, "berlin.pdf")64 output_file = os.path.join(DATA_DIR, "berlin2.pdf")65 new_order = [66 {'page_num': 2, 'page_order': 1},67 {'page_num': 1, 'page_order': 2},68 ]69 with mock.patch("mglib.stapler.run") as run_func:70 stapler.reorder_pages(input_file, output_file, new_order)71 run_func.assert_called()72 run_func.assert_called_with(73 [74 settings.BINARY_STAPLER,75 "sel",76 input_file,77 "2",78 "1",79 output_file80 ]81 )82 def test_paste_pages_into_existing_doc(self):83 input_file = os.path.join(DATA_DIR, "berlin.pdf")84 output_file = os.path.join(DATA_DIR, "berlin2.pdf")85 datalist = []86 with mock.patch("mglib.stapler.run") as run_func:87 stapler.paste_pages_into_existing_doc(88 input_file, output_file, datalist89 )90 run_func.assert_called()91 run_func.assert_called_with(92 [93 settings.BINARY_STAPLER,94 "sel", "A=" + input_file, "A1", "A2", output_file95 ]96 )97 datalist = [{"src": input_file, "page_nums": "34"}]98 with mock.patch("mglib.stapler.run") as run_func:99 stapler.paste_pages_into_existing_doc(100 input_file,101 output_file,102 datalist,103 1104 )105 run_func.assert_called()106 run_func.assert_called_with(107 [108 settings.BINARY_STAPLER,109 "sel", "A=" + input_file,110 "B=" + input_file, "A1", "B3",111 "B4", "A2", output_file112 ]113 )114 def test_paste_pages(self):115 input_file = os.path.join(DATA_DIR, "berlin.pdf")116 output_file = os.path.join(DATA_DIR, "berlin2.pdf")117 datalist = []118 with mock.patch("mglib.stapler.run") as run_func:119 stapler.paste_pages(input_file, output_file, datalist, False)120 run_func.assert_called()121 run_func.assert_called_with(122 [123 settings.BINARY_STAPLER,124 "sel",125 "A=" + input_file,126 "A1",127 "A2",128 output_file129 ]130 )131 datalist = [{"src": input_file, "page_nums": "34"}]132 with mock.patch("mglib.stapler.run") as run_func:133 stapler.paste_pages(input_file, output_file, datalist)134 run_func.assert_called()135 run_func.assert_called_with(136 [settings.BINARY_STAPLER, "sel", "A=" + input_file, "A3", "A4",137 output_file]...
cython_main.py
Source:cython_main.py
...5import click6import inspect7import numpy as np8import cython_examples as cyth9def run_func(func, *args, **kwargs):10 """Helper function for running examples"""11 ray.init()12 func = ray.remote(func)13 # NOTE: kwargs not allowed for now14 result = ray.get(func.remote(*args))15 # Inspect the stack to get calling example16 caller = inspect.stack()[1][3]17 print("%s: %s" % (caller, str(result)))18 return result19@click.group(context_settings={'help_option_names': ['-h', '--help']})20def cli():21 """Working with Cython actors and functions in Ray"""22@cli.command()23def example1():24 """Cython def function"""25 run_func(cyth.simple_func, 1, 2, 3)26@cli.command()27def example2():28 """Cython def function, recursive"""29 run_func(cyth.fib, 10)30@cli.command()31def example3():32 """Cython def function, built-in typed parameter"""33 # NOTE: Cython will attempt to cast argument to correct type34 # NOTE: Floats will be cast to int, but string, for example will error35 run_func(cyth.fib_int, 10)36@cli.command()37def example4():38 """Cython cpdef function"""39 run_func(cyth.fib_cpdef, 10)40@cli.command()41def example5():42 """Cython wrapped cdef function"""43 # NOTE: cdef functions are not exposed to Python44 run_func(cyth.fib_cdef, 10)45@cli.command()46def example6():47 """Cython simple class"""48 ray.init()49 cls = ray.remote(cyth.simple_class)50 a1 = cls.remote()51 a2 = cls.remote()52 result1 = ray.get(a1.increment.remote())53 result2 = ray.get(a2.increment.remote())54 print(result1, result2)55@cli.command()56def example7():57 """Cython with function from BrainIAK (masked log)"""58 run_func(cyth.masked_log, np.array([-1.0, 0.0, 1.0, 2.0]))59@cli.command()60def example8():61 """Cython with blas. NOTE: requires scipy"""62 # See cython_blas.pyx for argument documentation63 mat = np.array([[[2.0, 2.0], [2.0, 2.0]], [[2.0, 2.0], [2.0, 2.0]]],64 dtype=np.float32)65 result = np.zeros((2, 2), np.float32, order="C")66 run_func(cyth.compute_kernel_matrix,67 "L",68 "T",69 2,70 2,71 1.0,72 mat,73 0,74 2,75 1.0,76 result,77 278 )79if __name__ == "__main__":80 cli()
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
