fix ci

Author: maurycy

Lib/profiling/sampling/_control.py

@@ -90,8 +90,6 @@ def _close_listener(self):
                 created_stat.st_dev,
             ):
                 os.unlink(self._path)
-        except FileNotFoundError:
-            pass
         except OSError:
             pass
 

Lib/profiling/sampling/sample.py

@@ -225,9 +225,10 @@ def wait(timeout):
                 running_time_sec = time.perf_counter() - start_time
         except KeyboardInterrupt:
             interrupted = True
-            running_time_sec = time.perf_counter() - start_time
+            now = time.perf_counter()
+            running_time_sec = now - start_time
             if enabled_since is not None:
-                enabled_time_sec += time.perf_counter() - enabled_since
+                enabled_time_sec += now - enabled_since
             print("Interrupted by user.")
         finally:
             flush_pending()

Lib/test/test_profiling/test_sampling_profiler/test_control.py

@@ -2,34 +2,34 @@
 
 import io
 import os
-import shutil
 import socket
-import tempfile
 import time
 import unittest
 from unittest import mock
 
+from test.support import SHORT_TIMEOUT, os_helper, socket_helper
+
 try:
     from profiling.sampling._control import (
         ControlServer,
         _MAX_INBUF_BYTES,
         parse_control_uri,
     )
-    from profiling.sampling.cli import main
+    from profiling.sampling.cli import LiveStatsCollector, main
     from profiling.sampling.errors import ControlError, ControlURIError
 except ImportError:
     raise unittest.SkipTest(
         "Test only runs when profiling.sampling is available"
     )
 
 
+@socket_helper.skip_unless_bind_unix_socket
 class ControlServerTests(unittest.TestCase):
     """Tests for ControlServer protocol, lifecycle and CLI integration."""
 
     def setUp(self):
-        tmpdir = tempfile.mkdtemp()
-        self.addCleanup(shutil.rmtree, tmpdir, ignore_errors=True)
-        self.path = os.path.join(tmpdir, "control.sock")
+        self.path = socket_helper.create_unix_domain_name()
+        self.addCleanup(os_helper.unlink, self.path)
 
     def start_server(self):
         server = ControlServer(f"unix:{self.path}")
@@ -46,7 +46,7 @@ def connect(self):
 
     def request(self, server, client, command):
         client.sendall(command)
-        deadline = time.monotonic() + 1.0
+        deadline = time.monotonic() + SHORT_TIMEOUT
         while time.monotonic() < deadline:
             server.poll(timeout=0.05)
             try:
@@ -88,7 +88,7 @@ def test_start_creates_and_stop_unlinks_socket(self):
     def test_start_fails_on_occupied_path(self):
         """start() raises ControlError when the path is already bound."""
         squatter = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        squatter.bind(self.path)
+        socket_helper.bind_unix_socket(squatter, self.path)
         self.addCleanup(squatter.close)
         with self.assertRaisesRegex(ControlError, "failed to start"):
             ControlServer(f"unix:{self.path}").start()
@@ -124,7 +124,7 @@ def test_dispatch_quit_sets_running_and_closes(self):
         client = self.connect()
         self.assertEqual(self.request(server, client, b"quit\n"), b"ok\n")
         self.assertFalse(server.control.running)
-        deadline = time.monotonic() + 1.0
+        deadline = time.monotonic() + SHORT_TIMEOUT
         while time.monotonic() < deadline:
             server.poll(timeout=0.05)
             try:
@@ -142,7 +142,7 @@ def test_inbuf_overflow_drops_connection(self):
         client.connect(self.path)
         self.addCleanup(client.close)
         client.sendall(b"x" * (_MAX_INBUF_BYTES + 1))
-        deadline = time.monotonic() + 1.0
+        deadline = time.monotonic() + SHORT_TIMEOUT
         while time.monotonic() < deadline:
             server.poll(timeout=0.05)
             if not server._connections:
