How to use _process_environment_message method in lisa

Best Python code snippet using lisa_python

env_stats.py

Source:env_stats.py

...45    def _received_message(self, message: messages.MessageBase) -> None:46        if isinstance(message, TestResultMessage):47            self._process_test_result_message(message)48        elif isinstance(message, EnvironmentMessage):49            self._process_environment_message(message)50        else:51            raise LisaException(f"unsupported message received, {type(message)}")52    def _subscribed_message_type(self) -> List[Type[messages.MessageBase]]:53        return [TestResultMessage, EnvironmentMessage]54    def _initialize(self, *args: Any, **kwargs: Any) -> None:55        env_path = constants.RUN_LOCAL_LOG_PATH / "environments"56        env_path.mkdir(exist_ok=True, parents=True)57        self._file_path = env_path / "environment_stats.log"58        self._last_updated_time = create_timer()59        # result update at most 1 time per second60        self._update_frequency = 161        self._test_results: Dict[str, TestResultInformation] = {}62        self._environments: Dict[str, EnvironmentInformation] = {}63    def _process_test_result_message(self, test_result: TestResultMessage) -> None:64        result_info = self._test_results.get(test_result.id_, None)65        if not result_info:66            result_info = TestResultInformation(67                id=test_result.id_, name=test_result.full_name68            )69            self._test_results[test_result.id_] = result_info70        result_info.status = str(test_result.status)71        env_name = test_result.information.get("environment", "")72        if env_name:73            environment_info = self._environments.get(env_name, None)74            assert (75                environment_info76            ), f"cannot find environment for test result: {test_result}"77            result_info.environment = env_name78            if result_info not in environment_info.results:79                environment_info.results.append(result_info)80        self._update_information()81    def _process_environment_message(self, environment: EnvironmentMessage) -> None:82        env_info = self._environments.get(environment.name, None)83        if not env_info:84            env_info = EnvironmentInformation(85                name=environment.name,86                status=environment.status.name,87                information=str(environment.runbook),88            )89            self._environments[environment.name] = env_info90        env_info.status = environment.status.name91        if environment.status == EnvironmentStatus.Prepared:92            env_info.prepared_time = datetime.now()93        elif environment.status == EnvironmentStatus.Deployed:94            env_info.deployed_time = datetime.now()95        elif environment.status == EnvironmentStatus.Deleted:...

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.