Inheritance diagram for gnunet_testing.Peer:

Collaboration diagram for gnunet_testing.Peer:

Public Member Functions
def	__init__ (self, test, cfg_file)

def	__del__ (self)

def	start (self)

def	stop (self)

def	get_statistics_value (self, subsystem, name)

Data Fields
	id

	test

	started

	cfg

Detailed Description

Definition at line 278 of file gnunet_testing.py.

Constructor & Destructor Documentation

◆ init()

def gnunet_testing.Peer.__init__	(	self,
		test,
		cfg_file
	)

Definition at line 279 of file gnunet_testing.py.

     def __init__(self, test, cfg_file):
         if (False == os.path.isfile(cfg_file)):
             # print(("Peer cfg " + cfg_file + ": FILE NOT FOUND"))
             logger.debug('Peer cfg %s : FILE NOT FOUND', cfg_file)
         self.id = "<NaN>"
         self.test = test
         self.started = False
         self.cfg = cfg_file
  

◆ del()

def gnunet_testing.Peer.__del__ ( self )

Definition at line 288 of file gnunet_testing.py.

     def __del__(self):
         if (self.started == True):
             # print('ERROR! Peer using cfg ' + self.cfg + ' was not stopped')
             logger.debug('ERROR! Peer using cfg %s was not stopped', self.cfg)
             ret = self.stop()
             if (False == ret):
                 # print('ERROR! Peer using cfg ' +
                 #       self.cfg +
                 #       ' could not be stopped')
                 logger.debug(
                     'ERROR! Peer using cfg %s could not be stopped', self.cfg
                 )
                 self.started = False
             return ret
         else:
             return False
  

Here is the call graph for this function:

Member Function Documentation

◆ start()

def gnunet_testing.Peer.start ( self )

Definition at line 305 of file gnunet_testing.py.

     def start(self):
         os.unsetenv("XDG_CONFIG_HOME")
         os.unsetenv("XDG_DATA_HOME")
         os.unsetenv("XDG_CACHE_HOME")
         self.test.p("Starting peer using cfg " + self.cfg)
         try:
             server = subprocess.Popen([
                 self.test.gnunetarm, '-sq', '-c', self.cfg
             ])
             server.communicate()
         except OSError:
             # print("Can not start peer")
             logger.debug('Can not start peer')
             self.started = False
             return False
         self.started = True
         test = ''
         try:
             server = pexpect()
             server.spawn(
                 None, [self.test.gnunetpeerinfo, '-c', self.cfg, '-s'],
                 stdout=subprocess.PIPE,
                 stderr=subprocess.STDOUT
             )
             test = server.read("stdout", 1024)
         except OSError:
             # print("Can not get peer identity")
             logger.debug('Can not get peer identity')
         test = (test.split(b'`')[1])
         self.id = test.split(b'\'')[0]
         return True
  

◆ stop()

def gnunet_testing.Peer.stop ( self )

Definition at line 337 of file gnunet_testing.py.

     def stop(self):
         if (self.started == False):
             return False
         self.test.p("Stopping peer using cfg " + self.cfg)
         try:
             server = subprocess.Popen([
                 self.test.gnunetarm, '-eq', '-c', self.cfg
             ])
             server.communicate()
         except OSError:
             # print("Can not stop peer")
             logger.debug('Can not stop peer')
             return False
         self.started = False
         return True
  

Referenced by gnunet_testing.Peer.__del__().

Here is the caller graph for this function:

◆ get_statistics_value()

def gnunet_testing.Peer.get_statistics_value	(	self,
		subsystem,
		name
	)

Definition at line 353 of file gnunet_testing.py.

     def get_statistics_value(self, subsystem, name):
         server = pexpect()
         server.spawn(
             None, [
                 self.test.gnunetstatistics, '-c', self.cfg, '-q', '-n', name,
                 '-s', subsystem
             ],
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT
         )
         # server.expect ("stdout", re.compile (r""))
         test = server.read("stdout", 10240)
         tests = test.partition(b'\n')
         # On W32 GNUnet outputs with \r\n, rather than \n
         if os.name == 'nt' and tests[1] == b'\n' and tests[0][-1] == b'\r':
             tests = (tests[0][:-1], tests[1], tests[2])
         tests = tests[0]
         result = tests.decode("utf-8").strip()
         logger.debug(
             'running gnunet-statistics %s for %s "/" %s yields %s', self.cfg,
             name, subsystem, result
         )
         if (result.isdigit() == True):
             return result
         else:
             logger.debug(
                 'Invalid statistics value: %s is not a number!', result
             )
             return -1

Field Documentation

◆ id

gnunet_testing.Peer.id

Definition at line 283 of file gnunet_testing.py.

Referenced by gnunet_testing.Peer.start().

◆ test

gnunet_testing.Peer.test

Definition at line 284 of file gnunet_testing.py.

Referenced by gnunet_testing.Peer.get_statistics_value(), gnunet_testing.Peer.start(), and gnunet_testing.Peer.stop().

◆ started

gnunet_testing.Peer.started

Definition at line 285 of file gnunet_testing.py.

Referenced by gnunet_testing.Peer.__del__(), gnunet_testing.Peer.start(), and gnunet_testing.Peer.stop().

◆ cfg

gnunet_testing.Peer.cfg

Definition at line 286 of file gnunet_testing.py.

Referenced by gnunet_testing.Peer.__del__(), gnunet_testing.Peer.get_statistics_value(), gnunet_testing.Peer.start(), and gnunet_testing.Peer.stop().

The documentation for this class was generated from the following file:

src/integration-tests/gnunet_testing.py

Public Member Functions

Data Fields