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

rust-matchers: add `Matcher` trait and implement `AlwaysMatcher`...

rust-matchers: add `Matcher` trait and implement `AlwaysMatcher` In our quest of a faster Mercurial, we have arrived at the point where we need to implement the matchers in Rust. This RFC mainly for the `Matcher` trait to see if the changes proposed feel fine to people with more experience on the matter. While the `AlwaysMatcher` implementation is here as a trivial example, it should be the first step towards matchers use in Rust as it is currently the only supported one. Notable changes: - `exact` is renamed to `exact_match` - enums for `visit*` methods with `Recursive` instead of `'all'`, etc. - a new `roots`, separate from `file_set` - no `bad`, `explicitdir` or `traversedir` functions as they can be passed to the high functions instead of the matchers Thanks to Martin for suggesting the last two (most important) changes and for reaching out to help a few weeks ago. Differential Revision: https://phab.mercurial-scm.org/D7178

Gregory Szorc - - Load All Authors

File last commit:

r43487:6a350194 default


                r43742:a77d4fe3

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))
				)