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

How to use print_wrapped method in autotest

Best Python code snippet using autotest_python

client.py

Source:client.py

...30        return input()31'''32    wrapped function for testing purposes. same interface as print()33'''34def print_wrapped(message=""):35    if TESTING:36        return test_output(message)37    else:38        return print(message)39'''40    @param valid_actions: A dictionary mapping action code numbers to41    their associated functions42    @returns: -1 if there is an error and the action code if the input is43    valid44'''45def collect_user_input(valid_actions):46    try:47        action = int(input_wrapped().strip())48    except ValueError:49        print_wrapped("It looks like you did not input an integer. Try again!\n")50        return -151    if action not in valid_actions:52        print_wrapped("It looks like you input an invalid number. Try again")53        return -154    return action55# Universal action functions56"""57    Adds a ping message to the message queue to check connectivity58    message gets sent by the socket thread59"""60def ping():61    payload = messages.PingMessage()62    message_queue.put(payload.serialize())63# Logged out action functions64"""65    Exits the program when called66"""67def program_quit():68    exit("Goodbye!")69"""70    Asks the user for their username then adds a login request71    to the message queue. Updates global boolean logged_in 72    and global string username.73"""74def login():75    global logged_in76    global username77    local_username = input_wrapped("What is your username?: ")78    payload = messages.HereMessage(local_username)79    message_queue.put(payload.serialize())80    logged_in = True81    username = local_username82"""83    Asks the user the name of the account they want to create84    then adds that request to the message queue. Updates the85    global boolean logged_in and the global string username86"""87def create_account():88    global logged_in89    global username90    local_username = input_wrapped("What do you want your username to be?: ")91    payload = messages.CreateAccountMessage(local_username)92    message_queue.put(payload.serialize())93    logged_in = True94    time.sleep(.5)95    username = local_username96# Universal actions:97PING           = 998# Logged out actions99LOGIN          = 1100CREATE_ACCOUNT = 2101QUIT           = 3102# Dict mapping action numbers to action functions103LOGGED_OUT_ACTIONS = {104    LOGIN: login,105    CREATE_ACCOUNT: create_account,106    QUIT: program_quit,107    PING: ping108}109# Logged in action functions110"""111    Logs the user out, adding an Away Message to the112    message queue. Updates the global boolean logged_in113    to False114"""115def logout():116    global logged_in117    payload = messages.AwayMessage()118    message_queue.put(payload.serialize())119    print_wrapped("Logged out!")120    logged_in = False121"""122    Prompts the user for a recipient and their message. Places that message on123    the message queue.124"""125def chat_send():126    receiver = input_wrapped("Who do you want to send the message to?: ").strip()127    print_wrapped("Write your message below and press 'enter' to send:")128    message = input_wrapped()129    payload = messages.SendChatMessage(receiver, message)130    message_queue.put(payload.serialize())131    print_wrapped("Message sent!")132"""133    Places a the bytes for a user list request on the message queue134"""135def list_users():136    payload = messages.RequestUserListMessage()137    message_queue.put(payload.serialize())138"""139    Updates global boolean logged_in to False. Places message to 140    delete account on the message queue141"""142def delete_account():143    global logged_in144    payload = messages.DeleteAccountMessage()145    message_queue.put(payload.serialize())146    print_wrapped("Account deleted!")147    logged_in = False148"""149    Places a request to show undelivered messages onto the 150    message queue.151"""152def show_messages():153    payload = messages.ShowUndeliveredMessage()154    message_queue.put(payload.serialize())155# Logged in actions156LOGOUT = 1157CHAT_SEND = 2158LIST_USERS = 3159DELETE_ACCOUNT = 4160SHOW_MESSAGES = 5161# Maps actions to their respective functions162LOGGED_IN_ACTIONS = {163    LOGOUT: logout,164    CHAT_SEND: chat_send,165    LIST_USERS: list_users,166    DELETE_ACCOUNT: delete_account,167    SHOW_MESSAGES: show_messages,168    PING: ping169}170# The user flow when logged in.171def logged_in_sequence():172    global username173    print_wrapped(f"You are logged in as " + username + "!")174    print_wrapped(("What would you like to do now? Type '1' to log out, '2' to send a chat, "175        "'3' to list all available users, '4' to delete your account, '5' to receive undelivered messages, and '9' to test connection. "))176    action = collect_user_input(LOGGED_IN_ACTIONS)177    if action != -1:178        LOGGED_IN_ACTIONS[action]()179"""180    Updates the global boolean logged_in and is_connected.181    Logs the user out and shuts down the connection.182"""183def logout(sock):184    global logged_in185    global is_connected186    sock.close()187    is_connected = False188    logged_in = False189"""190    Reads bytes from the socket, timing out after .5 seconds.191    Returns None if no messages show up, or if it fails.192"""193def read_message_bytes(sock):194    # Look to see if any messages showed up.195    try:196        ready = select.select([sock], [], [], .5)197    except:198        return None199    # no message received200    if not ready[0]:201        return None202    # read the first 64 bytes203    try:204        message = sock.recv(64)205    except OSError as e:206        logout(sock)207        print_wrapped("Failed to receive data. Resetting connection.")208        return None209    # if we haven't received a message, nothing to do210    if not message or len(message) < 12:211        return None212    # read the rest of the message213    message_len = messages.extract_length(message)214    while len(message) < message_len:215        try:216            chunk = sock.recv(64)217        except OSError:218            logout(sock)219            print_wrapped("Failed to receive data. Resetting connection.")220            return None221        message += chunk222    return message223# The listener thread runs this loop, checking for and handling224# any new communication from the server.225def socket_loop(sock):226    global logged_in227    global is_connected228    while True:229        # First send any messages on the queue to the server.230        while not message_queue.empty():231            message = message_queue.get()232            sock.send(message)233        message_bytes = read_message_bytes(sock)234        # Nothing new from the server.235        if not message_bytes:236            continue237        try:238            message_object = messages.deserialize_message(message_bytes)239        except Exception as e:240            print_wrapped("Failed to deserialize message." + str(e.message))241            print_wrapped("Invalid message received. Closing program.")242            logout(sock)243            return244        # Handle messages from server, depending on type.245        message_type = type(message_object)246        # received response to a ping247        if message_type == messages.PongMessage:248            print_wrapped("Pong message received!")249        # received a response to a request for a list of users250        elif message_type == messages.UserListResponseMessage:251            print_wrapped("These are the users!:")252            print_wrapped(message_object.user_list)253            print_wrapped()254        # received messages255        elif message_type == messages.DeliverMessage:256            for message in message_object.message_list:257                sender, body = message258                print_wrapped(f"Message from " + sender + ":")259                print_wrapped(body)260                print_wrapped()261        # received an error262        elif message_type == messages.ErrorMessage:263            print_wrapped("Error received: " + str(message_object.error_message))264            print_wrapped("You may need to restart the client to resume normal behavior.")265        # Server is sending nonsense.266        else:267            print("Invalid message object.")268# The user flow when logged out.269def logged_out_sequence():270    global is_connected271    if not is_connected:272        print_wrapped("Establishing connection with server...")273        sock = None274        try:275            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)276            sock.connect((HOST, PORT))277        except OSError:278            print_wrapped("Failure to connect!")279            program_quit()280        is_connected = True281        threading.Thread(target=socket_loop, args=(sock,)).start()282        print_wrapped("Successfully connected to server!")283    print_wrapped()284    print_wrapped(("Welcome to Sooper Chat! Type '1' to log in, '2' to "285            "create an account, '3' to quit, and '9' to ping (test connection)."))286    action = collect_user_input(LOGGED_IN_ACTIONS)287    if action != -1:288        LOGGED_OUT_ACTIONS[action]()289# Main event loop290def main():291    global logged_in292    global is_connected293    while True:294        time.sleep(1)295        if logged_in:296            logged_in_sequence()297        else:298            logged_out_sequence()...

