MCPcopy Create free account
hub / github.com/apache/arrow / GdbSession

Class GdbSession

python/pyarrow/tests/test_gdb.py:84–175  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

82
83
84class GdbSession:
85 proc = None
86 verbose = True
87
88 def __init__(self, *args, **env):
89 # Let stderr through to let pytest display it separately on errors
90 gdb_env = environment_for_gdb()
91 gdb_env.update(env)
92 self.proc = subprocess.Popen(gdb_command + list(args),
93 env=gdb_env, bufsize=0,
94 stdin=subprocess.PIPE,
95 stdout=subprocess.PIPE)
96 self.last_stdout = []
97 self.last_stdout_line = b""
98
99 def wait_until_ready(self):
100 """
101 Record output until the gdb prompt displays. Return recorded output.
102 """
103 # TODO: add timeout?
104 while (not self.last_stdout_line.startswith(b"(gdb) ") and
105 self.proc.poll() is None):
106 block = self.proc.stdout.read(4096)
107 if self.verbose:
108 sys.stdout.buffer.write(block)
109 sys.stdout.buffer.flush()
110 block, sep, last_line = block.rpartition(b"\n")
111 if sep:
112 self.last_stdout.append(self.last_stdout_line)
113 self.last_stdout.append(block + sep)
114 self.last_stdout_line = last_line
115 else:
116 assert block == b""
117 self.last_stdout_line += last_line
118
119 if self.proc.poll() is not None:
120 raise IOError("gdb session terminated unexpectedly")
121
122 out = b"".join(self.last_stdout).decode('utf-8')
123 self.last_stdout = []
124 self.last_stdout_line = b""
125 return out
126
127 def issue_command(self, line):
128 line = line.encode('utf-8') + b"\n"
129 if self.verbose:
130 sys.stdout.buffer.write(line)
131 sys.stdout.buffer.flush()
132 self.proc.stdin.write(line)
133 self.proc.stdin.flush()
134
135 def run_command(self, line):
136 self.issue_command(line)
137 return self.wait_until_ready()
138
139 def print_value(self, expr):
140 """
141 Ask gdb to print the value of an expression and return the result.

Callers 1

gdbFunction · 0.85

Calls

no outgoing calls

Tested by 1

gdbFunction · 0.68