Python code coverage for Lib/test/test_dtrace.py

#	count	content
1	n/a	import dis
2	n/a	import os.path
3	n/a	import re
4	n/a	import subprocess
5	n/a	import sys
6	n/a	import types
7	n/a	import unittest
8	n/a
9	n/a	from test.support import findfile, run_unittest
10	n/a
11	n/a
12	n/a	def abspath(filename):
13	n/a	return os.path.abspath(findfile(filename, subdir="dtracedata"))
14	n/a
15	n/a
16	n/a	def normalize_trace_output(output):
17	n/a	"""Normalize DTrace output for comparison.
18	n/a
19	n/a	DTrace keeps a per-CPU buffer, and when showing the fired probes, buffers
20	n/a	are concatenated. So if the operating system moves our thread around, the
21	n/a	straight result can be "non-causal". So we add timestamps to the probe
22	n/a	firing, sort by that field, then strip it from the output"""
23	n/a
24	n/a	# When compiling with '--with-pydebug', strip '[# refs]' debug output.
25	n/a	output = re.sub(r"\[[0-9]+ refs\]", "", output)
26	n/a	try:
27	n/a	result = [
28	n/a	row.split("\t")
29	n/a	for row in output.splitlines()
30	n/a	if row and not row.startswith('#')
31	n/a	]
32	n/a	result.sort(key=lambda row: int(row[0]))
33	n/a	result = [row[1] for row in result]
34	n/a	return "\n".join(result)
35	n/a	except (IndexError, ValueError):
36	n/a	raise AssertionError(
37	n/a	"tracer produced unparseable output:\n{}".format(output)
38	n/a	)
39	n/a
40	n/a
41	n/a	class TraceBackend:
42	n/a	EXTENSION = None
43	n/a	COMMAND = None
44	n/a	COMMAND_ARGS = []
45	n/a
46	n/a	def run_case(self, name, optimize_python=None):
47	n/a	actual_output = normalize_trace_output(self.trace_python(
48	n/a	script_file=abspath(name + self.EXTENSION),
49	n/a	python_file=abspath(name + ".py"),
50	n/a	optimize_python=optimize_python))
51	n/a
52	n/a	with open(abspath(name + self.EXTENSION + ".expected")) as f:
53	n/a	expected_output = f.read().rstrip()
54	n/a
55	n/a	return (expected_output, actual_output)
56	n/a
57	n/a	def generate_trace_command(self, script_file, subcommand=None):
58	n/a	command = self.COMMAND + [script_file]
59	n/a	if subcommand:
60	n/a	command += ["-c", subcommand]
61	n/a	return command
62	n/a
63	n/a	def trace(self, script_file, subcommand=None):
64	n/a	command = self.generate_trace_command(script_file, subcommand)
65	n/a	stdout, _ = subprocess.Popen(command,
66	n/a	stdout=subprocess.PIPE,
67	n/a	stderr=subprocess.STDOUT,
68	n/a	universal_newlines=True).communicate()
69	n/a	return stdout
70	n/a
71	n/a	def trace_python(self, script_file, python_file, optimize_python=None):
72	n/a	python_flags = []
73	n/a	if optimize_python:
74	n/a	python_flags.extend(["-O"] * optimize_python)
75	n/a	subcommand = " ".join([sys.executable] + python_flags + [python_file])
76	n/a	return self.trace(script_file, subcommand)
77	n/a
78	n/a	def assert_usable(self):
79	n/a	try:
80	n/a	output = self.trace(abspath("assert_usable" + self.EXTENSION))
81	n/a	output = output.strip()
82	n/a	except (FileNotFoundError, PermissionError) as fnfe:
83	n/a	output = str(fnfe)
84	n/a	if output != "probe: success":
85	n/a	raise unittest.SkipTest(
86	n/a	"{}(1) failed: {}".format(self.COMMAND[0], output)
87	n/a	)
88	n/a
89	n/a
90	n/a	class DTraceBackend(TraceBackend):
91	n/a	EXTENSION = ".d"
92	n/a	COMMAND = ["dtrace", "-q", "-s"]
93	n/a
94	n/a
95	n/a	class SystemTapBackend(TraceBackend):
96	n/a	EXTENSION = ".stp"
97	n/a	COMMAND = ["stap", "-g"]
98	n/a
99	n/a
100	n/a	class TraceTests(unittest.TestCase):
101	n/a	# unittest.TestCase options
102	n/a	maxDiff = None
103	n/a
104	n/a	# TraceTests options
105	n/a	backend = None
106	n/a	optimize_python = 0
107	n/a
108	n/a	@classmethod
109	n/a	def setUpClass(self):
110	n/a	self.backend.assert_usable()
111	n/a
112	n/a	def run_case(self, name):
113	n/a	actual_output, expected_output = self.backend.run_case(
114	n/a	name, optimize_python=self.optimize_python)
115	n/a	self.assertEqual(actual_output, expected_output)
116	n/a
117	n/a	def test_function_entry_return(self):
118	n/a	self.run_case("call_stack")
119	n/a
120	n/a	def test_verify_call_opcodes(self):
121	n/a	"""Ensure our call stack test hits all function call opcodes"""
122	n/a
123	n/a	opcodes = set(["CALL_FUNCTION", "CALL_FUNCTION_EX", "CALL_FUNCTION_KW"])
124	n/a
125	n/a	with open(abspath("call_stack.py")) as f:
126	n/a	code_string = f.read()
127	n/a
128	n/a	def get_function_instructions(funcname):
129	n/a	# Recompile with appropriate optimization setting
130	n/a	code = compile(source=code_string,
131	n/a	filename="<string>",
132	n/a	mode="exec",
133	n/a	optimize=self.optimize_python)
134	n/a
135	n/a	for c in code.co_consts:
136	n/a	if isinstance(c, types.CodeType) and c.co_name == funcname:
137	n/a	return dis.get_instructions(c)
138	n/a	return []
139	n/a
140	n/a	for instruction in get_function_instructions('start'):
141	n/a	opcodes.discard(instruction.opname)
142	n/a
143	n/a	self.assertEqual(set(), opcodes)
144	n/a
145	n/a	def test_gc(self):
146	n/a	self.run_case("gc")
147	n/a
148	n/a	def test_line(self):
149	n/a	self.run_case("line")
150	n/a
151	n/a
152	n/a	class DTraceNormalTests(TraceTests):
153	n/a	backend = DTraceBackend()
154	n/a	optimize_python = 0
155	n/a
156	n/a
157	n/a	class DTraceOptimizedTests(TraceTests):
158	n/a	backend = DTraceBackend()
159	n/a	optimize_python = 2
160	n/a
161	n/a
162	n/a	class SystemTapNormalTests(TraceTests):
163	n/a	backend = SystemTapBackend()
164	n/a	optimize_python = 0
165	n/a
166	n/a
167	n/a	class SystemTapOptimizedTests(TraceTests):
168	n/a	backend = SystemTapBackend()
169	n/a	optimize_python = 2
170	n/a
171	n/a
172	n/a	def test_main():
173	n/a	run_unittest(DTraceNormalTests, DTraceOptimizedTests, SystemTapNormalTests,
174	n/a	SystemTapOptimizedTests)
175	n/a
176	n/a
177	n/a	if __name__ == '__main__':
178	n/a	test_main()