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

How to use redirect_output method in autotest

Best Python code snippet using autotest_python

tempfile_services.py

Source:tempfile_services.py

1import binascii2import collections3import datetime4import errno5import logging6import os7import shutil8import tempfile9import ray.utils10logger = logging.getLogger(__name__)11_incremental_dict = collections.defaultdict(lambda: 0)12_temp_root = None13def make_inc_temp(suffix="", prefix="", directory_name="/tmp/ray"):14    """Return a incremental temporary file name. The file is not created.15    Args:16        suffix (str): The suffix of the temp file.17        prefix (str): The prefix of the temp file.18        directory_name (str) : The base directory of the temp file.19    Returns:20        A string of file name. If there existing a file having the same name,21        the returned name will look like22        "{directory_name}/{prefix}.{unique_index}{suffix}"23    """24    index = _incremental_dict[suffix, prefix, directory_name]25    # `tempfile.TMP_MAX` could be extremely large,26    # so using `range` in Python2.x should be avoided.27    while index < tempfile.TMP_MAX:28        if index == 0:29            filename = os.path.join(directory_name, prefix + suffix)30        else:31            filename = os.path.join(directory_name,32                                    prefix + "." + str(index) + suffix)33        index += 134        if not os.path.exists(filename):35            _incremental_dict[suffix, prefix,36                              directory_name] = index  # Save the index.37            return filename38    raise FileExistsError(errno.EEXIST, "No usable temporary filename found")39def try_to_create_directory(directory_path):40    """Attempt to create a directory that is globally readable/writable.41    Args:42        directory_path: The path of the directory to create.43    """44    if not os.path.exists(directory_path):45        try:46            os.makedirs(directory_path)47        except OSError as e:48            if e.errno != os.errno.EEXIST:49                raise e50            logger.warning(51                "Attempted to create '{}', but the directory already "52                "exists.".format(directory_path))53        # Change the log directory permissions so others can use it. This is54        # important when multiple people are using the same machine.55        os.chmod(directory_path, 0o0777)56def get_temp_root():57    """Get the path of the temporary root. If not existing, it will be created.58    """59    global _temp_root60    date_str = datetime.datetime.today().strftime("%Y-%m-%d_%H-%M-%S")61    # Lazy creation. Avoid creating directories never used.62    if _temp_root is None:63        _temp_root = make_inc_temp(64            prefix="session_{date_str}_{pid}".format(65                pid=os.getpid(), date_str=date_str),66            directory_name="/tmp/ray")67    try_to_create_directory(_temp_root)68    return _temp_root69def set_temp_root(path):70    """Set the path of the temporary root. It will be created lazily."""71    global _temp_root72    _temp_root = path73def get_logs_dir_path():74    """Get a temp dir for logging."""75    logs_dir = os.path.join(get_temp_root(), "logs")76    try_to_create_directory(logs_dir)77    return logs_dir78def get_sockets_dir_path():79    """Get a temp dir for sockets."""80    sockets_dir = os.path.join(get_temp_root(), "sockets")81    try_to_create_directory(sockets_dir)82    return sockets_dir83def get_raylet_socket_name(suffix=""):84    """Get a socket name for raylet."""85    sockets_dir = get_sockets_dir_path()86    raylet_socket_name = make_inc_temp(87        prefix="raylet", directory_name=sockets_dir, suffix=suffix)88    return raylet_socket_name89def get_object_store_socket_name():90    """Get a socket name for plasma object store."""91    sockets_dir = get_sockets_dir_path()92    return make_inc_temp(prefix="plasma_store", directory_name=sockets_dir)93def get_ipython_notebook_path(port):94    """Get a new ipython notebook path"""95    notebook_filepath = os.path.join(96        os.path.dirname(os.path.abspath(__file__)), "WebUI.ipynb")97    # We copy the notebook file so that the original doesn't get modified by98    # the user.99    notebook_name = make_inc_temp(100        suffix=".ipynb", prefix="ray_ui", directory_name=get_temp_root())101    new_notebook_filepath = os.path.join(get_logs_dir_path(), notebook_name)102    shutil.copy(notebook_filepath, new_notebook_filepath)103    new_notebook_directory = os.path.dirname(new_notebook_filepath)104    token = ray.utils.decode(binascii.hexlify(os.urandom(24)))105    webui_url = ("http://localhost:{}/notebooks/{}?token={}".format(106        port, os.path.basename(notebook_name), token))107    return new_notebook_directory, webui_url, token108def new_log_files(name, redirect_output):109    """Generate partially randomized filenames for log files.110    Args:111        name (str): descriptive string for this log file.112        redirect_output (bool): True if files should be generated for logging113            stdout and stderr and false if stdout and stderr should not be114            redirected.115    Returns:116        If redirect_output is true, this will return a tuple of two117            filehandles. The first is for redirecting stdout and the second is118            for redirecting stderr. If redirect_output is false, this will119            return a tuple of two None objects.120    """121    if not redirect_output:122        return None, None123    # Create a directory to be used for process log files.124    logs_dir = get_logs_dir_path()125    # Create another directory that will be used by some of the RL algorithms.126    # TODO(suquark): This is done by the old code.127    # We should be able to control its path later.128    try_to_create_directory("/tmp/ray")129    log_stdout = make_inc_temp(130        suffix=".out", prefix=name, directory_name=logs_dir)131    log_stderr = make_inc_temp(132        suffix=".err", prefix=name, directory_name=logs_dir)133    # Line-buffer the output (mode 1)134    log_stdout_file = open(log_stdout, "a", buffering=1)135    log_stderr_file = open(log_stderr, "a", buffering=1)136    return log_stdout_file, log_stderr_file137def new_redis_log_file(redirect_output, shard_number=None):138    """Create new logging files for redis"""139    if shard_number is None:140        redis_stdout_file, redis_stderr_file = new_log_files(141            "redis", redirect_output)142    else:143        redis_stdout_file, redis_stderr_file = new_log_files(144            "redis-shard_{}".format(shard_number), redirect_output)145    return redis_stdout_file, redis_stderr_file146def new_raylet_log_file(local_scheduler_index, redirect_output):147    """Create new logging files for raylet."""148    raylet_stdout_file, raylet_stderr_file = new_log_files(149        "raylet_{}".format(local_scheduler_index),150        redirect_output=redirect_output)151    return raylet_stdout_file, raylet_stderr_file152def new_webui_log_file():153    """Create new logging files for web ui."""154    ui_stdout_file, ui_stderr_file = new_log_files(155        "webui", redirect_output=True)156    return ui_stdout_file, ui_stderr_file157def new_worker_redirected_log_file(worker_id):158    """Create new logging files for workers to redirect its output."""159    worker_stdout_file, worker_stderr_file = (new_log_files(160        "worker-" + ray.utils.binary_to_hex(worker_id), True))161    return worker_stdout_file, worker_stderr_file162def new_log_monitor_log_file():163    """Create new logging files for the log monitor."""164    log_monitor_stdout_file, log_monitor_stderr_file = new_log_files(165        "log_monitor", redirect_output=True)166    return log_monitor_stdout_file, log_monitor_stderr_file167def new_plasma_store_log_file(local_scheduler_index, redirect_output):168    """Create new logging files for the plasma store."""169    plasma_store_stdout_file, plasma_store_stderr_file = new_log_files(170        "plasma_store_{}".format(local_scheduler_index), redirect_output)171    return plasma_store_stdout_file, plasma_store_stderr_file172def new_monitor_log_file(redirect_output):173    """Create new logging files for the monitor."""174    monitor_stdout_file, monitor_stderr_file = new_log_files(175        "monitor", redirect_output)...

