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

How to use update_object method in autotest

Best Python code snippet using autotest_python

auto_fwd.py

Source:auto_fwd.py

1import telethon2from telethon import TelegramClient3#from telethon.tl.functions.messages.forward_messages import ForwardMessagesRequest4#from telethon.tl.functions.messages import ForwardMessagesRequest5from telethon.tl.types import InputPeerChannel, InputPeerChat6from pprint import pprint7from time import gmtime, strftime8FROM_TO_CHATS = {9	1130120557: 389614156, # kajkam10}11def sprint(string, *args, **kwargs):12	"""Safe Print (handle UnicodeEncodeErrors on some terminals)"""13	try:14		print(string, *args, **kwargs)15	except UnicodeEncodeError:16		string = string.encode('utf-8', errors='ignore')\17					   .decode('ascii', errors='ignore')18		print(string, *args, **kwargs)19class FwdClient(TelegramClient):20	def __init__(self, session_user_id, user_phone, api_id, api_hash, proxy=None):21		super().__init__(session_user_id, api_id, api_hash, proxy)22		if not self.connect():23			print('Initial connection failed. Retrying...')24			if not self.connect():25				print('Could not connect to Telegram servers.')26				return27		# Then, ensure we're authorized and have access28		if not self.is_user_authorized():29			print('First run. Sending code request...')30			self.send_code_request(user_phone)31			self_user = None32			while self_user is None:33				code = input('Enter the code you just received: ')34				try:35					self_user = self.sign_in(user_phone, code)36				# Two-step verification may be enabled37				except SessionPasswordNeededError:38					pw = getpass('Two step verification is enabled. '39								 'Please enter your password: ')40					self_user = self.sign_in(password=pw)41		# now get input peers42		self.get_input_peers()43	def get_input_peers(self):44		global FROM_TO_CHATS45		self.from_to_dict = {}46		TO_FROM = {}47		for k, v in FROM_TO_CHATS.items():48			self.from_to_dict[k] = [None, None]49			if v in TO_FROM:50				TO_FROM[v].append(k)51			else:52				TO_FROM[v] = [k]53		_, entities = self.get_dialogs(0)54		for e in entities:55			# print(e.id, '\n------')56			if e.id in FROM_TO_CHATS:57				self.from_to_dict[e.id][0] = telethon.utils.get_input_peer(e)58			elif e.id in TO_FROM:59				for o_f in TO_FROM[e.id]:60					self.from_to_dict[o_f][1] = telethon.utils.get_input_peer(e)61		pprint(self.from_to_dict)62	def run(self):63		# Listen for updates64		self.add_update_handler(self.update_handler)65		try:66			while True:67				pass68		except KeyboardInterrupt:69			pass70	def check_update(self, update_object):71		if len(update_object.updates) == 0:72			return73		try:74			u_id = update_object.updates[0].message.to_id.channel_id75		except:76			if len(update_object.chats) == 0:77				return78			u_id = update_object.chats[0].id79		if not (u_id in self.from_to_dict):80			return81		from_ipc, to_ipc = self.from_to_dict[u_id]82		return update_object.updates, from_ipc, to_ipc83	def check_msg(self, upd):84		try:85			u_id = upd.chat_id 86			if (u_id in self.from_to_dict):87				upd_arr = [upd]88				upd_arr.extend(self.from_to_dict[u_id])89				return(upd_arr)90		except Exception as e:91			print('u.messed.up', e)92			pass93		return94	def fwd_msgs_from_chan(self, fwd_wat):95		FWD_OBJ, FROM_IPC, TO_IPC = fwd_wat[0], fwd_wat[1], fwd_wat[2]96		idz = [u.message.id for u in FWD_OBJ]97		r_idz = [telethon.helpers.generate_random_long() for _ in range(len(idz))]98		sendresp = ForwardMessagesRequest(from_peer=FROM_IPC,99									 id=idz, 100									 random_id=r_idz,101									 to_peer=TO_IPC)102		self(sendresp)103	def fwd_msgs_from_chat(self, fwd_wat):104		FWD_OBJ, FROM_IPC, TO_IPC = fwd_wat[0], fwd_wat[1], fwd_wat[2]105		iid = FWD_OBJ.id106		r_id = telethon.helpers.generate_random_long()107		sendresp = ForwardMessagesRequest(from_peer=FROM_IPC,108									 id=[iid], 109									 random_id=[r_id],110									 to_peer=TO_IPC)111		self(sendresp)112	def update_handler(self, update_object):113		try:114			self._auaua(update_object)115		except Exception as e:116			print('------------')117			print(e)118			print('------------')119			pprint(update_object)120			pprint(vars(update_object))121	def _auaua(self, update_object):122		tstamp = strftime("%Y-%m-%d %H:%M:%S", gmtime())123		print(tstamp, 'got an update')124		# sprint(update_object)125		if isinstance(update_object, telethon.tl.types.updates_tg.UpdatesTg):126			any_fwd = self.check_update(update_object)127			if any_fwd is not None:128				self.fwd_msgs_from_chan(any_fwd)129		if isinstance(update_object, telethon.tl.types.UpdateShortMessage):130			if update_object.out:131				sprint('You sent {} to user #{}'.format(132					update_object.message, update_object.user_id))133			else:134				sprint('[User #{} sent {}]'.format(135					update_object.user_id, update_object.message))136		elif isinstance(update_object, telethon.tl.types.UpdateShortChatMessage):137			if update_object.out:138				sprint('You sent {} to chat #{}'.format(139					update_object.message, update_object.chat_id))140			else:141				# from group..142				any_fwd = self.check_msg(update_object)143				if any_fwd is not None:144					self.fwd_msgs_from_chat(any_fwd)145				else:146					sprint('[Chat #{}, user #{} sent {}]'.format(147						   update_object.chat_id, update_object.from_id,148						   update_object.message))149		else:150			print(type(update_object))151if __name__ == '__main__':152	api_id = 78862 # change this153	api_hash = '36d610d36b1aa79625169abe74205779'154	phone_no = '+386 31803597'155	fwww = FwdClient('RecniGollum', phone_no, api_id, api_hash)...

