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

How to use _send_result_message method in lisa

Best Python code snippet using lisa_python

testsuite.py

Source:testsuite.py

...52    information: Dict[str, Any] = field(default_factory=dict)53    log_file: str = ""54    stacktrace: Optional[str] = None55    def __post_init__(self, *args: Any, **kwargs: Any) -> None:56        self._send_result_message()57        self._timer: Timer58    @property59    def is_queued(self) -> bool:60        return self.status == TestStatus.QUEUED61    @property62    def can_run(self) -> bool:63        return self.status in [TestStatus.QUEUED, TestStatus.ASSIGNED]64    @property65    def is_completed(self) -> bool:66        return _is_completed_status(self.status)67    @property68    def name(self) -> str:69        return self.runtime_data.metadata.name70    @hookspec71    def update_test_result_message(self, message: TestResultMessage) -> None:72        ...73    def handle_exception(74        self, exception: Exception, log: Logger, phase: str = ""75    ) -> None:76        self.stacktrace = traceback.format_exc()77        if phase:78            phase = f"{phase} "79        if isinstance(exception, SkippedException):80            log.info(f"case skipped: {exception}")81            log.debug("case skipped", exc_info=exception)82            # case is skipped dynamically83            self.set_status(84                TestStatus.SKIPPED,85                f"{phase}skipped: {exception}",86            )87        elif isinstance(exception, PassedException):88            log.info(f"case passed with warning: {exception}")89            log.debug("case passed with warning", exc_info=exception)90            # case can be passed with a warning.91            self.set_status(92                TestStatus.PASSED,93                f"{phase}warning: {exception}",94            )95        elif isinstance(exception, BadEnvironmentStateException) or isinstance(96            exception, TcpConnectionException97        ):98            log.error("case failed with environment in bad state", exc_info=exception)99            self.set_status(TestStatus.FAILED, f"{phase}{exception}")100            assert self.environment101            self.environment.status = EnvironmentStatus.Bad102        else:103            if self.runtime_data.ignore_failure:104                log.info(105                    f"case failed and ignored. "106                    f"{exception.__class__.__name__}: {exception}"107                )108                self.set_status(TestStatus.ATTEMPTED, f"{phase}{exception}")109            else:110                log.error("case failed", exc_info=exception)111                self.set_status(112                    TestStatus.FAILED,113                    f"{phase}failed. {exception.__class__.__name__}: {exception}",114                )115    def set_status(116        self, new_status: TestStatus, message: Union[str, List[str]]117    ) -> None:118        if message:119            if isinstance(message, str):120                message = [message]121            if self.message:122                message.insert(0, self.message)123            self.message = "\n".join(message)124        if self.status != new_status:125            self.status = new_status126            if new_status == TestStatus.RUNNING:127                self._timer = create_timer()128            self._send_result_message(self.stacktrace)129    def check_environment(130        self, environment: Environment, save_reason: bool = False131    ) -> bool:132        requirement = self.runtime_data.metadata.requirement133        assert requirement.environment134        check_result = requirement.environment.check(environment.capability)135        if (136            check_result.result137            and requirement.os_type138            and environment.status == EnvironmentStatus.Connected139        ):140            for node in environment.nodes.list():141                # the UT has no OS initialized, skip the check142                if not hasattr(node, "os"):143                    continue144                # use __mro__ to match any super types.145                # for example, Ubuntu satisfies Linux146                node_os_capability = search_space.SetSpace[Type[OperatingSystem]](147                    is_allow_set=True, items=type(node.os).__mro__148                )149                os_result = requirement.os_type.check(node_os_capability)150                # If one of OS mismatches, mark the test case is skipped. It151                # assumes no more env can meet the requirements, instead of152                # checking the rest envs one by one. The reason is this checking153                # is a dynamic checking, and it needs to be checked in each154                # deployed environment. It may cause to deploy a lot of155                # environment for checking. In another hand, the OS should be156                # the same for all environments in the same lisa runner. So it's157                # safe to skip a test case on first os mismatched.158                if not os_result.result:159                    raise SkippedException(f"OS type mismatch: {os_result.reasons}")160        if save_reason:161            if self.check_results:162                self.check_results.merge(check_result)163            else:164                self.check_results = check_result165        return check_result.result166    def _send_result_message(self, stacktrace: Optional[str] = None) -> None:167        if hasattr(self, "_timer"):168            self.elapsed = self._timer.elapsed(False)169        fields = ["status", "elapsed", "id_", "log_file"]170        result_message = TestResultMessage()171        set_filtered_fields(self, result_message, fields=fields)172        metadata_fields = [173            "area",174            "category",175            "tags",176            "description",177            "priority",178            "owner",179        ]180        metadata_information = fields_to_dict(...

