Best Python code snippet using Testify_python
inotify_file_watcher.py
Source:inotify_file_watcher.py
...108 self._directory_to_watch_descriptor[directory_path] = watch_descriptor109 self._directory_to_subdirs[directory_path] = set()110 def start(self):111 """Start watching the directory for changes."""112 self._class_setup()113 self._inotify_fd = InotifyFileWatcher._libc.inotify_init()114 if self._inotify_fd < 0:115 error = OSError('failed call to inotify_init')116 error.errno = ctypes.get_errno()117 error.strerror = errno.errorcode[ctypes.get_errno()]118 raise error119 self._inotify_poll = select.poll()120 self._inotify_poll.register(self._inotify_fd, select.POLLIN)121 self._add_watch_for_path(self._directory)122 def quit(self):123 """Stop watching the directory for changes."""124 os.close(self._inotify_fd)125 def _get_changed_paths(self):126 """Return paths for changed files and directories.127 start() must be called before this method.128 Returns:129 A set of strings representing file and directory paths that have changed130 since the last call to get_changed_paths.131 """132 paths = set()133 while True:134 if not self._inotify_poll.poll(0):135 break136 self._inotify_events += os.read(self._inotify_fd, 1024)137 while len(self._inotify_events) > _INOTIFY_EVENT_SIZE:138 wd, mask, cookie, length = _INOTIFY_EVENT.unpack(139 self._inotify_events[:_INOTIFY_EVENT_SIZE])140 if len(self._inotify_events) < _INOTIFY_EVENT_SIZE + length:141 break142 name = self._inotify_events[143 _INOTIFY_EVENT_SIZE:_INOTIFY_EVENT_SIZE+length]144 name = name.rstrip('\0')145 logging.debug('wd=%s, mask=%s, cookie=%s, length=%s, name=%r',146 wd, hex(mask), cookie, length, name)147 self._inotify_events = self._inotify_events[_INOTIFY_EVENT_SIZE+length:]148 if mask & IN_IGNORED:149 continue150 try:151 directory = self._watch_to_directory[wd]152 except KeyError:153 logging.debug('Watch deleted for watch descriptor=%d', wd)154 continue155 path = os.path.join(directory, name)156 if os.path.isdir(path) or path in self._directory_to_watch_descriptor:157 if mask & IN_DELETE:158 self._remove_watch_for_path(path)159 elif mask & IN_MOVED_FROM:160 self._remove_watch_for_path(path)161 elif mask & IN_CREATE:162 self._add_watch_for_path(path)163 elif mask & IN_MOVED_TO:164 self._add_watch_for_path(path)165 if path not in paths:166 paths.add(path)167 return paths168 def has_changes(self):169 return bool(self._get_changed_paths())170 @classmethod171 def _class_setup(cls):172 if cls._libc:173 return174 libc_name = ctypes.util.find_library('c')175 cls._libc = ctypes.CDLL(libc_name, use_errno=True)176 cls._libc.inotify_init.argtypes = []177 cls._libc.inotify_init.restype = ctypes.c_int178 cls._libc.inotify_add_watch.argtypes = [ctypes.c_int,179 ctypes.c_char_p,180 ctypes.c_uint32]181 cls._libc.inotify_add_watch.restype = ctypes.c_int182 cls._libc.inotify_rm_watch.argtypes = [ctypes.c_int,183 ctypes.c_int]...
conversation.py
Source:conversation.py
...8 self._setup, self._setup_callback = setup_methods9 self.ctx = None10 self.bot = None11 self.user = None12 async def _class_setup(self, user_id: int):13 self.ctx = self._setup()14 self.bot, self.user = self._setup_callback(user_id)15 async def back_to_welcome(self, callback_query: CallbackQuery):16 await self._class_setup(callback_query.from_user.id)17 await self.bot.answer_callback_query(callback_query.id)18 context = self.dialog.welcome(self.user.locale)19 message = context.format_lazy(self.user.username)20 markup = context.get_markup()21 await self.bot.edit_message_text(22 message_id=callback_query.message.message_id,23 chat_id=callback_query.message.chat.id,24 text=message,25 reply_markup=markup,26 )27class DocsScenario(BaseScenario):28 async def get_docs_handler(self, callback_query: CallbackQuery):29 await self._class_setup(callback_query.from_user.id)30 await self.bot.answer_callback_query(callback_query.id)31 with open(f"{ABS_PATH}/agbot/Resources/docs.jpg", "rb") as photo:32 await self.bot.send_photo(33 chat_id=callback_query.message.chat.id,34 photo=photo,35 )36class AETResultsScenario(BaseScenario):37 async def get_aet_resutls_handler(self, callback_query: CallbackQuery):38 await self._class_setup(callback_query.from_user.id)39 await self.bot.answer_callback_query(callback_query.id)40 aet_links = self.ctx["orm"].aet.get_results_as_string()41 context = self.dialog.aet_results()42 message = context.format_lazy(aet_links)43 markup = context.get_markup()44 await self.bot.edit_message_text(45 message_id=callback_query.message.message_id,46 chat_id=callback_query.message.chat.id,47 text=message,48 reply_markup=markup,49 )50class ImportantDatesScenario(BaseScenario):51 async def get_important_dates_handler(self, callback_query: CallbackQuery):52 await self._class_setup(callback_query.from_user.id)53 await self.bot.answer_callback_query(callback_query.id)54 important_dates = self.ctx["orm"].dates.get()55 dates = (56 "Something went wrong. Try this feature later."57 if important_dates is None58 else important_dates.descr59 )60 context = self.dialog.job_image()61 message = context.format_lazy(dates)62 markup = context.get_markup()63 await self.bot.edit_message_text(64 message_id=callback_query.message.message_id,65 chat_id=callback_query.message.chat.id,66 text=message,67 reply_markup=markup,68 )69class JobsImageScenario(BaseScenario):70 async def get_jobs_image_handler(self, callback_query: CallbackQuery):71 await self._class_setup(callback_query.from_user.id)72 await self.bot.answer_callback_query(callback_query.id)73 img_href, _ = get_jobs_image(static=True)74 if img_href is None:75 img_href = "Something went wrong. Please try this funciton later."76 context = self.dialog.job_image()77 message = context.format_lazy(img_href)78 markup = context.get_markup()79 await self.bot.edit_message_text(80 message_id=callback_query.message.message_id,81 chat_id=callback_query.message.chat.id,82 text=message,83 reply_markup=markup,84 )85class ScolarScenario(BaseScenario):86 async def send_scolar_instr(self, callback_query: CallbackQuery):87 await self._class_setup(callback_query.from_user.id)88 await self.bot.answer_callback_query(callback_query.id)89 context = self.dialog.scolar()90 message = context.format_lazy()91 markup = context.get_markup()92 await self.bot.edit_message_text(93 message_id=callback_query.message.message_id,94 chat_id=callback_query.message.chat.id,95 text=message,96 reply_markup=markup,97 )98class ENTScenario(BaseScenario):99 async def send_past_year_ent_results(self, callback_query: CallbackQuery):100 await self._class_setup(callback_query.from_user.id)101 await self.bot.answer_callback_query(callback_query.id)102 data = self.ctx["orm"].ent.match_by_categories()103 context = self.dialog.ent()104 message = context.format_lazy(datetime.now().year - 1)105 markup = context.get_markup()106 _string = "\n\n"107 prev_category = None108 for item in data:109 if prev_category is None:110 _string += f"{item.category} \n"111 elif prev_category != item.category:112 _string += f"\n{item.category} \n"113 prev_category = item.category114 _string += f"{item.faculty}: {item.score}\n"...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!