test_commands.py

Source:test_commands.py

...3import api.weather.yandex as weather4from telegram.commands import execute_command5@pytest.mark.parametrize('txt', [None, '', 'start', 'ping', 'echo', 'test', 'some text'])6async def test_no_command(update_object, txt):7    send_message = await execute_command(123, update_object(txt))8    assert send_message.text == 'No such command'9@pytest.mark.parametrize('txt', ['/ping', '/ping 123', '/ping\n123', '/ping\t123'])10async def test_ping(update_object, txt):11    send_message = await execute_command(123, update_object(txt))12    assert send_message.text == 'pong!'13@pytest.mark.parametrize('txt', ['/echo', '/echo text', '/echo     text', '/echo\n\ttext\ntest\t'])14async def test_echo(update_object, txt):15    send_message = await execute_command(123, update_object(txt))16    assert send_message.text == f'```\n{update_object(txt).to_str()}\n```'17@pytest.mark.parametrize('txt', ['/start', '/start text', '/start     text', '/start\n\ttext\ntest\t'])18async def test_start(update_object, txt):19    send_message = await execute_command(123, update_object(txt))20    assert send_message.text == '\n'.join(['Hello! I can process such commands:',21                                           '/start - print available commands',22                                           '/echo - print input telegram Update object',23                                           '/ping - check bot availability',24                                           '/weather - returns current weather in SPb'])25@pytest.mark.parametrize('txt', ['/weather', '/weather 123', '/weather\n123', '/weather\t123'])26async def test_weather(txt, yandex_weather_json, async_context_response, update_object, redis_fixture):27    with patch('aiohttp.request') as request_mock, \28            patch('telegram.commands.error_log') as log_mock:29        request_mock.return_value = async_context_response(200, yandex_weather_json)30        first_chars = 13331        result = (await execute_command(123, update_object(txt))).text32        assert result[:first_chars] == weather.YandexWeather.stringify(yandex_weather_json)[:first_chars]33        assert result.startswith('Detailed: https://yandex.ru/pogoda/')34        log_mock.exception.assert_not_called()35        await redis_fixture[1]36async def test_weather_key_error(update_object, redis_fixture):37    with patch('api.weather.yandex.YandexWeather.weather_description') as weather_mock, \38            patch('telegram.commands.error_log') as log_mock:39        weather_mock.side_effect = KeyError('some key error')40        error_result = (await execute_command(123, update_object('/weather'))).text41        assert error_result == 'Sorry. Something went wrong while parsing answer from API.'42        log_mock.exception.assert_called_once()43        await redis_fixture[1]44async def test_weather_bad_answer(update_object, redis_fixture, yandex_weather_json):45    yandex_weather_json.pop('info')46    with patch('api.weather.yandex.YandexWeather.get_weather') as weather_mock, \47            patch('telegram.commands.error_log') as log_mock:48        async def new_weather():49            return yandex_weather_json50        weather_mock.return_value = new_weather()51        error_result = (await execute_command(123, update_object('/weather'))).text52        assert error_result == 'Sorry. Something went wrong while parsing answer from API.'53        log_mock.exception.assert_called_once()54        await redis_fixture[1]55async def test_weather_bad_status(update_object, redis_fixture):56    with patch('api.weather.yandex.YandexWeather.weather_description') as weather_mock, \57            patch('telegram.commands.error_log') as log_mock:58        weather_mock.side_effect = weather.ForbiddenRequestError()59        error_result = (await execute_command(123, update_object('/weather'))).text60        assert error_result.startswith('Sorry. Error occurred during the request to API.')61        log_mock.exception.assert_called_once()...

tracker.py

Source:tracker.py

1class Tracker:2    def __init__(self, _id, url, watcher_id, curr_price, alert_price, lowest_price, highest_price):3        self._id = _id4        self.url = url5        self.watcher_id = watcher_id6        self.curr_price = curr_price7        self.alert_price = alert_price8        self.lowest_price = lowest_price9        self.highest_price = highest_price1011    def should_alert_user(self, price):12        return price <= self.alert_price1314    def has_price_changed(self, new_price):15        return new_price != self.curr_price1617    def create_db_object_for_update(self, new_price):18        update_object = {}1920        if new_price < self.lowest_price:21            update_object['lowestPrice'] = new_price22        23        if new_price > self.highest_price:24            update_object['highestPrice'] = new_price2526        update_object['currentPrice'] = new_price27
...

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.