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

How to use _send_message_to_server method in Playwright Python

Best Python code snippet using playwright-python

Michael.py

Source:Michael.py

...90            self._send_telegram_message(SuperUser, "The bot {}, tried to connect with me. I blocked".format(username))91    def do_ping(self, update: Update, context: CallbackContext):92        """Pings the TCP server and waits for a response. Any response from the serves counts as success."""93        ID = update.effective_user.id94        response = self._send_message_to_server("PING", ID)95        if response:96            update.message.reply_text("Server answered and is ready")97        else:98            update.message.reply_text("Server seems to be offline. A reboot can solve this problem.")99    def do_help(self, update: Update, context: CallbackContext):100        """Generates the help text. It takes the doc strings of all possible telegram commands and replies them."""101        text = ""102        for com in self._get_all_telegram_commands():103            text += "{} - {} \n\n".format(com[0].strip(), com[1].strip())104        update.message.reply_text(text)105    def do_info(self, update: Update, context: CallbackContext):106        """Gathers statistics information about the system and writes it on telegram"""107        uptime = "Uptime: {} h\n".format(round(getuptime()/3600,2))108        temp = "CPU Temperature: {} C\n".format(getCPUtemperature())109        CPU = "Uptime: {} %\n".format(getCPUuse())110        RAM = "RAM usage: {} \n".format(getRAMinfo())111        DISK = "Diskspace: {} \n".format(getDiskSpace())112        update.message.reply_text("Warning: This statistics apply to UNIX machines only!\n\n {} {} {} {} {}"113                                             "".format(uptime, temp, CPU, RAM, DISK))114    def do_getIP(self, update: Update, context: CallbackContext):115        """Gets the IP of the machine and sends it to the user."""116        try:117            IP = get_ip()118        except Exception as err:119            IP = "ERROR: Could not obtain IP. ERRORCODE: {}".format(err)120            self.log.error("ERROR: Could not obtain IP. ERRORCODE: {}".format(err))121        finally:122            update.message.reply_text("The IP is {}:".format(IP))123    def do_reboot(self, update: Update, context: CallbackContext):124        """125        Ask if the system should be rebooted or not.126        """127        ID = update.effective_user.id128        is_bot = update.effective_user.is_bot129        username = update.effective_user.name130        SuperUser = self.SuperUser131        if not is_bot:132            keyboard = [InlineKeyboardButton(text='Reboot Now',133                                             callback_data='{"name": "reboot_callback", "value": "yes"}'),134                        InlineKeyboardButton(text='Abort',135                                             callback_data='{"name": "reboot_callback", "value": "no"}')]136            keyboard = InlineKeyboardMarkup([keyboard])137            text = "Do you want to reboot the system?"138            self._send_telegram_message(ID, text, reply_markup=keyboard)139        else:140            self._send_telegram_message(SuperUser, "The bot {}, tried to reboot me. I blocked".format(username))141    def do_start(self, update: Update, context: CallbackContext):142        """143        Reports to the owner if someone is connecting with the bot.144        """145        # If its the start message, just report this to the admin146        ID = update.effective_user.id147        name = update.effective_user.full_name148        is_bot = update.effective_user.is_bot149        lastname = update.effective_user.last_name150        username = update.effective_user.username151        SuperUser = self.SuperUser152        self._send_telegram_message(SuperUser, "A new user has started a conversation with me: \n\n" \153                                             "ID: {} \n" \154                                             "User Name: {} \n" \155                                             "First Name: {} \n" \156                                             "Last Name: {} \n " \157                                             "is Bot: {}".format(ID, username, name, lastname, is_bot))158    # Bot callback functions############################################################################################159    def reboot_callback(self, update, context, value):160        """Restarts the respberry if necessary"""161        if value.lower() == 'no':162            update.message.reply_text("Aborted reboot")163        else:164            update.message.reply_text("Reboot initiated")165            try:166                os.system("sudo /sbin/reboot -n&")167            except:168                update.message.reply_text("Reboot failed. Only works on LINUX machines")169    def newuser_callback(self, update, context, value):170            """171            Does or does not add the user.172            chat_id: The one sending the accept or decline response173            value: the value the admit set for accepting/not accepting174            query: the initial query object from the user asked for entry175            """176            chat_id = update.effective_user.id177            newID = int(value.split("/ID:")[1]) # The user requesting178            value = value.split("/ID:")[0]179            self.newUserrequests.pop(newID)180            if value.lower() == 'no':181                self._send_telegram_message(chat_id, "User {} not added to the family".format(newID))182                self._send_telegram_message(newID, "Your request has been declined by the admin")183            else:184                self.config["Users"].append(newID)185                self.log.info("Try adding user to config file and write to file...")186                try:187                    with open(self.config_path, 'w') as outfile:188                        yaml.dump(self.config, outfile, default_flow_style=False)189                except Exception as err:190                    self.log.warning("Could not save config to file, user not permanently added. Error: {}".format(err))191                    self._send_telegram_message(chat_id, "Could not save config to file, user not permanently added")192                self._send_telegram_message(chat_id, "User {} added to the family".format(newID))193                self._send_telegram_message(newID, "Welcome to the family. Your request has been approved by an admin.")194    def report_back_callback(self, query_id, chat_id, value, query):195        """Reports the custom keyboard markup response to the client"""196        response = self._send_message_to_underlings(value, chat_id)197        if response:198            return self._extract_result(response)199    # Telegram Bot handlers ############################################################################################200    def handle_text(self, update: Update, context: CallbackContext):201        """This function simply handles all text based message, entry point for messages comming from telegram"""202        ID = update.effective_user.id203        message = update.message.text204        if self.check_user_ID(update.effective_user):205            self.log.info("Text message arrived with content: '{}' from user {}".format(message.strip(), ID))206            response = self._send_message_to_server(message)207            if not response:208                self.log.error("Server did not answer")209                response = {"result": "The server seems to be offline..."}210            x = Thread(target=self._process_message, args=(self._extract_result(response),ID))211            x.start()212        # If any other message is send and the user is not valid213        else: # If the ID is not recognised214            if ID not in self.blocked_user_ID:215                self.blocked_user_ID.append(ID)216                self.report_to_owner(update, update.message.text)217    def handle_callback(self, update: Update, context: CallbackContext):218        """Handles callbacks from telegram"""219        ID = update.effective_user.id # The ID from the person making the response220        is_bot = update.effective_user.is_bot # Check if it is not a bot221        if is_bot:222            self.report_to_owner(update, update.callback_query.data)223        elif self.check_user_ID(update.effective_user):224            update.callback_query.answer()  # Answer the call so that everything is in order for the other party225            # Extract data from callback226            cb_info = json.loads(update.callback_query.data)227            func_name = cb_info['name']228            try:229                func = getattr(self, func_name)230                # Funktion aufrufen231                response = func(update, context, cb_info['value'])232            except Exception as err:233                self.log.error("Could not run callback function {} with error: {}".format(func_name, err))234                return235            if response:236                self._process_message(response, ID)237        else:238            self.log.critical("Unauthorized person tried to make a callback. ID: {}".format(ID))239            self.report_to_owner(update, update.callback_query.data)240    # Internal Server Handler ##########################################################################################241    def handle_server_requests(self, action, value):242        """Handles all request which came from a client"""243        self.log.info("Got server message {}: {}".format(action, value))244        if action == "TelegramBot": # Only accept request for the telegram bot245            # Each value must contain as key the ID to whom I should send something246            if isinstance(value, dict):247                self._process_message({"ID": value}, ID=None) # Never do that with the ID!!! Only if you know what you are doing!!!248            else:249                self.log.critical("Client request was not a dictionary. Type: {}".format(type(value)))250                return "Request value must be a dictionary with keys beeing the ID to send to."251        else:252            self.log.critical("Got a message which was not for me! {}: {}".format(action, value))253            return "Wrong message action header for TelegramBot"254    # Private functions ################################################################################################255    def _add_telegram_commands(self):256        """257        Adds all functions starting with 'do_' as handlers to the telegram bot.258        """259        for names in self._get_all_telegram_commands():260            self.log.info("Adding telegram command {} as handler.".format(names))261            try:262                self.dispatcher.add_handler(CommandHandler(names[0], getattr(self, "do_"+names[0])))263            except Exception as err:264                self.log.error("Could not add telegram command handler {} with error {}".format(names, err))265    def _get_all_telegram_commands(self):266        """267        Finds all do_* members the framework and returns them without the do. Furthermore the doc string of all found268        commands will be returned as well. As return value tuples with (name, docstring) in as list will be returned.269        """270        names = []271        for poss in dir(self):272            if "do_" in poss[:3]:273                try:274                    names.append((poss[3:], getattr(self, poss).__doc__))275                except Exception as err:276                    self.log.error("Could not obtain docstring from function {} with error: {}".format(poss, err))277        return names278    def _load_config(self):279        """Loads the config file either from args or passed config file. Args are more important!"""280        if self.args.config:281            self.config = load_yaml(self.args.config)282            self.config_path = self.args.config283        elif self.config_path:284            self.config = load_yaml(self.config_path)285        else:286            self.log.error("{} cannot work without his cheat sheet. Please add a config file and retry.".format(self.name))287    def _init_connection_to_Bot(self):288        """Ta= N the information (token) from the config file and connects to the bot"""289        self.updater = Updater(self.config["token"])290        self.bot = self.updater.bot291        self.dispatcher = self.updater.dispatcher292    def _send_telegram_message(self, ID, msg, **kwargs):293        """Sends a telegram message to the father"""294        self.log.info("Sending message '{}' to ID: {}".format(msg, ID))295        if ID:296            self.bot.send_message(chat_id=ID, text=msg, **kwargs)297        else:298            self.log.warning("No ID passed, no message sent!")299    def _send_message_to_server(self, message, ID=None):300        """Sends a message via tcp to the server.301        :param ID: Sends the ID of the requester as well.302        """303        if isinstance(message, dict):304            msg = message["text"]305            from_ID = message["from"]["id"]306        else:307            msg = str(message)308            from_ID = ID309        try:310            response = self.Client.send_request("TelegramBot", {str(from_ID):"{}".format(msg)})311        except Exception as err:312            self.log.info("Server Error {}".format(err))313            return err...

