1  # Copyright (c) Microsoft Corporation. All rights reserved.
   2  # Licensed under the MIT License. See LICENSE in the project root
   3  # for license information.
   4  
   5  import os
   6  import sys
   7  
   8  import debugpy
   9  from debugpy import launcher
  10  from debugpy.common import json
  11  from debugpy.launcher import debuggee
  12  
  13  
  14  def launch_request(request):
  15      debug_options = set(request("debugOptions", json.array(str)))
  16  
  17      # Handling of properties that can also be specified as legacy "debugOptions" flags.
  18      # If property is explicitly set to false, but the flag is in "debugOptions", treat
  19      # it as an error. Returns None if the property wasn't explicitly set either way.
  20      def property_or_debug_option(prop_name, flag_name):
  21          assert prop_name[0].islower() and flag_name[0].isupper()
  22  
  23          value = request(prop_name, bool, optional=True)
  24          if value == ():
  25              value = None
  26  
  27          if flag_name in debug_options:
  28              if value is False:
  29                  raise request.isnt_valid(
  30                      '{0}:false and "debugOptions":[{1}] are mutually exclusive',
  31                      json.repr(prop_name),
  32                      json.repr(flag_name),
  33                  )
  34              value = True
  35  
  36          return value
  37  
  38      python = request("python", json.array(str, size=(1,)))
  39      cmdline = list(python)
  40  
  41      if not request("noDebug", json.default(False)):
  42          # see https://github.com/microsoft/debugpy/issues/861
  43          if sys.version_info[:2] >= (3, 11):
  44              cmdline += ["-X", "frozen_modules=off"]
  45  
  46          port = request("port", int)
  47          cmdline += [
  48              os.path.dirname(debugpy.__file__),
  49              "--connect",
  50              launcher.adapter_host + ":" + str(port),
  51          ]
  52  
  53          if not request("subProcess", True):
  54              cmdline += ["--configure-subProcess", "False"]
  55  
  56          qt_mode = request(
  57              "qt",
  58              json.enum(
  59                  "none", "auto", "pyside", "pyside2", "pyqt4", "pyqt5", optional=True
  60              ),
  61          )
  62          cmdline += ["--configure-qt", qt_mode]
  63  
  64          adapter_access_token = request("adapterAccessToken", str, optional=True)
  65          if adapter_access_token != ():
  66              cmdline += ["--adapter-access-token", adapter_access_token]
  67  
  68          debugpy_args = request("debugpyArgs", json.array(str))
  69          cmdline += debugpy_args
  70  
  71      # Use the copy of arguments that was propagated via the command line rather than
  72      # "args" in the request itself, to allow for shell expansion.
  73      cmdline += sys.argv[1:]
  74  
  75      process_name = request("processName", sys.executable)
  76  
  77      env = os.environ.copy()
  78      env_changes = request("env", json.object((str, type(None))))
  79      if sys.platform == "win32":
  80          # Environment variables are case-insensitive on Win32, so we need to normalize
  81          # both dicts to make sure that env vars specified in the debug configuration
  82          # overwrite the global env vars correctly. If debug config has entries that
  83          # differ in case only, that's an error.
  84          env = {k.upper(): v for k, v in os.environ.items()}
  85          new_env_changes = {}
  86          for k, v in env_changes.items():
  87              k_upper = k.upper()
  88              if k_upper in new_env_changes:
  89                  if new_env_changes[k_upper] == v:
  90                      continue
  91                  else:
  92                      raise request.isnt_valid(
  93                          'Found duplicate in "env": {0}.'.format(k_upper)
  94                      )
  95              new_env_changes[k_upper] = v
  96          env_changes = new_env_changes
  97      if "DEBUGPY_TEST" in env:
  98          # If we're running as part of a debugpy test, make sure that codecov is not
  99          # applied to the debuggee, since it will conflict with pydevd.
 100          env.pop("COV_CORE_SOURCE", None)
 101      env.update(env_changes)
 102      env = {k: v for k, v in env.items() if v is not None}
 103  
 104      if request("gevent", False):
 105          env["GEVENT_SUPPORT"] = "True"
 106  
 107      console = request(
 108          "console",
 109          json.enum(
 110              "internalConsole", "integratedTerminal", "externalTerminal", optional=True
 111          ),
 112      )
 113  
 114      redirect_output = property_or_debug_option("redirectOutput", "RedirectOutput")
 115      if redirect_output is None:
 116          # If neither the property nor the option were specified explicitly, choose
 117          # the default depending on console type - "internalConsole" needs it to
 118          # provide any output at all, but it's unnecessary for the terminals.
 119          redirect_output = console == "internalConsole"
 120      if redirect_output:
 121          # sys.stdout buffering must be disabled - otherwise we won't see the output
 122          # at all until the buffer fills up.
 123          env["PYTHONUNBUFFERED"] = "1"
 124          # Force UTF-8 output to minimize data loss due to re-encoding.
 125          env["PYTHONIOENCODING"] = "utf-8"
 126  
 127      if property_or_debug_option("waitOnNormalExit", "WaitOnNormalExit"):
 128          if console == "internalConsole":
 129              raise request.isnt_valid(
 130                  '"waitOnNormalExit" is not supported for "console":"internalConsole"'
 131              )
 132          debuggee.wait_on_exit_predicates.append(lambda code: code == 0)
 133      if property_or_debug_option("waitOnAbnormalExit", "WaitOnAbnormalExit"):
 134          if console == "internalConsole":
 135              raise request.isnt_valid(
 136                  '"waitOnAbnormalExit" is not supported for "console":"internalConsole"'
 137              )
 138          debuggee.wait_on_exit_predicates.append(lambda code: code != 0)
 139  
 140      debuggee.spawn(process_name, cmdline, env, redirect_output)
 141      return {}
 142  
 143  
 144  def terminate_request(request):
 145      del debuggee.wait_on_exit_predicates[:]
 146      request.respond({})
 147      debuggee.kill()
 148  
 149  
 150  def disconnect():
 151      del debuggee.wait_on_exit_predicates[:]
 152      debuggee.kill()
 153