brain_subprocess.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136
  1. # Copyright (c) 2016-2020 Claudiu Popa <pcmanticore@gmail.com>
  2. # Copyright (c) 2017 Hugo <hugovk@users.noreply.github.com>
  3. # Copyright (c) 2018 Peter Talley <peterctalley@gmail.com>
  4. # Copyright (c) 2018 Bryce Guinta <bryce.paul.guinta@gmail.com>
  5. # Copyright (c) 2019 Hugo van Kemenade <hugovk@users.noreply.github.com>
  6. # Copyright (c) 2020-2021 hippo91 <guillaume.peillex@gmail.com>
  7. # Copyright (c) 2020 Peter Pentchev <roam@ringlet.net>
  8. # Copyright (c) 2021 Pierre Sassoulas <pierre.sassoulas@gmail.com>
  9. # Copyright (c) 2021 Daniël van Noord <13665637+DanielNoord@users.noreply.github.com>
  10. # Copyright (c) 2021 Marc Mueller <30130371+cdce8p@users.noreply.github.com>
  11. # Copyright (c) 2021 Damien Baty <damien@damienbaty.com>
  12. # Licensed under the LGPL: https://www.gnu.org/licenses/old-licenses/lgpl-2.1.en.html
  13. # For details: https://github.com/PyCQA/astroid/blob/main/LICENSE
  14. import textwrap
  15. from astroid.brain.helpers import register_module_extender
  16. from astroid.builder import parse
  17. from astroid.const import PY37_PLUS, PY39_PLUS
  18. from astroid.manager import AstroidManager
  19. def _subprocess_transform():
  20. communicate = (bytes("string", "ascii"), bytes("string", "ascii"))
  21. communicate_signature = "def communicate(self, input=None, timeout=None)"
  22. args = """\
  23. self, args, bufsize=0, executable=None, stdin=None, stdout=None, stderr=None,
  24. preexec_fn=None, close_fds=False, shell=False, cwd=None, env=None,
  25. universal_newlines=False, startupinfo=None, creationflags=0, restore_signals=True,
  26. start_new_session=False, pass_fds=(), *, encoding=None, errors=None"""
  27. if PY37_PLUS:
  28. args += ", text=None"
  29. init = f"""
  30. def __init__({args}):
  31. pass"""
  32. wait_signature = "def wait(self, timeout=None)"
  33. ctx_manager = """
  34. def __enter__(self): return self
  35. def __exit__(self, *args): pass
  36. """
  37. py3_args = "args = []"
  38. if PY37_PLUS:
  39. check_output_signature = """
  40. check_output(
  41. args, *,
  42. stdin=None,
  43. stderr=None,
  44. shell=False,
  45. cwd=None,
  46. encoding=None,
  47. errors=None,
  48. universal_newlines=False,
  49. timeout=None,
  50. env=None,
  51. text=None,
  52. restore_signals=True,
  53. preexec_fn=None,
  54. pass_fds=(),
  55. input=None,
  56. bufsize=0,
  57. executable=None,
  58. close_fds=False,
  59. startupinfo=None,
  60. creationflags=0,
  61. start_new_session=False
  62. ):
  63. """.strip()
  64. else:
  65. check_output_signature = """
  66. check_output(
  67. args, *,
  68. stdin=None,
  69. stderr=None,
  70. shell=False,
  71. cwd=None,
  72. encoding=None,
  73. errors=None,
  74. universal_newlines=False,
  75. timeout=None,
  76. env=None,
  77. restore_signals=True,
  78. preexec_fn=None,
  79. pass_fds=(),
  80. input=None,
  81. bufsize=0,
  82. executable=None,
  83. close_fds=False,
  84. startupinfo=None,
  85. creationflags=0,
  86. start_new_session=False
  87. ):
  88. """.strip()
  89. code = textwrap.dedent(
  90. f"""
  91. def {check_output_signature}
  92. if universal_newlines:
  93. return ""
  94. return b""
  95. class Popen(object):
  96. returncode = pid = 0
  97. stdin = stdout = stderr = file()
  98. {py3_args}
  99. {communicate_signature}:
  100. return {communicate!r}
  101. {wait_signature}:
  102. return self.returncode
  103. def poll(self):
  104. return self.returncode
  105. def send_signal(self, signal):
  106. pass
  107. def terminate(self):
  108. pass
  109. def kill(self):
  110. pass
  111. {ctx_manager}
  112. """
  113. )
  114. if PY39_PLUS:
  115. code += """
  116. @classmethod
  117. def __class_getitem__(cls, item):
  118. pass
  119. """
  120. init_lines = textwrap.dedent(init).splitlines()
  121. indented_init = "\n".join(" " * 4 + line for line in init_lines)
  122. code += indented_init
  123. return parse(code)
  124. register_module_extender(AstroidManager(), "subprocess", _subprocess_transform)