_connection.py

Source:_connection.py

...34        self, method: str, params: Optional[Dict], return_as_dict: bool35    ) -> Any:36        if params is None:37            params = {}38        callback = self._connection._send_message_to_server(self._guid, method, params)39        result = await callback.future40        # Protocol now has named return values, assume result is one level deeper unless41        # there is explicit ambiguity.42        if not result:43            return None44        assert isinstance(result, dict)45        if return_as_dict:46            return result47        if len(result) == 0:48            return None49        assert len(result) == 150        key = next(iter(result))51        return result[key]52    def send_no_reply(self, method: str, params: Dict = None) -> None:53        if params is None:54            params = {}55        self._connection._send_message_to_server(self._guid, method, params)56class ChannelOwner(AsyncIOEventEmitter):57    def __init__(58        self,59        parent: Union["ChannelOwner", "Connection"],60        type: str,61        guid: str,62        initializer: Dict,63    ) -> None:64        super().__init__(loop=parent._loop)65        self._loop: asyncio.AbstractEventLoop = parent._loop66        self._dispatcher_fiber: Any = parent._dispatcher_fiber67        self._type = type68        self._guid = guid69        self._connection: Connection = (70            parent._connection if isinstance(parent, ChannelOwner) else parent71        )72        self._parent: Optional[ChannelOwner] = (73            parent if isinstance(parent, ChannelOwner) else None74        )75        self._objects: Dict[str, "ChannelOwner"] = {}76        self._channel = Channel(self._connection, guid)77        self._channel._object = self78        self._initializer = initializer79        self._connection._objects[guid] = self80        if self._parent:81            self._parent._objects[guid] = self82    def _dispose(self) -> None:83        # Clean up from parent and connection.84        if self._parent:85            del self._parent._objects[self._guid]86        del self._connection._objects[self._guid]87        # Dispose all children.88        for object in list(self._objects.values()):89            object._dispose()90        self._objects.clear()91class ProtocolCallback:92    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:93        self.stack_trace = "".join(traceback.format_stack()[-10:])94        self.future = loop.create_future()95class RootChannelOwner(ChannelOwner):96    def __init__(self, connection: "Connection") -> None:97        super().__init__(connection, "", "", {})98class Connection:99    def __init__(100        self, dispatcher_fiber: Any, object_factory: Any, driver_executable: Path101    ) -> None:102        self._dispatcher_fiber: Any = dispatcher_fiber103        self._transport = Transport(driver_executable)104        self._transport.on_message = lambda msg: self._dispatch(msg)105        self._waiting_for_object: Dict[str, Any] = {}106        self._last_id = 0107        self._objects: Dict[str, ChannelOwner] = {}108        self._callbacks: Dict[int, ProtocolCallback] = {}109        self._object_factory = object_factory110        self._is_sync = False111    async def run_as_sync(self) -> None:112        self._is_sync = True113        await self.run()114    async def run(self) -> None:115        self._loop = asyncio.get_running_loop()116        self._root_object = RootChannelOwner(self)117        await self._transport.run()118    def stop_sync(self) -> None:119        self._transport.stop()120        self._dispatcher_fiber.switch()121    def stop_async(self) -> None:122        self._transport.stop()123    async def wait_for_object_with_known_name(self, guid: str) -> Any:124        if guid in self._objects:125            return self._objects[guid]126        callback = self._loop.create_future()127        def callback_wrapper(result: Any) -> None:128            callback.set_result(result)129        self._waiting_for_object[guid] = callback_wrapper130        return await callback131    def call_on_object_with_known_name(132        self, guid: str, callback: Callable[[Any], None]133    ) -> None:134        self._waiting_for_object[guid] = callback135    def _send_message_to_server(136        self, guid: str, method: str, params: Dict137    ) -> ProtocolCallback:138        self._last_id += 1139        id = self._last_id140        message = dict(141            id=id,142            guid=guid,143            method=method,144            params=self._replace_channels_with_guids(params, "params"),145        )146        self._transport.send(message)147        callback = ProtocolCallback(self._loop)148        self._callbacks[id] = callback149        return callback...

