mirror of
https://github.com/JakeHillion/drgn.git
synced 2024-12-23 09:43:06 +00:00
12c2de2956
This implements the existing thread API methods for live processes other than drgn_thread_stack_trace(). It also doesn't yet add support for full-blown tracing, but it at least brings live processes to feature parity. This is taken from the non-ptrace parts of Kevin Svetlitski's PR #142, with some modifications. Signed-off-by: Omar Sandoval <osandov@osandov.com>
98 lines
2.5 KiB
Python
98 lines
2.5 KiB
Python
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
import os
|
|
import os.path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
from drgn import Program
|
|
from tests import TestCase
|
|
|
|
|
|
class TestLive(TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.prog = Program()
|
|
cls.prog.set_pid(os.getpid())
|
|
|
|
def test_threads(self):
|
|
tids = [thread.tid for thread in self.prog.threads()]
|
|
self.assertIn(os.getpid(), tids)
|
|
for tid in tids:
|
|
self.assertEqual(self.prog.thread(tid).tid, tid)
|
|
|
|
def test_thread_not_found(self):
|
|
self.assertRaises(LookupError, self.prog.thread, 1)
|
|
|
|
def test_main_thread(self):
|
|
self.assertEqual(self.prog.main_thread().tid, os.getpid())
|
|
|
|
def test_crashed_thread(self):
|
|
self.assertRaisesRegex(
|
|
ValueError,
|
|
"crashed thread is only defined for core dumps",
|
|
self.prog.crashed_thread,
|
|
)
|
|
|
|
|
|
class TestCoreDump(TestCase):
|
|
TIDS = (
|
|
2265413,
|
|
2265414,
|
|
2265415,
|
|
2265416,
|
|
2265417,
|
|
2265418,
|
|
2265419,
|
|
2265420,
|
|
2265421,
|
|
2265422,
|
|
2265423,
|
|
2265424,
|
|
2265425,
|
|
)
|
|
|
|
MAIN_TID = 2265413
|
|
CRASHED_TID = 2265419
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
with tempfile.NamedTemporaryFile() as core_dump_file:
|
|
try:
|
|
subprocess.check_call(
|
|
[
|
|
"zstd",
|
|
"--quiet",
|
|
"--decompress",
|
|
"--stdout",
|
|
"tests/sample.coredump.zst",
|
|
],
|
|
stdout=core_dump_file,
|
|
)
|
|
except FileNotFoundError:
|
|
raise unittest.SkipTest("zstd not found")
|
|
cls.prog = Program()
|
|
cls.prog.set_core_dump(core_dump_file.name)
|
|
|
|
def test_threads(self):
|
|
self.assertSequenceEqual(
|
|
sorted(thread.tid for thread in self.prog.threads()),
|
|
self.TIDS,
|
|
)
|
|
|
|
def test_thread(self):
|
|
for tid in self.TIDS:
|
|
self.assertEqual(self.prog.thread(tid).tid, tid)
|
|
|
|
def test_thread_not_found(self):
|
|
self.assertRaises(LookupError, self.prog.thread, 99)
|
|
|
|
def test_main_thread(self):
|
|
self.assertEqual(self.prog.main_thread().tid, self.MAIN_TID)
|
|
|
|
def test_crashed_thread(self):
|
|
self.assertEqual(self.prog.crashed_thread().tid, self.CRASHED_TID)
|