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

grep: warn on censored revisions instead of erroring out...

grep: warn on censored revisions instead of erroring out We need most of the grep logic to go through in case we encounter a censored revision, so we just return a None body for a censored node, and we stop just short of trying to record matches with the contents of that censored body. The other parts such as recording that the censored file has been considered at this revision needs to go into the proper dicts. I have also gotten weary of all the abbreviations, so while I did a small refactor to move the file-data-getting operation into a common function, I also expanded the abbreviations of the relevant variables within this little function. Hopefully some day this helps someone figure out what all the abbreviations mean. Although the censoring docs currently state that some commands error out or are ignored depending on the `censor.policy` config, I cannot see a benefit for grep to ever stop dead in its tracks when a censored revision is encountered. I will also amend the docs to indicate that some commands, such as grep, unconditionally ignore censored revisions.

Gregory Szorc - - Load All Authors

File last commit:

r43487:6a350194 default


                r43622:13b8097d

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