attach_server.py

Source:attach_server.py

1# Copyright (c) Microsoft Corporation. All rights reserved.2# Licensed under the MIT License. See LICENSE in the project root3# for license information.45import sys6import warnings78import ptvsd.log9from ptvsd._remote import (10    attach as ptvsd_attach,11    enable_attach as ptvsd_enable_attach,12)13from ptvsd.wrapper import debugger_attached1415import pydevd16from _pydevd_bundle.pydevd_constants import get_global_debugger1718WAIT_TIMEOUT = 1.01920DEFAULT_HOST = '0.0.0.0'21DEFAULT_PORT = 56782223_pending_threads = set()2425_redirect_output_deprecation_msg = (26    "'redirect_output' setting via enable_attach will be deprecated in the future versions of the debugger. "27    "This can be set using redirectOutput in Launch config in VS Code, using Tee output option in Visual Studio, "28    "or debugOptions configuration for any client.")293031def wait_for_attach(timeout=None):32    """If a remote debugger is attached, returns immediately. Otherwise,33    blocks until a remote debugger attaches to this process, or until the34    optional timeout occurs.3536    Parameters37    ----------38    timeout : float, optional39        The timeout for the operation in seconds (or fractions thereof).40    """41    ptvsd.log.info('wait_for_attach{0!r}', (timeout,))42    debugger_attached.wait(timeout)434445def enable_attach(address=(DEFAULT_HOST, DEFAULT_PORT), redirect_output=None, log_dir=None):46    """Enables a client to attach to this process remotely to debug Python code.4748    Parameters49    ----------50    address : (str, int), optional51        Specifies the interface and port on which the debugging server should52        listen for TCP connections. It is in the same format as used for53        regular sockets of the `socket.AF_INET` family, i.e. a tuple of54        ``(hostname, port)``. On client side, the server is identified by the55        Qualifier string in the usual ``'hostname:port'`` format, e.g.:56        ``'myhost.cloudapp.net:5678'``. Default is ``('0.0.0.0', 5678)``.57    redirect_output : bool, optional58        (Deprecated) Specifies whether any output (on both `stdout` and `stderr`) produced59        by this program should be sent to the debugger. Default is ``True``.60    log_dir : str, optional61        Name of the directory that debugger will create its log files in.62        If not specified, logging is disabled.6364    Return65    ------66    Returns tuple (host, port) as used to by the debugging server. If `enable_attach` was67    called with port '0'. The return value will contain the actual ephemeral port number.6869    Notes70    -----71    This function returns immediately after setting up the debugging server,72    and does not block program execution. If you need to block until debugger73    is attached, call `ptvsd.wait_for_attach`. The debugger can be detached74    and re-attached multiple times after `enable_attach` is called.7576    Only the thread on which this function is called, and any threads that are77    created after it returns, will be visible in the debugger once it is78    attached. Any threads that are already running before this function is79    called will not be visible.80    """8182    if log_dir:83        ptvsd.options.log_dir = log_dir84    ptvsd.log.to_file()85    ptvsd.log.info('enable_attach{0!r}', (address, redirect_output))8687    if redirect_output is not None:88        ptvsd.log.info('redirect_output deprecation warning.')89        warnings.warn(_redirect_output_deprecation_msg, DeprecationWarning, stacklevel=2)9091    if is_attached():92        ptvsd.log.info('enable_attach() ignored - already attached.')93        return9495    debugger_attached.clear()9697    # Ensure port is int98    port = address[1]99    address = (address[0], port if type(port) is int else int(port))100101    ptvsd_enable_attach(address)102    return (address[0], ptvsd.options.port)103104105def attach(address, redirect_output=None, log_dir=None):106    """Attaches this process to the debugger listening on a given address.107108    Parameters109    ----------110    address : (str, int), optional111        Specifies the interface and port on which the debugger is listening112        for TCP connections. It is in the same format as used for113        regular sockets of the `socket.AF_INET` family, i.e. a tuple of114        ``(hostname, port)``.115    redirect_output : bool, optional116        (Deprecated) Specifies whether any output (on both `stdout` and `stderr`) produced117        by this program should be sent to the debugger. Default is ``True``.118    log_dir : str, optional119        Name of the directory that debugger will create its log files in.120        If not specified, logging is disabled.121    """122123    if log_dir:124        ptvsd.options.log_dir = log_dir125    ptvsd.log.to_file()126    ptvsd.log.info('attach{0!r}', (address, redirect_output))127128    if redirect_output is not None:129        ptvsd.log.info('redirect_output deprecation warning.')130        warnings.warn(_redirect_output_deprecation_msg, DeprecationWarning)131132    if is_attached():133        ptvsd.log.info('attach() ignored - already attached.')134        return135136    debugger_attached.clear()137138    # Ensure port is int139    port = address[1]140    address = (address[0], port if type(port) is int else int(port))141142    ptvsd_attach(address)143144145def is_attached():146    """Returns ``True`` if debugger is attached, ``False`` otherwise."""147    return debugger_attached.isSet()148149150def break_into_debugger():151    """If a remote debugger is attached, pauses execution of all threads,152    and breaks into the debugger with current thread as active.153    """154155    ptvsd.log.info('break_into_debugger()')156157    if not is_attached():158        ptvsd.log.info('break_into_debugger() ignored - debugger not attached')159        return160161    # Get the first frame in the stack that's not an internal frame.162    global_debugger = get_global_debugger()163    stop_at_frame = sys._getframe().f_back164    while stop_at_frame is not None and global_debugger.get_file_type(165            stop_at_frame) == global_debugger.PYDEV_FILE:166        stop_at_frame = stop_at_frame.f_back167168    # pydevd.settrace() only enables debugging of the current169    # thread and all future threads.  PyDevd is not enabled for170    # existing threads (other than the current one).  Consequently,171    # pydevd.settrace() must be called ASAP in the current thread.172    # See issue #509.173    #174    # This is tricky, however, because settrace() will block until175    # it receives a CMD_RUN message.  You can't just call it in a176    # thread to avoid blocking; doing so would prevent the current177    # thread from being debugged.178    pydevd.settrace(179        suspend=True,180        trace_only_current_thread=True,181        patch_multiprocessing=False,182        stop_at_frame=stop_at_frame,183    )184    stop_at_frame = None185186187def debug_this_thread():
...

