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

posix: always seek to EOF when opening a file in append mode...

posix: always seek to EOF when opening a file in append mode Python 3 already does this, so skip it there. Consider the program: #include <stdio.h> int main() { FILE *f = fopen("narf", "w"); fprintf(f, "narf\n"); fclose(f); f = fopen("narf", "a"); printf("%ld\n", ftell(f)); fprintf(f, "troz\n"); printf("%ld\n", ftell(f)); return 0; } on macOS, FreeBSD, and Linux with glibc, this program prints 5 10 but on musl libc (Alpine Linux and probably others) this prints 0 10 By my reading of https://pubs.opengroup.org/onlinepubs/009695399/functions/fopen.html this is technically correct, specifically: > Opening a file with append mode (a as the first character in the > mode argument) shall cause all subsequent writes to the file to be > forced to the then current end-of-file, regardless of intervening > calls to fseek(). in other words, the file position doesn't really matter in append-mode files, and we can't depend on it being at all meaningful unless we perform a seek() before tell() after open(..., 'a'). Experimentally after a .write() we can do a .tell() and it'll always be reasonable, but I'm unclear from reading the specification if that's a smart thing to rely on. This matches what we do on Windows and what Python 3 does for free, so let's just be consistent. Thanks to Yuya for the idea.

Gregory Szorc - - Load All Authors

File last commit:

r42465:f3018448 default


                r42778:97ada9b8

5.0.2 stable

Download file

             winrm.py
        
                    82 lines
            
             | 1.9 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=120, 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_cmd('echo "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 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=120, 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_cmd('echo "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 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)))