bot.py

Source:bot.py

...16def _remove_previous_task(update):17    if update.effective_user.id in TASKS:18        TASKS.pop(update.effective_user.id)19        update.message.reply_text('Your previous query has been cancelled.')20def _send_result_message(update, context):21    def report(direct, with_stops):22        direct = direct[:10]23        with_stops = with_stops[:10]24        if direct or with_stops:25            message = 'Here are the best options I have found:\n\n'26            flights = direct if direct else with_stops27            for f in flights:28                message += str(f) + '\n'29            message += '\nYou can go to https://www.skyscanner.com/ to confirm a reservation if you wish.'30        else:31            message = f"I am sorry, there are no flights from {context.chat_data['query'].origin} to " \32                      f"{context.chat_data['query'].destination} that meet the conditions you have specified. :'("33        update.message.reply_text(message)34        TASKS.pop(update.effective_user.id)35        context.chat_data['query'].results_date = datetime.datetime.now()36        context.chat_data['query'].save()37        logger.info('"%s" has received the result.', update.effective_user.full_name)38    return report39def _validate_date(update):40    try:41        date = update.message.text42        date = date.replace('-', '/')43        date = datetime.datetime.strptime(date, '%d/%m/%Y').date()44    except ValueError:45        update.message.reply_text(46            f'{update.message.text} is not a valid date. '47            'Please be careful and try again...'48        )49        return None50    if date < datetime.datetime.now().date():51        update.message.reply_text(52            'The date you have provided is in the past. '53            'I am not a magician! '54            'Please be careful and try again with a date from the future...'55        )56        return None57    if date > datetime.date.today() + datetime.timedelta(days=365):58        update.message.reply_text(59            'The date you have provided is too far in the future. '60            'I am not a magician! '61            'Please be careful and try again with a date closer than a year from today...'62        )63        return None64    return date65def _validate_integer(update, end_date):66    try:67        n = int(update.message.text)68    except ValueError:69        update.message.reply_text(70            f'{update.message.text} is not a valid number of days. '71            'Please be careful and try again...'72        )73        return None74    if n <= 0:75        update.message.reply_text(76            'You have to specify a positive number of days. '77            'Please be careful and try again...'78        )79        return None80    if end_date + datetime.timedelta(days=n) > datetime.date.today() + datetime.timedelta(days=365):81        update.message.reply_text(82            'Flights that far from today have not been scheduled yet. '83            'Please try again with less days...'84        )85        return None86    return n87def start(update, context):88    logger.info('Received /start from "%s".', update.effective_user.full_name)89    if update.effective_user.id in TASKS:90        update.message.reply_text('Please wait until your previous query finishes.')91        return ConversationHandler.END92    context.chat_data['query'] = model.Query(93        user_id=update.effective_user.id,94        username=update.effective_user.full_name,95        creation_date=datetime.datetime.now()96    )97    update.message.reply_text(98        'Hi! My name is SkyscannerBot. I will help you find a cheap flight. '99        'Send /cancel to stop talking to me.\n\n'100        'What is the origin of your flight?'101    )102    return ORIGIN103def origin(update, context):104    origin_code = get_place(CONFIG['x-rapidapi-keys'][0], update.message.text)105    if not origin_code:106        update.message.reply_text(107            'It seems like you made a mistake spelling it or there are no flights from a city named '108            f'{update.message.text}. '109            'Try again with a different origin...'110        )111        return ORIGIN112    context.chat_data['origin'] = origin_code113    context.chat_data['query'].origin = update.message.text114    update.message.reply_text(115        f'I have heard {update.message.text} is a very nice place! '116        'Where are you going to?'117    )118    return DESTINATION119def destination(update, context):120    destination_code = get_place(CONFIG['x-rapidapi-keys'][0], update.message.text)121    if not destination_code:122        update.message.reply_text(123            'It seems like you made a mistake spelling it or there are no flights from a city named '124            f'{update.message.text}. '125            'Try again with a different destination...'126        )127        return DESTINATION128    if destination_code == context.chat_data['origin']:129        update.message.reply_text(130            f'Your origin and destination cannot be both {update.message.text}! '131            'Try again with a different destination...'132        )133        return DESTINATION134    context.chat_data['destination'] = destination_code135    context.chat_data['query'].destination = update.message.text136    reply_keyboard = [['Round trip', 'One way']]137    update.message.reply_text(138        f'I would like to go to {update.message.text} as well! '139        'Are you planning to do a one way or a round trip?',140        reply_markup=ReplyKeyboardMarkup(reply_keyboard, one_time_keyboard=True)141    )142    return TRIP_TYPE143def trip_type(update, context):144    context.chat_data['trip_type'] = update.message.text145    if update.message.text == 'Round trip':146        context.chat_data['query'].round_trip = True147    update.message.reply_text(148        'We are almost done, I just need a couple more questions. '149        'From which date should I start looking for flights? (DD/MM/YYYY)'150    )151    return START_DATE152def start_date(update, context):153    date = _validate_date(update)154    if not date:155        return START_DATE156    context.chat_data['start_date'] = date157    context.chat_data['query'].start_date = context.chat_data['start_date']158    update.message.reply_text(159        'And when should I stop? (DD/MM/YYYY)'160    )161    return END_DATE162def end_date(update, context):163    date = _validate_date(update)164    if not date:165        return END_DATE166    if date < context.chat_data['start_date']:167        update.message.reply_text(168            f'You must set a date after the one you set before {context.chat_data["start_date"].strftime("%d/%m/%Y")}. '169            'Please try again...'170        )171        return END_DATE172    if (date - context.chat_data['start_date']).days > 60:173        update.message.reply_text(174            'That range of dates for search is too broad. '175            'Please, set a date with less than 60 days of difference with one you set before '176            f'({context.chat_data["start_date"]})...'177        )178        return END_DATE179    context.chat_data['end_date'] = date180    context.chat_data['query'].end_date = context.chat_data['end_date']181    if context.chat_data['trip_type'] == 'One way':182        return finish_conversation(update, context)183    else:184        update.message.reply_text(185            f"What is the minimum number of days you would like to stay in {context.chat_data['query'].destination}?"186        )187        return MIN_DAYS188def min_days(update, context):189    days = _validate_integer(update, context.chat_data['end_date'])190    if not days:191        return MIN_DAYS192    context.chat_data['min_days'] = days193    context.chat_data['query'].min_days = context.chat_data['min_days']194    update.message.reply_text(195        'Just one more question... '196        'What is the maximum number of days you could be there?'197    )198    return MAX_DAYS199def max_days(update, context):200    days = _validate_integer(update, context.chat_data['end_date'])201    if not days:202        return MAX_DAYS203    if days < context.chat_data['min_days']:204        update.message.reply_text(205            'You have to set a maximum number of days greater or equal than the minimum you set before '206            f"({context.chat_data['min_days']}). "207            'Please, set a maximum of days with less than 30 days of difference with the minimum...'208        )209        return MAX_DAYS210    if days - context.chat_data['min_days'] > 30:211        update.message.reply_text(212            'That range of days for search is too broad. '213            'Please, set a maximum of days with less than 30 days of difference with the minimum...'214        )215        return MAX_DAYS216    context.chat_data['max_days'] = days217    context.chat_data['query'].max_days = context.chat_data['max_days']218    return finish_conversation(update, context)219def finish_conversation(update, context):220    context.chat_data['query'].save()221    update.message.reply_text(222        'Alright, I think I have everything I needed. '223        'I will be back as soon as I find the best flights for you...'224    )225    fq = FlightQuery(226        CONFIG,227        context.chat_data,228        _send_result_message(update, context)229    )230    t = threading.Thread(target=fq.execute)231    t.start()232    TASKS[update.effective_user.id] = t233    logger.info('Completed query for "%s".', update.effective_user.full_name)234    return ConversationHandler.END235def error(update, context):236    """Log Errors caused by Updates."""237    logger.warning('Update "%s" caused error "%s"', update, context.error)238    _remove_previous_task(update)239    context.chat_data['query'].cancelled = True240    context.chat_data['query'].save()241def cancel(update, context):242    user = update.message.from_user...

component_template.py

Source:component_template.py

...142        # - if it is now possible that the component has received all input messages143        #   for the current epoch call start_epoch()144        #       await self.start_epoch()145    # an example of method for sending an output message146    async def _send_result_message(self):147        """148        Sends a result message based on the current variable values.149        """150        try:151            # The code is commented because no MyResultMessage class is imported.152            # Replace the class name, the message attributes and the topic name as appropriate.153            # result_message = self._message_generator.get_message(154            #     MyResultMessage,155            #     EpochNumber=self._latest_epoch,156            #     TriggeringMessageIds=self._triggering_message_ids,157            #     MyAttribute1=self._my_variable_1,158            #     MyAttribute2=self._my_variable_2159            # )160            # await self._rabbitmq_client.send_message(...

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.