upstream/mercurial-mirror Files · contrib/automation/hgautomation/winrm.py

pycompat: use os.fsencode() to re-encode sys.argv...

pycompat: use os.fsencode() to re-encode sys.argv Historically, the previous code made sense, as Py_EncodeLocale() and fs.fsencode() could possibly use different encodings. However, this is not the case anymore for Python 3.2, which uses the locale encoding as the filesystem encoding (this is not true for later Python versions, but see below). See https://vstinner.github.io/painful-history-python-filesystem-encoding.html for a source and more background information. Using os.fsencode() is safer, as the documentation for sys.argv says that it can be used to get the original bytes. When doing further changes, the Python developers will take care that this continues to work. One concrete case where os.fsencode() is more correct is when enabling Python's UTF-8 mode. Py_DecodeLocale() will use UTF-8 in this case. Our previous code would have encoded it using the locale encoding (which might be different), whereas os.fsencode() will encode it with UTF-8. Since we don’t claim to support the UTF-8 mode, this is not really a bug and the patch can go to the default branch. It might be a good idea to not commit this to the stable branch, as it could in theory introduce regressions.

Gregory Szorc - - Load All Authors

File last commit:

r43487:6a350194 default


                r45533:f2de8f31

default

Download file

             winrm.py
        
                    87 lines
            
             | 2.0 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / contrib / automation / hgautomation / winrm.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      # winrm.py - Interact with Windows Remote Management (WinRM)

      #

      # Copyright 2019 Gregory Szorc <gregory.szorc@gmail.com>

      #

      # This software may be used and distributed according to the terms of the

      # GNU General Public License version 2 or any later version.

      # no-check-code because Python 3 native.

      import logging

      import pprint

      import time

      from pypsrp.client import Client

      from pypsrp.powershell import (

          PowerShell,

          PSInvocationState,

          RunspacePool,

      )

      import requests.exceptions

      logger = logging.getLogger(__name__)

      def wait_for_winrm(host, username, password, timeout=180, ssl=False):

          """Wait for the Windows Remoting (WinRM) service to become available.

          Returns a ``psrpclient.Client`` instance.

          """

          end_time = time.time() + timeout

          while True:

              try:

                  client = Client(

                      host,

                      username=username,

                      password=password,

                      ssl=ssl,

                      connection_timeout=5,

                  )

                  client.execute_ps("Write-Host 'Hello, World!'")

                  return client

              except requests.exceptions.ConnectionError:

                  if time.time() >= end_time:

                      raise

                  time.sleep(1)

      def format_object(o):

          if isinstance(o, str):

              return o

          try:

              o = str(o)

          except (AttributeError, TypeError):

              o = pprint.pformat(o.extended_properties)

          return o

      def run_powershell(client, script):

          with RunspacePool(client.wsman) as pool:

              ps = PowerShell(pool)

              ps.add_script(script)

              ps.begin_invoke()

              while ps.state == PSInvocationState.RUNNING:

                  ps.poll_invoke()

                  for o in ps.output:

                      print(format_object(o))

                  ps.output[:] = []

              ps.end_invoke()

              for o in ps.output:

                  print(format_object(o))

              if ps.state == PSInvocationState.FAILED:

                  raise Exception(

                      'PowerShell execution failed: %s'

                      % ' '.join(map(format_object, ps.streams.error))

                  )

	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

				# winrm.py - Interact with Windows Remote Management (WinRM)
				#
				# Copyright 2019 Gregory Szorc <gregory.szorc@gmail.com>
				#
				# This software may be used and distributed according to the terms of the
				# GNU General Public License version 2 or any later version.

				# no-check-code because Python 3 native.

				import logging
				import pprint
				import time

				from pypsrp.client import Client
				from pypsrp.powershell import (
				PowerShell,
				PSInvocationState,
				RunspacePool,
				)
				import requests.exceptions


				logger = logging.getLogger(__name__)


				def wait_for_winrm(host, username, password, timeout=180, ssl=False):
				"""Wait for the Windows Remoting (WinRM) service to become available.

				Returns a ``psrpclient.Client`` instance.
				"""

				end_time = time.time() + timeout

				while True:
				try:
				client = Client(
				host,
				username=username,
				password=password,
				ssl=ssl,
				connection_timeout=5,
				)
				client.execute_ps("Write-Host 'Hello, World!'")
				return client
				except requests.exceptions.ConnectionError:
				if time.time() >= end_time:
				raise

				time.sleep(1)


				def format_object(o):
				if isinstance(o, str):
				return o

				try:
				o = str(o)
				except (AttributeError, TypeError):
				o = pprint.pformat(o.extended_properties)

				return o


				def run_powershell(client, script):
				with RunspacePool(client.wsman) as pool:
				ps = PowerShell(pool)
				ps.add_script(script)

				ps.begin_invoke()

				while ps.state == PSInvocationState.RUNNING:
				ps.poll_invoke()
				for o in ps.output:
				print(format_object(o))

				ps.output[:] = []

				ps.end_invoke()

				for o in ps.output:
				print(format_object(o))

				if ps.state == PSInvocationState.FAILED:
				raise Exception(
				'PowerShell execution failed: %s'
				% ' '.join(map(format_object, ps.streams.error))
				)