@@ -154,7 +154,7 @@ def test_close_listener_preserves_replaced_path(self):
         server = self.start_server()
         os.unlink(self.path)
         replacement = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        replacement.bind(self.path)
+        socket_helper.bind_unix_socket(replacement, self.path)
         self.addCleanup(replacement.close)
         server.stop()
         self.assertTrue(os.path.exists(self.path))
@@ -198,6 +198,8 @@ def test_cli_rejects_bad_uri(self):
         self.assertEqual(cm.exception.code, 2)
         self.assertIn("unsupported control URI scheme", stderr.getvalue())
 
+    @unittest.skipUnless(LiveStatsCollector is not None,
+                         "requires curses for --live")
     def test_cli_accepts_control_with_live(self):
         """--control and --live coexist after the mutex was dropped."""
         argv = [

Lib/test/test_profiling/test_sampling_profiler/test_integration.py

@@ -9,6 +9,7 @@
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 from unittest import mock
@@ -17,6 +18,7 @@
     import _remote_debugging
     import profiling.sampling
     import profiling.sampling.sample
+    from profiling.sampling._control import ControlServer
     from profiling.sampling.pstats_collector import PstatsCollector
     from profiling.sampling.stack_collector import CollapsedStackCollector
     from profiling.sampling.sample import SampleProfiler, _is_process_running
@@ -29,12 +31,13 @@
 from test.support import (
     requires_remote_subprocess_debugging,
     SHORT_TIMEOUT,
+    os_helper,
+    socket_helper,
 )
 
 from .helpers import (
     test_subprocess,
     close_and_unlink,
-    _cleanup_process,
 )
 from .mocks import MockFrameInfo, MockThreadInfo, MockInterpreterInfo
 
@@ -959,19 +962,20 @@ def test_all_stacks_share_same_base_frame(self):
 
 
 @requires_remote_subprocess_debugging()
+@socket_helper.skip_unless_bind_unix_socket
 @unittest.skipIf(
     sys.platform == "darwin" and os.geteuid() != 0,
     "macOS profiling requires elevated permissions",
 )
 class TestControlSocketIntegration(unittest.TestCase):
-    """End-to-end tests for the --control socket via a real profiler subprocess."""
+    """End-to-end tests for the --control socket via in-process sample()."""
 
     def _send_recv(self, client, request, expected_reply):
         client.sendall(request)
         self.assertEqual(client.recv(4096), expected_reply)
 
     def test_control_socket_disable_enable_quit_cycle(self):
-        """Drive disable/enable/quit through a real ControlServer subprocess."""
+        """Drive disable/enable/quit through a real ControlServer."""
         script = '''
 import time
 _test_sock.sendall(b"working")
@@ -980,27 +984,31 @@ def test_control_socket_disable_enable_quit_cycle(self):
     time.sleep(0.05)
 '''
         with test_subprocess(script, wait_for_working=True) as target:
-            tmpdir = tempfile.mkdtemp()
-            self.addCleanup(shutil.rmtree, tmpdir, ignore_errors=True)
-            socket_path = os.path.join(tmpdir, "tachyon.sock")
-
-            profiler = subprocess.Popen(
-                [sys.executable, "-m", "profiling.sampling", "attach",
-                 "--control", f"unix:{socket_path}",
-                 "-d", "30",
-                 str(target.process.pid)],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-            )
-            self.addCleanup(_cleanup_process, profiler)
+            socket_path = socket_helper.create_unix_domain_name()
+            self.addCleanup(os_helper.unlink, socket_path)
 
-            deadline = time.monotonic() + SHORT_TIMEOUT
-            while time.monotonic() < deadline:
-                if os.path.exists(socket_path):
-                    break
-                time.sleep(0.05)
-            else:
-                self.fail("profiler did not create the control socket")
+            server = ControlServer(f"unix:{socket_path}")
+            server.start()
+            self.addCleanup(server.stop)
+
+            exception = [None]
+
+            collector = PstatsCollector(sample_interval_usec=1000, skip_idle=False)
+
+            def sample_worker():
+                try:
+                    with contextlib.redirect_stdout(io.StringIO()):
+                        profiling.sampling.sample.sample(
+                            target.process.pid,
+                            collector,
+                            duration_sec=SHORT_TIMEOUT,
+                            control_server=server,
+                        )
+                except Exception as exc:
+                    exception[0] = exc
+
+            thread = threading.Thread(target=sample_worker, daemon=True)
+            thread.start()
 
             with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
                 client.settimeout(SHORT_TIMEOUT)
@@ -1010,9 +1018,7 @@ def test_control_socket_disable_enable_quit_cycle(self):
                 self._send_recv(client, b"enable\n", b"ok\n")
                 self._send_recv(client, b"quit\n", b"ok\n")
 
-            stdout, stderr = profiler.communicate(timeout=SHORT_TIMEOUT)
-            self.assertEqual(
-                profiler.returncode, 0,
-                msg=f"stdout={stdout!r} stderr={stderr!r}",
-            )
-            self.assertFalse(os.path.exists(socket_path))
+            thread.join(timeout=SHORT_TIMEOUT)
+            self.assertFalse(thread.is_alive(), "sample() did not exit on quit")
+            if exception[0] is not None:
+                raise exception[0]

Lib/test/test_profiling/test_sampling_profiler/test_profiler.py

@@ -210,7 +210,6 @@ def test_sample_profiler_does_not_buffer_non_aggregating_collectors(self):
 
         stack_frames = [mock.sentinel.stack_frames]
         mock_collector = mock.MagicMock()
-        mock_collector.aggregating = False
 
         with self._patched_unwinder() as u:
             u.instance.get_stack_trace.return_value = stack_frames
@@ -445,7 +444,6 @@ def test_sample_polls_control_server_periodically(self):
         """Test that sample() drives control_server.poll() during the loop."""
         control_server = self._fake_control_server()
         mock_collector = mock.MagicMock()
-        mock_collector.aggregating = False
         times = [1000.0 + i * 0.01 for i in range(60)]
         with self._patched_unwinder() as u:
             u.instance.get_stack_trace.return_value = []
@@ -466,7 +464,6 @@ def test_sample_breaks_when_control_running_false(self):
         """Test that sample() exits immediately when control.running is False."""
         control_server = self._fake_control_server(running=False)
         mock_collector = mock.MagicMock()
-        mock_collector.aggregating = False
         with self._patched_unwinder() as u:
             u.instance.get_stack_trace.return_value = []
             profiler = SampleProfiler(
@@ -485,7 +482,6 @@ def test_sample_pauses_when_control_disabled(self):
         """Test that disabled state stops sample collection."""
         control_server = self._fake_control_server(enabled=False)
         mock_collector = mock.MagicMock()
-        mock_collector.aggregating = False
         times = [1000.0 + i * 0.01 for i in range(60)]
         with self._patched_unwinder() as u:
             u.instance.get_stack_trace.return_value = []
@@ -514,7 +510,6 @@ def poll_side_effect(timeout=0):
         control_server.poll.side_effect = poll_side_effect
 
         mock_collector = mock.MagicMock()
-        mock_collector.aggregating = False
         times = [1000.0 + i * 0.01 for i in range(80)]
         with self._patched_unwinder() as u:
             u.instance.get_stack_trace.return_value = [
@@ -545,7 +540,6 @@ def poll_side_effect(timeout=0):
         control_server.poll.side_effect = poll_side_effect
 
         mock_collector = mock.MagicMock()
-        mock_collector.aggregating = False
         times = [1000.0 + i * 0.01 for i in range(120)]
         with self._patched_unwinder() as u:
             u.instance.get_stack_trace.return_value = [