wizard.py

Source:wizard.py

...8from ehforwarderbot import coordinator, utils9from ehforwarderbot.types import ModuleID10from . import FBMessengerChannel11from .__main__ import run12def print_wrapped(text):13    paras = text.split("\n")14    for i in paras:15        print(*cjkwrap.wrap(i), sep="\n")16translator = translation("efb_fb_messenger_slave",17                         resource_filename("efb_fb_messenger_slave", "locale"),18                         fallback=True)19_ = translator.gettext20ngettext = translator.ngettext21class DataModel:22    data: dict23    def __init__(self, profile: str, instance_id: str):24        coordinator.profile = profile25        self.profile = profile26        self.instance_id = instance_id27        self.channel_id = FBMessengerChannel.channel_id28        if instance_id:29            self.channel_id = ModuleID(self.channel_id + "#" + instance_id)30        self.config_path = utils.get_config_path(self.channel_id)31        self.session_path = utils.get_data_path(self.channel_id) / "session.pickle"32        self.yaml = YAML()33        if not self.config_path.exists():34            self.build_default_config()35        else:36            self.data = self.yaml.load(self.config_path.open())37    def build_default_config(self):38        # TRANSLATORS: This part of text must be formatted in a monospaced font and no line shall exceed the width of a 70-cell-wide terminal.39        s = _(40            "# =======================================================\n"41            "# EFB Facebook Messenger Slave Channel Configuration File\n"42            "# =======================================================\n"43            "#\n"44            "# This file can help you to adjust some experimental features provided in\n"45            "# EFMS. It is not required to have this file for EFMS to work.\n"46        )47        s += "\n"48        s += "flags: {}"49        str_io = StringIO(s)50        str_io.seek(0)51        self.data = self.yaml.load(str_io)52    def save(self):53        with self.config_path.open('w') as f:54            self.yaml.dump(self.data, f)55def setup_account(data):56    print_wrapped(_(57        "Log in to your Facebook (Messenger) account\n"58        "-------------------------------------------"59    ))60    print()61    if data.session_path.exists():62        print_wrapped(_(63            "We find that you have successfully logged in once before. "64            "You don't need to log in again if everything is already "65            "working properly."66        ))67        widget = bullet.YesNo(prompt=_("Do you want to log in again? "),68                              prompt_prefix="[yN] ",69                              default="n")70        if not widget.launch():71            return72    run(data.instance_id, data.profile)73flags_settings = {74    "proxy_links_by_facebook":75        (False, 'bool', None,76         _('Deliver links (including links in share entities and thumbnails) '77           'using Facebookâs proxy. Disable this option to show the source '78           'directly.')79         ),80    "send_link_with_description":81        (False, 'bool', None,82         _('When processing link message from the Master Channel, attach '83           'the title and description besides the link when the option is '84           'enabled.\n'85           '\n'86           'Note: Regardless of this option, link messages are sent as text, '87           'and Facebook Messenger may or may not attach its own link '88           'preview per its system configuration.')89         ),90    "show_pending_threads":91        (False, 'bool', None,92         _('When showing the threads list, include threads '93           'pending approval.')94         ),95    "show_archived_threads":96        (False, 'bool', None,97         _('When showing the threads list, include archived threads.')98         )99}100def setup_experimental_flags(data):101    print()102    print_wrapped(_(103        "EFMS has also provided some experimental features that you can use. "104        "They are not required to be enabled for EFMS to work."105    ))106    widget = YesNo(prompt=_("Do you want to config experimental features? "),107                   prompt_prefix="[yN] ", default="n")108    if not widget.launch():109        return110    for key, value in flags_settings.items():111        default, cat, params, desc = value112        if data.data['flags'].get(key) is not None:113            default = data.data['flags'].get(key)114        if cat == 'bool':115            prompt_prefix = '[Yn] ' if default else '[yN] '116            print()117            print(key)118            print_wrapped(desc)119            ans = YesNo(prompt=f"{key}? ",120                        prompt_prefix=prompt_prefix,121                        default='y' if default else 'n') \122                .launch()123            data.data['flags'][key] = ans124        elif cat == 'int':125            print()126            print(key)127            print_wrapped(desc)128            ans = Numbers(prompt=f"{key} [{default}]? ", type=int) \129                .launch(default=default)130            data.data['flags'][key] = ans131        elif cat == 'choices':132            print()133            print(key)134            print_wrapped(desc)135            ans = Bullet(prompt=f"{key}?", choices=params) \136                .launch(default=default)137            data.data['flags'][key] = ans138    print(_("Saving configurations..."), end="", flush=True)139    data.save()140    print(_("OK"))141def wizard(profile, instance_id):142    data = DataModel(profile, instance_id)143    print_wrapped(_(144        "=================================================\n"145        "EFB Facebook Messenger Slave Channel Setup Wizard\n"146        "=================================================\n"147        "\n"148        "This wizard will guide you to setup your EFB Facebook Messenger "149        "Slave channel (EFMS). This would be really quick and simple."150    ))151    print()152    setup_account(data)153    setup_experimental_flags(data)154    print()155    print_wrapped(_(156        "Congratulations! You have finished the setup wizard for EFB Facebook "157        "Messenger Slave channel."...

