upstream/mercurial-mirror Files · mercurial/testing/ps_util.py

interfaces: move peer `capabilities()` to the `ipeercapabilities` interface...

interfaces: move peer `capabilities()` to the `ipeercapabilities` interface I'm not sure why this was on the `ipeercommands` interface. It appears to be because these interfaces started out as `_basewirecommands` to hold wire commands, back in 558f5b2ee10e. The capabilities interface wasn't split out until 98861a2298b5, when it pulled the capability related methods off of the `ipeerbase` interface. Perhaps it was an oversight to not look at the commands interface because, while this is a wire command, both `sshpeer` and `httppeer` now perform a handshake while instantiating the peer object, and cache a fixed list of capabilities in that object. Likewise, `localpeer` is given a fixed set of capabilities when instantiated. Back in 558f5b2ee10e, `httppeer` looks like it issued a wire command when this method was called, but `sshpeer` obtained and cached the capabilities when instantiated, and this method always returned a fixed value. There's a perfectly good interface with other capability related methods, and having it here makes it easier to implement the base `peer` mixin class.

Matt Harbison - - Load All Authors

File last commit:

r53247:d4e30c15 default


                r53417:1554bd50

default

Download file

             ps_util.py
        
                    53 lines
            
             | 1.3 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / mercurial / testing / ps_util.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      # This python code can be imported into tests in order to terminate a process

      # with signal.SIGKILL on posix, or a roughly equivalent procedure on Windows.

      from __future__ import annotations

      import os

      import signal

      import subprocess

      import sys

      import tempfile

      from .. import (

          encoding,

          pycompat,

      )

      from ..utils import procutil

      def kill_nt(pid: int, exit_code: int):

          fd, pidfile = tempfile.mkstemp(

              prefix=b"sigkill-", dir=encoding.environ[b"HGTMP"], text=False

          )

          try:

              os.write(fd, b'%d\n' % pid)

          finally:

              os.close(fd)

          env = dict(encoding.environ)

          env[b"DAEMON_EXITCODE"] = b"%d" % exit_code

          # Simulate the message written to stderr for this process on non-Windows

          # platforms, for test consistency.

          print("Killed!", file=sys.stderr)

          subprocess.run(

              [

                  encoding.environ[b"PYTHON"],

                  b"%s/killdaemons.py"

                  % encoding.environ[b'RUNTESTDIR_FORWARD_SLASH'],

                  pidfile,

              ],

              env=procutil.tonativeenv(env),

          )

      def kill(pid: int):

          """Kill the process with the given PID with SIGKILL or equivalent."""

          if pycompat.iswindows:

              exit_code = 128 + 9

              kill_nt(pid, exit_code)

          else:

              os.kill(pid, signal.SIGKILL)

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				# This python code can be imported into tests in order to terminate a process
				# with signal.SIGKILL on posix, or a roughly equivalent procedure on Windows.

				from __future__ import annotations

				import os
				import signal
				import subprocess
				import sys
				import tempfile

				from .. import (
				encoding,
				pycompat,
				)

				from ..utils import procutil


				def kill_nt(pid: int, exit_code: int):
				fd, pidfile = tempfile.mkstemp(
				prefix=b"sigkill-", dir=encoding.environ[b"HGTMP"], text=False
				)
				try:
				os.write(fd, b'%d\n' % pid)
				finally:
				os.close(fd)

				env = dict(encoding.environ)
				env[b"DAEMON_EXITCODE"] = b"%d" % exit_code

				# Simulate the message written to stderr for this process on non-Windows
				# platforms, for test consistency.
				print("Killed!", file=sys.stderr)

				subprocess.run(
				[
				encoding.environ[b"PYTHON"],
				b"%s/killdaemons.py"
				% encoding.environ[b'RUNTESTDIR_FORWARD_SLASH'],
				pidfile,
				],
				env=procutil.tonativeenv(env),
				)


				def kill(pid: int):
				"""Kill the process with the given PID with SIGKILL or equivalent."""
				if pycompat.iswindows:
				exit_code = 128 + 9
				kill_nt(pid, exit_code)
				else:
				os.kill(pid, signal.SIGKILL)