_remote.py

Source:_remote.py

1# Copyright (c) Microsoft Corporation. All rights reserved.2# Licensed under the MIT License. See LICENSE in the project root3# for license information.45import pydevd6import threading7import time89from _pydevd_bundle.pydevd_comm import get_global_debugger1011import ptvsd12import ptvsd.options13from ptvsd._util import new_hidden_thread14from ptvsd.pydevd_hooks import install15from ptvsd.daemon import session_not_bound, DaemonClosedError161718def _pydevd_settrace(redirect_output=None, _pydevd=pydevd, **kwargs):19    if redirect_output is not None:20        kwargs.setdefault('stdoutToServer', redirect_output)21        kwargs.setdefault('stderrToServer', redirect_output)22    # pydevd.settrace() only enables debugging of the current23    # thread and all future threads.  PyDevd is not enabled for24    # existing threads (other than the current one).  Consequently,25    # pydevd.settrace() must be called ASAP in the current thread.26    # See issue #509.27    #28    # This is tricky, however, because settrace() will block until29    # it receives a CMD_RUN message.  You can't just call it in a30    # thread to avoid blocking; doing so would prevent the current31    # thread from being debugged.32    _pydevd.settrace(**kwargs)333435# TODO: Split up enable_attach() to align with module organization.36# This should including making better use of Daemon (e,g, the37# start_server() method).38# Then move at least some parts to the appropriate modules.  This module39# is focused on running the debugger.4041global_next_session = lambda: None424344def enable_attach(address,45                  redirect_output=True,46                  _pydevd=pydevd,47                  _install=install,48                  on_attach=lambda: None,49                  **kwargs):5051    ptvsd.main_thread = threading.current_thread()5253    host, port = address5455    def wait_for_connection(daemon, host, port, next_session=None):56        debugger = get_global_debugger()57        while debugger is None:58            time.sleep(0.1)59            debugger = get_global_debugger()6061        debugger.ready_to_run = True6263        while True:64            session_not_bound.wait()65            try:66                global_next_session()67                on_attach()68            except DaemonClosedError:69                return7071    def start_daemon():72        daemon._sock = daemon._start()73        _, next_session = daemon.start_server(addr=(host, port))74        global global_next_session75        global_next_session = next_session76        return daemon._sock7778    daemon = _install(_pydevd,79                      address,80                      start_server=None,81                      start_client=(lambda daemon, h, port: start_daemon()),82                      singlesession=False,83                      **kwargs)8485    connection_thread = new_hidden_thread('ptvsd.listen_for_connection',86                                          wait_for_connection,87                                          args=(daemon, host, port))88    connection_thread.start()8990    if ptvsd.options.no_debug:91        _setup_nodebug()92    else:93        _pydevd.settrace(host=host,94                         stdoutToServer=redirect_output,95                         stderrToServer=redirect_output,96                         port=port,97                         suspend=False,98                         patch_multiprocessing=ptvsd.options.multiprocess)99100    return daemon101102103def attach(address,104           redirect_output=True,105           _pydevd=pydevd,106           _install=install,107           **kwargs):108109    ptvsd.main_thread = threading.current_thread()110111    host, port = address112    daemon = _install(_pydevd, address, singlesession=False, **kwargs)113114    if ptvsd.options.no_debug:115        _setup_nodebug()116    else:117        _pydevd.settrace(host=host,118                         port=port,119                         stdoutToServer=redirect_output,120                         stderrToServer=redirect_output,121                         suspend=False,122                         patch_multiprocessing=ptvsd.options.multiprocess)123124    return daemon125126127def _setup_nodebug():128    debugger = pydevd.PyDB()129    debugger.init_matplotlib_support = lambda *arg: None130    # We are invoking run() solely for side effects here - setting up the131    # debugger and connecting to our socket - so the code run is a no-op.132    debugger.run(133        file='ptvsd._remote:_nop',134        globals=None,135        locals=None,136        is_module=True,137        set_trace=False)138139140def _nop():
...

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.