chat_window.py

Source:chat_window.py

...124            widget_class = MessageWidget(msg)125            widget_class.setupUi(widget)126            list_item.setSizeHint(widget.sizeHint())127            self.listWidget.setItemWidget(list_item, widget)128    def _send_message_to_server(self, message: str) -> None:129        sender_public_key = rsa.PublicKey.load_pkcs1(self.state["public_key"].encode())130        if self.chat.other.public_key:131            reciever_public_key = rsa.PublicKey.load_pkcs1(132                self.chat.other.public_key.encode()133            )134        else:135            reciever_public_key = rsa.PublicKey.load_pkcs1(136                requests.get(f"{self.state['url']}/user/{self.chat.other.handle}").json[137                    "response"138                ]["public_key"]139            )140        payload: Dict[str, Any] = {141            "sender": self.state["handle"],142            "reciever": self.chat.other.handle,...

challenge35.py

Source:challenge35.py

...42        # Build encrypted message again, this time using the secret derived by43        # the client (which will be 1; see above).44        message = self._build_message_from(self.message, 1)45        self._send(self.iv+message, socket=self.server_socket)46    def _send_message_to_server(self, iv_and_message):47        # Encrypt expected message by server using its secret (already48        # found above).49        message = self._build_message_from(self.message[::-1],50                                           self.server_secret)51        self._send(message, socket=self.client_socket)52        53    def _decrypt_message(self, iv_and_message, secret):54        # Nothing to do since we already have the secret message.55        return self.message56    def _get_custom_g(self):57        raise NotImplementedError58    def _get_possible_server_secrets(self):59        raise NotImplementedError60class KeyExchangeMITMAttackWithGEquals1(KeyExchangeMITMAttackWithCustomG):...

Playwright tutorial

LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.