contrib/scripts/gnunet_pyexpect.py.in

   1 #!@PYTHONEXE@
   2 #    This file is part of GNUnet.
   3 #    (C) 2010, 2018 Christian Grothoff (and other contributing authors)
   4 #
   5 #    GNUnet is free software: you can redistribute it and/or modify it
   6 #    under the terms of the GNU Affero General Public License as published
   7 #    by the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    GNUnet is distributed in the hope that it will be useful, but
  11 #    WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13 #    Affero General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU Affero General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 #    SPDX-License-Identifier: AGPL3.0-or-later
  19 #
  20 # Testcase for gnunet-peerinfo
  21
  22 import os
  23 import re
  24 import subprocess
  25 import sys
  26 import shutil
  27 import time
  28
  29
  30 class pexpect(object):
  31     def __init__(self):
  32         super(pexpect, self).__init__()
  33
  34     def spawn(self, stdin, arglist, *pargs, **kwargs):
  35         env = kwargs.pop('env', None)
  36         if env is None:
  37             env = os.environ.copy()
  38         # This messes up some testcases, disable log redirection
  39         env.pop('GNUNET_FORCE_LOGFILE', None)
  40         self.proc = subprocess.Popen(arglist, *pargs, env=env, **kwargs)
  41         if self.proc is None:
  42             print("Failed to spawn a process {0}".format(arglist))
  43             sys.exit(1)
  44         if stdin is not None:
  45             self.stdo, self.stde = self.proc.communicate(stdin)
  46         else:
  47             self.stdo, self.stde = self.proc.communicate()
  48         return self.proc
  49
  50     def expect(self, s, r, flags=0):
  51         stream = self.stdo if s == 'stdout' else self.stde
  52         if isinstance(r, str):
  53             if r == "EOF":
  54                 if len(stream) == 0:
  55                     return True
  56                 else:
  57                     print(
  58                         "Failed to find `{1}' in {0}, which is `{2}' ({3})".
  59                         format(s, r, stream, len(stream))
  60                     )
  61                     sys.exit(2)
  62             raise ValueError(
  63                 "Argument `r' should be an instance of re.RegexObject or a special string, but is `{0}'"
  64                 .format(r)
  65             )
  66         m = r.search(stream.decode(), flags)
  67         if not m:
  68             print(
  69                 "Failed to find `{1}' in {0}, which is is `{2}'".format(
  70                     s, r.pattern, stream
  71                 )
  72             )
  73             sys.exit(2)
  74         stream = stream[m.end():]
  75         if s == 'stdout':
  76             self.stdo = stream
  77         else:
  78             self.stde = stream
  79         return m
  80
  81     def read(self, s, size=-1):
  82         stream = self.stdo if s == 'stdout' else self.stde
  83         result = ""
  84         if size < 0:
  85             result = stream
  86             new_stream = ""
  87         else:
  88             result = stream[0:size]
  89             new_stream = stream[size:]
  90         if s == 'stdout':
  91             self.stdo = new_stream
  92         else:
  93             self.stde = new_stream
  94         return result