parfor_diagnostics.py

Source:parfor_diagnostics.py

...15            all_lines = self.get_all_lines(parfors_simple)16            print(" Auto-offloading ".center(_termwidth, "-"))17            self.print_auto_offloading(all_lines)18            if "kernel" in self.extra_info.keys():19                print_wrapped("Device - '%s'" % self.extra_info["kernel"])20            print(_termwidth * "-")21    def print_auto_offloading(self, lines):22        # Code partially borrowed from https://github.com/IntelPython/numba/blob/97fe221b3704bd17567b57ea47f4fc6604476cf9/numba/parfors/parfor.py#L98223        sword = "+--"24        fac = len(sword)25        fadj, froots = self.compute_graph_info(self.fusion_info)26        nadj, _nroots = self.compute_graph_info(self.nested_fusion_info)27        if len(fadj) > len(nadj):28            lim = len(fadj)29            tmp = nadj30        else:31            lim = len(nadj)32            tmp = fadj33        for x in range(len(tmp), lim):34            tmp.append([])35        summary = dict()36        # region : {fused, serialized}37        def print_nest(fadj_, nadj_, theroot, reported, region_id):38            def print_g(fadj_, nadj_, nroot, depth):39                for k in nadj_[nroot]:40                    msg = fac * depth * " " + "%s%s %s" % (sword, k, "(serial")41                    if nadj_[k] == []:42                        fused = []43                        if fadj_[k] != [] and k not in reported:44                            fused = sorted(self.reachable_nodes(fadj_, k))45                            msg += ", fused with loop(s): "46                            msg += ", ".join([str(x) for x in fused])47                        msg += ")"48                        reported.append(k)49                        print_wrapped(msg)50                        summary[region_id]["fused"] += len(fused)51                    else:52                        print_wrapped(msg + ")")53                        print_g(fadj_, nadj_, k, depth + 1)54                    summary[region_id]["serialized"] += 155            if nadj_[theroot] != []:56                print_wrapped("Parallel region %s:" % region_id)57                print_wrapped("%s%s %s" % (sword, theroot, "(parallel)"))58                summary[region_id] = {59                    "root": theroot,60                    "fused": 0,61                    "serialized": 0,62                }63                print_g(fadj_, nadj_, theroot, 1)64                print("\n")65                region_id = region_id + 166            return region_id67        def print_fuse(ty, pf_id, adj, depth, region_id):68            print_wrapped("Parallel region %s:" % region_id)69            msg = fac * depth * " " + "%s%s %s" % (sword, pf_id, "(parallel")70            fused = []71            if adj[pf_id] != []:72                fused = sorted(self.reachable_nodes(adj, pf_id))73                msg += ", fused with loop(s): "74                msg += ", ".join([str(x) for x in fused])75            summary[region_id] = {76                "root": pf_id,77                "fused": len(fused),78                "serialized": 0,79            }80            msg += ")"81            print_wrapped(msg)82            extra_info = self.extra_info.get(str(region_id))83            if extra_info:84                print_wrapped("Device - '%s'" % extra_info)85            region_id = region_id + 186            return region_id87        # Walk the parfors by src line and print optimised structure88        region_id = 089        reported = []90        for line, info in sorted(lines.items()):91            opt_ty, pf_id, adj = info92            if opt_ty == "fuse":93                if pf_id not in reported:94                    region_id = print_fuse("f", pf_id, adj, 0, region_id)95            elif opt_ty == "nest":96                region_id = print_nest(fadj, nadj, pf_id, reported, region_id)97            else:98                assert 099        # print the summary of the fuse/serialize rewrite100        if summary:101            for k, v in sorted(summary.items()):102                msg = (103                    "\n \nParallel region %s (loop #%s) had %s " "loop(s) fused"104                )105                root = v["root"]106                fused = v["fused"]107                serialized = v["serialized"]108                if serialized != 0:109                    msg += (110                        " and %s loop(s) "111                        "serialized as part of the larger "112                        "parallel loop (#%s)."113                    )114                    print_wrapped(msg % (k, root, fused, serialized, root))115                else:116                    msg += "."117                    print_wrapped(msg % (k, root, fused))118        else:...

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.