-"""Python interface for the 'lsprof' profiler.
- Compatible with the 'profile' module.
-"""
-
-__all__ = ["run", "runctx", "Profile"]
-
-import _lsprof
-import importlib.machinery
-import importlib.util
-import io
-import profile as _pyprofile
-
-# ____________________________________________________________
-# Simple interface
-
-def run(statement, filename=None, sort=-1):
- return _pyprofile._Utils(Profile).run(statement, filename, sort)
-
-def runctx(statement, globals, locals, filename=None, sort=-1):
- return _pyprofile._Utils(Profile).runctx(statement, globals, locals,
- filename, sort)
-
-run.__doc__ = _pyprofile.run.__doc__
-runctx.__doc__ = _pyprofile.runctx.__doc__
-
-# ____________________________________________________________
-
-class Profile(_lsprof.Profiler):
- """Profile(timer=None, timeunit=None, subcalls=True, builtins=True)
-
- Builds a profiler object using the specified timer function.
- The default timer is a fast built-in one based on real time.
- For custom timer functions returning integers, timeunit can
- be a float specifying a scale (i.e. how long each integer unit
- is, in seconds).
- """
-
- # Most of the functionality is in the base class.
- # This subclass only adds convenient and backward-compatible methods.
-
- def print_stats(self, sort=-1):
- import pstats
- if not isinstance(sort, tuple):
- sort = (sort,)
- pstats.Stats(self).strip_dirs().sort_stats(*sort).print_stats()
-
- def dump_stats(self, file):
- import marshal
- with open(file, 'wb') as f:
- self.create_stats()
- marshal.dump(self.stats, f)
-
- def create_stats(self):
- self.disable()
- self.snapshot_stats()
+"""Compatibility wrapper for cProfile module.
- def snapshot_stats(self):
- entries = self.getstats()
- self.stats = {}
- callersdicts = {}
- # call information
- for entry in entries:
- func = label(entry.code)
- nc = entry.callcount # ncalls column of pstats (before '/')
- cc = nc - entry.reccallcount # ncalls column of pstats (after '/')
- tt = entry.inlinetime # tottime column of pstats
- ct = entry.totaltime # cumtime column of pstats
- callers = {}
- callersdicts[id(entry.code)] = callers
- self.stats[func] = cc, nc, tt, ct, callers
- # subcall information
- for entry in entries:
- if entry.calls:
- func = label(entry.code)
- for subentry in entry.calls:
- try:
- callers = callersdicts[id(subentry.code)]
- except KeyError:
- continue
- nc = subentry.callcount
- cc = nc - subentry.reccallcount
- tt = subentry.inlinetime
- ct = subentry.totaltime
- if func in callers:
- prev = callers[func]
- nc += prev[0]
- cc += prev[1]
- tt += prev[2]
- ct += prev[3]
- callers[func] = nc, cc, tt, ct
-
- # The following two methods can be called by clients to use
- # a profiler to profile a statement, given as a string.
-
- def run(self, cmd):
- import __main__
- dict = __main__.__dict__
- return self.runctx(cmd, dict, dict)
-
- def runctx(self, cmd, globals, locals):
- self.enable()
- try:
- exec(cmd, globals, locals)
- finally:
- self.disable()
- return self
-
- # This method is more useful to profile a single function call.
- def runcall(self, func, /, *args, **kw):
- self.enable()
- try:
- return func(*args, **kw)
- finally:
- self.disable()
-
- def __enter__(self):
- self.enable()
- return self
-
- def __exit__(self, *exc_info):
- self.disable()
-
-# ____________________________________________________________
+This module maintains backward compatibility by importing from the new
+profiling.tracing module.
+"""
-def label(code):
- if isinstance(code, str):
- return ('~', 0, code) # built-in functions ('~' sorts at the end)
- else:
- return (code.co_filename, code.co_firstlineno, code.co_name)
+from profiling.tracing import run, runctx, Profile
-# ____________________________________________________________
+__all__ = ["run", "runctx", "Profile"]
-def main():
- import os
+if __name__ == "__main__":
import sys
- import runpy
- import pstats
- from optparse import OptionParser
- usage = "cProfile.py [-o output_file_path] [-s sort] [-m module | scriptfile] [arg] ..."
- parser = OptionParser(usage=usage)
- parser.allow_interspersed_args = False
- parser.add_option('-o', '--outfile', dest="outfile",
- help="Save stats to <outfile>", default=None)
- parser.add_option('-s', '--sort', dest="sort",
- help="Sort order when printing to stdout, based on pstats.Stats class",
- default=2,
- choices=sorted(pstats.Stats.sort_arg_dict_default))
- parser.add_option('-m', dest="module", action="store_true",
- help="Profile a library module", default=False)
-
- if not sys.argv[1:]:
- parser.print_usage()
- sys.exit(2)
-
- (options, args) = parser.parse_args()
- sys.argv[:] = args
-
- # The script that we're profiling may chdir, so capture the absolute path
- # to the output file at startup.
- if options.outfile is not None:
- options.outfile = os.path.abspath(options.outfile)
-
- if len(args) > 0:
- if options.module:
- code = "run_module(modname, run_name='__main__')"
- globs = {
- 'run_module': runpy.run_module,
- 'modname': args[0]
- }
- else:
- progname = args[0]
- sys.path.insert(0, os.path.dirname(progname))
- with io.open_code(progname) as fp:
- code = compile(fp.read(), progname, 'exec')
- spec = importlib.machinery.ModuleSpec(name='__main__', loader=None,
- origin=progname)
- module = importlib.util.module_from_spec(spec)
- # Set __main__ so that importing __main__ in the profiled code will
- # return the same namespace that the code is executing under.
- sys.modules['__main__'] = module
- # Ensure that we're using the same __dict__ instance as the module
- # for the global variables so that updates to globals are reflected
- # in the module's namespace.
- globs = module.__dict__
- globs.update({
- '__spec__': spec,
- '__file__': spec.origin,
- '__name__': spec.name,
- '__package__': None,
- '__cached__': None,
- })
-
- try:
- runctx(code, globs, None, options.outfile, options.sort)
- except BrokenPipeError as exc:
- # Prevent "Exception ignored" during interpreter shutdown.
- sys.stdout = None
- sys.exit(exc.errno)
- else:
- parser.print_usage()
- return parser
-
-# When invoked as main program, invoke the profiler on a script
-if __name__ == '__main__':
+ from profiling.tracing.__main__ import main
main()
--- /dev/null
+"""Tracing profiler for Python.
+
+This module provides deterministic profiling of Python programs by tracing
+every function call and return.
+"""
+
+__all__ = ("run", "runctx", "Profile")
+
+import _lsprof
+import importlib.machinery
+import importlib.util
+import io
+from profiling.tracing._utils import _Utils
+
+# ____________________________________________________________
+# Simple interface
+
+def run(statement, filename=None, sort=-1):
+ """Run statement under profiler optionally saving results in filename
+
+ This function takes a single argument that can be passed to the
+ "exec" statement, and an optional file name. In all cases this
+ routine attempts to "exec" its first argument and gather profiling
+ statistics from the execution. If no file name is present, then this
+ function automatically prints a simple profiling report, sorted by the
+ standard name string (file/line/function-name) that is presented in
+ each line.
+ """
+ return _Utils(Profile).run(statement, filename, sort)
+
+def runctx(statement, globals, locals, filename=None, sort=-1):
+ """Run statement under profiler, supplying your own globals and locals,
+ optionally saving results in filename.
+
+ statement and filename have the same semantics as profile.run
+ """
+ return _Utils(Profile).runctx(statement, globals, locals,
+ filename, sort)
+
+# ____________________________________________________________
+
+class Profile(_lsprof.Profiler):
+ """Profile(timer=None, timeunit=None, subcalls=True, builtins=True)
+
+ Builds a profiler object using the specified timer function.
+ The default timer is a fast built-in one based on real time.
+ For custom timer functions returning integers, timeunit can
+ be a float specifying a scale (i.e. how long each integer unit
+ is, in seconds).
+ """
+
+ # Most of the functionality is in the base class.
+ # This subclass only adds convenient and backward-compatible methods.
+
+ def print_stats(self, sort=-1):
+ import pstats
+ if not isinstance(sort, tuple):
+ sort = (sort,)
+ pstats.Stats(self).strip_dirs().sort_stats(*sort).print_stats()
+
+ def dump_stats(self, file):
+ import marshal
+ with open(file, 'wb') as f:
+ self.create_stats()
+ marshal.dump(self.stats, f)
+
+ def create_stats(self):
+ self.disable()
+ self.snapshot_stats()
+
+ def snapshot_stats(self):
+ entries = self.getstats()
+ self.stats = {}
+ callersdicts = {}
+ # call information
+ for entry in entries:
+ func = label(entry.code)
+ nc = entry.callcount # ncalls column of pstats (before '/')
+ cc = nc - entry.reccallcount # ncalls column of pstats (after '/')
+ tt = entry.inlinetime # tottime column of pstats
+ ct = entry.totaltime # cumtime column of pstats
+ callers = {}
+ callersdicts[id(entry.code)] = callers
+ self.stats[func] = cc, nc, tt, ct, callers
+ # subcall information
+ for entry in entries:
+ if entry.calls:
+ func = label(entry.code)
+ for subentry in entry.calls:
+ try:
+ callers = callersdicts[id(subentry.code)]
+ except KeyError:
+ continue
+ nc = subentry.callcount
+ cc = nc - subentry.reccallcount
+ tt = subentry.inlinetime
+ ct = subentry.totaltime
+ if func in callers:
+ prev = callers[func]
+ nc += prev[0]
+ cc += prev[1]
+ tt += prev[2]
+ ct += prev[3]
+ callers[func] = nc, cc, tt, ct
+
+ # The following two methods can be called by clients to use
+ # a profiler to profile a statement, given as a string.
+
+ def run(self, cmd):
+ import __main__
+ dict = __main__.__dict__
+ return self.runctx(cmd, dict, dict)
+
+ def runctx(self, cmd, globals, locals):
+ self.enable()
+ try:
+ exec(cmd, globals, locals)
+ finally:
+ self.disable()
+ return self
+
+ # This method is more useful to profile a single function call.
+ def runcall(self, func, /, *args, **kw):
+ self.enable()
+ try:
+ return func(*args, **kw)
+ finally:
+ self.disable()
+
+ def __enter__(self):
+ self.enable()
+ return self
+
+ def __exit__(self, *exc_info):
+ self.disable()
+
+# ____________________________________________________________
+
+def label(code):
+ if isinstance(code, str):
+ return ('~', 0, code) # built-in functions ('~' sorts at the end)
+ else:
+ return (code.co_filename, code.co_firstlineno, code.co_name)
+
+# ____________________________________________________________
+
+def main():
+ import os
+ import sys
+ import runpy
+ import pstats
+ from optparse import OptionParser
+ usage = "cProfile.py [-o output_file_path] [-s sort] [-m module | scriptfile] [arg] ..."
+ parser = OptionParser(usage=usage)
+ parser.allow_interspersed_args = False
+ parser.add_option('-o', '--outfile', dest="outfile",
+ help="Save stats to <outfile>", default=None)
+ parser.add_option('-s', '--sort', dest="sort",
+ help="Sort order when printing to stdout, based on pstats.Stats class",
+ default=2,
+ choices=sorted(pstats.Stats.sort_arg_dict_default))
+ parser.add_option('-m', dest="module", action="store_true",
+ help="Profile a library module", default=False)
+
+ if not sys.argv[1:]:
+ parser.print_usage()
+ sys.exit(2)
+
+ (options, args) = parser.parse_args()
+ sys.argv[:] = args
+
+ # The script that we're profiling may chdir, so capture the absolute path
+ # to the output file at startup.
+ if options.outfile is not None:
+ options.outfile = os.path.abspath(options.outfile)
+
+ if len(args) > 0:
+ if options.module:
+ code = "run_module(modname, run_name='__main__')"
+ globs = {
+ 'run_module': runpy.run_module,
+ 'modname': args[0]
+ }
+ else:
+ progname = args[0]
+ sys.path.insert(0, os.path.dirname(progname))
+ with io.open_code(progname) as fp:
+ code = compile(fp.read(), progname, 'exec')
+ spec = importlib.machinery.ModuleSpec(name='__main__', loader=None,
+ origin=progname)
+ module = importlib.util.module_from_spec(spec)
+ # Set __main__ so that importing __main__ in the profiled code will
+ # return the same namespace that the code is executing under.
+ sys.modules['__main__'] = module
+ # Ensure that we're using the same __dict__ instance as the module
+ # for the global variables so that updates to globals are reflected
+ # in the module's namespace.
+ globs = module.__dict__
+ globs.update({
+ '__spec__': spec,
+ '__file__': spec.origin,
+ '__name__': spec.name,
+ '__package__': None,
+ '__cached__': None,
+ })
+
+ try:
+ runctx(code, globs, None, options.outfile, options.sort)
+ except BrokenPipeError as exc:
+ # Prevent "Exception ignored" during interpreter shutdown.
+ sys.stdout = None
+ sys.exit(exc.errno)
+ else:
+ parser.print_usage()
+ return parser
+
+# When invoked as main program, invoke the profiler on a script
+if __name__ == '__main__':
+ main()
-"""Tests for the sampling profiler (profile.sample)."""
+"""Tests for the sampling profiler (profiling.sampling)."""
import contextlib
import io
import unittest
from unittest import mock
-from profile.pstats_collector import PstatsCollector
-from profile.stack_collector import (
+from profiling.sampling.pstats_collector import PstatsCollector
+from profiling.sampling.stack_collector import (
CollapsedStackCollector,
)
"Test only runs when _remote_debugging is available"
)
else:
- import profile.sample
- from profile.sample import SampleProfiler
+ import profiling.sampling
+ from profiling.sampling.sample import SampleProfiler
def test_sample_profiler_initialization(self):
"""Test SampleProfiler initialization with various parameters."""
- from profile.sample import SampleProfiler
+ from profiling.sampling.sample import SampleProfiler
# Mock RemoteUnwinder to avoid permission issues
with mock.patch(
def test_sample_profiler_sample_method_timing(self):
"""Test that the sample method respects duration and handles timing correctly."""
- from profile.sample import SampleProfiler
+ from profiling.sampling.sample import SampleProfiler
# Mock the unwinder to avoid needing a real process
mock_unwinder = mock.MagicMock()
def test_sample_profiler_error_handling(self):
"""Test that the sample method handles errors gracefully."""
- from profile.sample import SampleProfiler
+ from profiling.sampling.sample import SampleProfiler
# Mock unwinder that raises errors
mock_unwinder = mock.MagicMock()
def test_sample_profiler_missed_samples_warning(self):
"""Test that the profiler warns about missed samples when sampling is too slow."""
- from profile.sample import SampleProfiler
+ from profiling.sampling.sample import SampleProfiler
mock_unwinder = mock.MagicMock()
mock_unwinder.get_stack_trace.return_value = [
def test_print_sampled_stats_basic(self):
"""Test basic print_sampled_stats functionality."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Capture output
with io.StringIO() as output:
def test_print_sampled_stats_sorting(self):
"""Test different sorting options."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Test sort by calls
with io.StringIO() as output:
def test_print_sampled_stats_limit(self):
"""Test limiting output rows."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
with io.StringIO() as output:
with mock.patch("sys.stdout", output):
def test_print_sampled_stats_time_units(self):
"""Test proper time unit selection."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
with io.StringIO() as output:
with mock.patch("sys.stdout", output):
def test_print_sampled_stats_summary(self):
"""Test summary section generation."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
with io.StringIO() as output:
with mock.patch("sys.stdout", output):
def test_print_sampled_stats_no_summary(self):
"""Test disabling summary output."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
with io.StringIO() as output:
with mock.patch("sys.stdout", output):
def test_print_sampled_stats_empty_stats(self):
"""Test with empty stats."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
empty_stats = mock.MagicMock()
empty_stats.stats = {}
def test_print_sampled_stats_sample_percentage_sorting(self):
"""Test sample percentage sorting options."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Add a function with high sample percentage (more direct calls than func3's 200)
self.mock_stats.stats[("expensive.py", 60, "expensive_func")] = (
def test_print_sampled_stats_with_recursive_calls(self):
"""Test print_sampled_stats with recursive calls where nc != cc."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Create stats with recursive calls (nc != cc)
recursive_stats = mock.MagicMock()
def test_print_sampled_stats_with_zero_call_counts(self):
"""Test print_sampled_stats with zero call counts to trigger division protection."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Create stats with zero call counts
zero_stats = mock.MagicMock()
def test_print_sampled_stats_sort_by_name(self):
"""Test sort by function name option."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
with io.StringIO() as output:
with mock.patch("sys.stdout", output):
def test_print_sampled_stats_with_zero_time_functions(self):
"""Test summary sections with functions that have zero time."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Create stats with zero-time functions
zero_time_stats = mock.MagicMock()
def test_print_sampled_stats_with_malformed_qualified_names(self):
"""Test summary generation with function names that don't contain colons."""
- from profile.sample import print_sampled_stats
+ from profiling.sampling.sample import print_sampled_stats
# Create stats with function names that would create malformed qualified names
malformed_stats = mock.MagicMock()
mock.patch("sys.stdout", captured_output),
):
try:
- profile.sample.sample(
+ profiling.sampling.sample.sample(
proc.pid,
duration_sec=2,
sample_interval_usec=1000, # 1ms
mock.patch("sys.stdout", captured_output),
):
try:
- profile.sample.sample(
+ profiling.sampling.sample.sample(
proc.pid,
duration_sec=1,
filename=pstats_out.name,
mock.patch("sys.stdout", captured_output),
):
try:
- profile.sample.sample(
+ profiling.sampling.sample.sample(
proc.pid,
duration_sec=1,
filename=collapsed_file.name,
mock.patch("sys.stdout", captured_output),
):
try:
- profile.sample.sample(
+ profiling.sampling.sample.sample(
proc.pid,
duration_sec=1,
all_threads=True,
script_file.flush()
self.addCleanup(close_and_unlink, script_file)
- test_args = ["profile.sample", "-d", "1", script_file.name]
+ test_args = ["profiling.sampling.sample", "-d", "1", script_file.name]
with (
mock.patch("sys.argv", test_args),
mock.patch("sys.stdout", captured_output),
):
try:
- profile.sample.main()
+ profiling.sampling.sample.main()
except PermissionError:
self.skipTest("Insufficient permissions for remote profiling")
with open(module_path, "w") as f:
f.write(self.test_script)
- test_args = ["profile.sample", "-d", "1", "-m", "test_module"]
+ test_args = ["profiling.sampling.sample", "-d", "1", "-m", "test_module"]
with (
mock.patch("sys.argv", test_args),
contextlib.chdir(tempdir.name),
):
try:
- profile.sample.main()
+ profiling.sampling.sample.main()
except PermissionError:
self.skipTest("Insufficient permissions for remote profiling")
class TestSampleProfilerErrorHandling(unittest.TestCase):
def test_invalid_pid(self):
with self.assertRaises((OSError, RuntimeError)):
- profile.sample.sample(-1, duration_sec=1)
+ profiling.sampling.sample.sample(-1, duration_sec=1)
def test_process_dies_during_sampling(self):
with test_subprocess("import time; time.sleep(0.5); exit()") as proc:
mock.patch("sys.stdout", captured_output),
):
try:
- profile.sample.sample(
+ profiling.sampling.sample.sample(
proc.pid,
duration_sec=2, # Longer than process lifetime
sample_interval_usec=50000,
def test_invalid_output_format(self):
with self.assertRaises(ValueError):
- profile.sample.sample(
+ profiling.sampling.sample.sample(
os.getpid(),
duration_sec=1,
output_format="invalid_format",
def test_invalid_output_format_with_mocked_profiler(self):
"""Test invalid output format with proper mocking to avoid permission issues."""
with mock.patch(
- "profile.sample.SampleProfiler"
+ "profiling.sampling.sample.SampleProfiler"
) as mock_profiler_class:
mock_profiler = mock.MagicMock()
mock_profiler_class.return_value = mock_profiler
with self.assertRaises(ValueError) as cm:
- profile.sample.sample(
+ profiling.sampling.sample.sample(
12345,
duration_sec=1,
output_format="unknown_format",
coordinator_cmd = args[0]
self.assertEqual(coordinator_cmd[0], sys.executable)
self.assertEqual(coordinator_cmd[1], "-m")
- self.assertEqual(coordinator_cmd[2], "profile._sync_coordinator")
+ self.assertEqual(coordinator_cmd[2], "profiling.sampling._sync_coordinator")
self.assertEqual(coordinator_cmd[3], "12345") # port
# cwd is coordinator_cmd[4]
self.assertEqual(coordinator_cmd[5:], expected_target_args)
@unittest.skipIf(is_emscripten, "socket.SO_REUSEADDR does not exist")
def test_cli_module_argument_parsing(self):
- test_args = ["profile.sample", "-m", "mymodule"]
+ test_args = ["profiling.sampling.sample", "-m", "mymodule"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
self._setup_sync_mocks(mock_socket, mock_popen)
- profile.sample.main()
+ profiling.sampling.sample.main()
self._verify_coordinator_command(mock_popen, ("-m", "mymodule"))
mock_sample.assert_called_once_with(
@unittest.skipIf(is_emscripten, "socket.SO_REUSEADDR does not exist")
def test_cli_module_with_arguments(self):
- test_args = ["profile.sample", "-m", "mymodule", "arg1", "arg2", "--flag"]
+ test_args = ["profiling.sampling.sample", "-m", "mymodule", "arg1", "arg2", "--flag"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
self._setup_sync_mocks(mock_socket, mock_popen)
- profile.sample.main()
+ profiling.sampling.sample.main()
self._verify_coordinator_command(mock_popen, ("-m", "mymodule", "arg1", "arg2", "--flag"))
mock_sample.assert_called_once_with(
@unittest.skipIf(is_emscripten, "socket.SO_REUSEADDR does not exist")
def test_cli_script_argument_parsing(self):
- test_args = ["profile.sample", "myscript.py"]
+ test_args = ["profiling.sampling.sample", "myscript.py"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
self._setup_sync_mocks(mock_socket, mock_popen)
- profile.sample.main()
+ profiling.sampling.sample.main()
self._verify_coordinator_command(mock_popen, ("myscript.py",))
mock_sample.assert_called_once_with(
@unittest.skipIf(is_emscripten, "socket.SO_REUSEADDR does not exist")
def test_cli_script_with_arguments(self):
- test_args = ["profile.sample", "myscript.py", "arg1", "arg2", "--flag"]
+ test_args = ["profiling.sampling.sample", "myscript.py", "arg1", "arg2", "--flag"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
# Override specific behavior for this test
mock_process.wait.side_effect = [subprocess.TimeoutExpired(test_args, 0.1), None]
- profile.sample.main()
+ profiling.sampling.sample.main()
# Verify the coordinator command was called
args, kwargs = mock_popen.call_args
coordinator_cmd = args[0]
self.assertEqual(coordinator_cmd[0], sys.executable)
self.assertEqual(coordinator_cmd[1], "-m")
- self.assertEqual(coordinator_cmd[2], "profile._sync_coordinator")
+ self.assertEqual(coordinator_cmd[2], "profiling.sampling._sync_coordinator")
self.assertEqual(coordinator_cmd[3], "12345") # port
# cwd is coordinator_cmd[4]
self.assertEqual(coordinator_cmd[5:], ("myscript.py", "arg1", "arg2", "--flag"))
def test_cli_mutually_exclusive_pid_module(self):
- test_args = ["profile.sample", "-p", "12345", "-m", "mymodule"]
+ test_args = ["profiling.sampling.sample", "-p", "12345", "-m", "mymodule"]
with (
mock.patch("sys.argv", test_args),
mock.patch("sys.stderr", io.StringIO()) as mock_stderr,
self.assertRaises(SystemExit) as cm,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
self.assertEqual(cm.exception.code, 2) # argparse error
error_msg = mock_stderr.getvalue()
self.assertIn("not allowed with argument", error_msg)
def test_cli_mutually_exclusive_pid_script(self):
- test_args = ["profile.sample", "-p", "12345", "myscript.py"]
+ test_args = ["profiling.sampling.sample", "-p", "12345", "myscript.py"]
with (
mock.patch("sys.argv", test_args),
mock.patch("sys.stderr", io.StringIO()) as mock_stderr,
self.assertRaises(SystemExit) as cm,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
self.assertEqual(cm.exception.code, 2) # argparse error
error_msg = mock_stderr.getvalue()
self.assertIn("only one target type can be specified", error_msg)
def test_cli_no_target_specified(self):
- test_args = ["profile.sample", "-d", "5"]
+ test_args = ["profiling.sampling.sample", "-d", "5"]
with (
mock.patch("sys.argv", test_args),
mock.patch("sys.stderr", io.StringIO()) as mock_stderr,
self.assertRaises(SystemExit) as cm,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
self.assertEqual(cm.exception.code, 2) # argparse error
error_msg = mock_stderr.getvalue()
@unittest.skipIf(is_emscripten, "socket.SO_REUSEADDR does not exist")
def test_cli_module_with_profiler_options(self):
test_args = [
- "profile.sample", "-i", "1000", "-d", "30", "-a",
+ "profiling.sampling.sample", "-i", "1000", "-d", "30", "-a",
"--sort-tottime", "-l", "20", "-m", "mymodule",
]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
self._setup_sync_mocks(mock_socket, mock_popen)
- profile.sample.main()
+ profiling.sampling.sample.main()
self._verify_coordinator_command(mock_popen, ("-m", "mymodule"))
mock_sample.assert_called_once_with(
def test_cli_script_with_profiler_options(self):
"""Test script with various profiler options."""
test_args = [
- "profile.sample", "-i", "2000", "-d", "60",
+ "profiling.sampling.sample", "-i", "2000", "-d", "60",
"--collapsed", "-o", "output.txt",
"myscript.py", "scriptarg",
]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
self._setup_sync_mocks(mock_socket, mock_popen)
- profile.sample.main()
+ profiling.sampling.sample.main()
self._verify_coordinator_command(mock_popen, ("myscript.py", "scriptarg"))
# Verify profiler options were passed correctly
)
def test_cli_empty_module_name(self):
- test_args = ["profile.sample", "-m"]
+ test_args = ["profiling.sampling.sample", "-m"]
with (
mock.patch("sys.argv", test_args),
mock.patch("sys.stderr", io.StringIO()) as mock_stderr,
self.assertRaises(SystemExit) as cm,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
self.assertEqual(cm.exception.code, 2) # argparse error
error_msg = mock_stderr.getvalue()
@unittest.skipIf(is_emscripten, "socket.SO_REUSEADDR does not exist")
def test_cli_long_module_option(self):
- test_args = ["profile.sample", "--module", "mymodule", "arg1"]
+ test_args = ["profiling.sampling.sample", "--module", "mymodule", "arg1"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
mock.patch("subprocess.Popen") as mock_popen,
mock.patch("socket.socket") as mock_socket,
):
self._setup_sync_mocks(mock_socket, mock_popen)
- profile.sample.main()
+ profiling.sampling.sample.main()
self._verify_coordinator_command(mock_popen, ("-m", "mymodule", "arg1"))
def test_cli_complex_script_arguments(self):
test_args = [
- "profile.sample", "script.py",
+ "profiling.sampling.sample", "script.py",
"--input", "file.txt", "-v", "--output=/tmp/out", "positional"
]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
- mock.patch("profile.sample._run_with_sync") as mock_run_with_sync,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample._run_with_sync") as mock_run_with_sync,
):
mock_process = mock.MagicMock()
mock_process.pid = 12345
mock_process.poll.return_value = None
mock_run_with_sync.return_value = mock_process
- profile.sample.main()
+ profiling.sampling.sample.main()
mock_run_with_sync.assert_called_once_with((
sys.executable, "script.py",
test_cases = [
# Test sort options are invalid with collapsed
(
- ["profile.sample", "--collapsed", "--sort-nsamples", "-p", "12345"],
+ ["profiling.sampling.sample", "--collapsed", "--sort-nsamples", "-p", "12345"],
"sort",
),
(
- ["profile.sample", "--collapsed", "--sort-tottime", "-p", "12345"],
+ ["profiling.sampling.sample", "--collapsed", "--sort-tottime", "-p", "12345"],
"sort",
),
(
[
- "profile.sample",
+ "profiling.sampling.sample",
"--collapsed",
"--sort-cumtime",
"-p",
),
(
[
- "profile.sample",
+ "profiling.sampling.sample",
"--collapsed",
"--sort-sample-pct",
"-p",
),
(
[
- "profile.sample",
+ "profiling.sampling.sample",
"--collapsed",
"--sort-cumul-pct",
"-p",
"sort",
),
(
- ["profile.sample", "--collapsed", "--sort-name", "-p", "12345"],
+ ["profiling.sampling.sample", "--collapsed", "--sort-name", "-p", "12345"],
"sort",
),
# Test limit option is invalid with collapsed
- (["profile.sample", "--collapsed", "-l", "20", "-p", "12345"], "limit"),
+ (["profiling.sampling.sample", "--collapsed", "-l", "20", "-p", "12345"], "limit"),
(
- ["profile.sample", "--collapsed", "--limit", "20", "-p", "12345"],
+ ["profiling.sampling.sample", "--collapsed", "--limit", "20", "-p", "12345"],
"limit",
),
# Test no-summary option is invalid with collapsed
(
- ["profile.sample", "--collapsed", "--no-summary", "-p", "12345"],
+ ["profiling.sampling.sample", "--collapsed", "--no-summary", "-p", "12345"],
"summary",
),
]
mock.patch("sys.stderr", io.StringIO()) as mock_stderr,
self.assertRaises(SystemExit) as cm,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
self.assertEqual(cm.exception.code, 2) # argparse error code
error_msg = mock_stderr.getvalue()
def test_cli_default_collapsed_filename(self):
"""Test that collapsed format gets a default filename when not specified."""
- test_args = ["profile.sample", "--collapsed", "-p", "12345"]
+ test_args = ["profiling.sampling.sample", "--collapsed", "-p", "12345"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
# Check that filename was set to default collapsed format
mock_sample.assert_called_once()
"""Test custom output filenames for both formats."""
test_cases = [
(
- ["profile.sample", "--pstats", "-o", "custom.pstats", "-p", "12345"],
+ ["profiling.sampling.sample", "--pstats", "-o", "custom.pstats", "-p", "12345"],
"custom.pstats",
"pstats",
),
(
- ["profile.sample", "--collapsed", "-o", "custom.txt", "-p", "12345"],
+ ["profiling.sampling.sample", "--collapsed", "-o", "custom.txt", "-p", "12345"],
"custom.txt",
"collapsed",
),
for test_args, expected_filename, expected_format in test_cases:
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
mock_sample.assert_called_once()
call_args = mock_sample.call_args[1]
def test_cli_missing_required_arguments(self):
"""Test that CLI requires PID argument."""
with (
- mock.patch("sys.argv", ["profile.sample"]),
+ mock.patch("sys.argv", ["profiling.sampling.sample"]),
mock.patch("sys.stderr", io.StringIO()),
):
with self.assertRaises(SystemExit):
- profile.sample.main()
+ profiling.sampling.sample.main()
def test_cli_mutually_exclusive_format_options(self):
"""Test that pstats and collapsed options are mutually exclusive."""
with (
mock.patch(
"sys.argv",
- ["profile.sample", "--pstats", "--collapsed", "-p", "12345"],
+ ["profiling.sampling.sample", "--pstats", "--collapsed", "-p", "12345"],
),
mock.patch("sys.stderr", io.StringIO()),
):
with self.assertRaises(SystemExit):
- profile.sample.main()
+ profiling.sampling.sample.main()
def test_argument_parsing_basic(self):
- test_args = ["profile.sample", "-p", "12345"]
+ test_args = ["profiling.sampling.sample", "-p", "12345"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
mock_sample.assert_called_once_with(
12345,
]
for option, expected_sort_value in sort_options:
- test_args = ["profile.sample", option, "-p", "12345"]
+ test_args = ["profiling.sampling.sample", option, "-p", "12345"]
with (
mock.patch("sys.argv", test_args),
- mock.patch("profile.sample.sample") as mock_sample,
+ mock.patch("profiling.sampling.sample.sample") as mock_sample,
):
- profile.sample.main()
+ profiling.sampling.sample.main()
mock_sample.assert_called_once()
call_args = mock_sample.call_args[1]
projects / thirdparty / Python / cpython.git / commitdiff
