src/statistics/test_gnunet_statistics.py.in

   1 #!@PYTHON@
   2 from __future__ import print_function
   3 import os
   4 import sys
   5 import shutil
   6 import re
   7 import subprocess
   8 import time
   9
  10 if os.name == "nt":
  11     tmp = os.getenv("TEMP")
  12 elif None != os.environ.get("TMPDIR"):
  13     tmp = os.getenv("TMPDIR")
  14 elif None != os.environ.get("TMP"):
  15     tmp = os.getenv("TMP")
  16 else:
  17     tmp = "/tmp"
  18
  19 if os.name == 'nt':
  20     st = './gnunet-statistics.exe'
  21     arm = 'gnunet-arm.exe'
  22 else:
  23     st = './gnunet-statistics'
  24     arm = 'gnunet-arm'
  25
  26 run_st = [st, '-c', 'test_statistics_api_data.conf']
  27 run_arm = [arm, '-c', 'test_statistics_api_data.conf']
  28 debug = os.getenv('DEBUG')
  29 if debug:
  30     run_arm += [debug.split(' ')]
  31
  32
  33 def cleanup():
  34     shutil.rmtree(os.path.join(tmp, "gnunet/test-gnunet-statistics"), True)
  35
  36
  37 def sub_run(args, want_stdo=True, want_stde=False, nofail=False):
  38     if want_stdo:
  39         stdo = subprocess.PIPE
  40     else:
  41         stdo = None
  42     if want_stde:
  43         stde = subprocess.PIPE
  44     else:
  45         stde = None
  46     p = subprocess.Popen(args, stdout=stdo, stderr=stde)
  47     stdo, stde = p.communicate()
  48     if not nofail:
  49         if p.returncode != 0:
  50             sys.exit(p.returncode)
  51     return (p.returncode, stdo, stde)
  52
  53
  54 def fail(result):
  55     print(result)
  56     r_arm(['-e'], want_stdo=False)
  57     sys.exit(1)
  58
  59
  60 def r_arm(extra_args, **kw):
  61     rc, stdo, stde = sub_run(run_arm + extra_args, **kw)
  62     if rc != 0:
  63         fail("FAIL: error running {}".format(run_arm))
  64     return (rc, stdo, stde)
  65
  66
  67 def r_st(extra_args, normal=True, **kw):
  68     rc, stdo, stde = sub_run(run_st + extra_args, **kw)
  69     if normal:
  70         if rc != 0:
  71             fail("FAIL: error running {}".format(run_st))
  72     else:
  73         if rc == 0:
  74             fail("FAIL: expected error while running {}".format(run_st))
  75     return (rc, stdo, stde)
  76
  77
  78 def restart():
  79     print("Restarting service...")
  80     t = r_arm(['-k', 'statistics'])
  81     time.sleep(1)
  82     t = r_arm(['-i', 'statistics'])
  83     time.sleep(1)
  84
  85
  86 cleanup()
  87
  88 print("Preparing: Starting service...")
  89 t = r_arm(['-s'], want_stdo=False)
  90 time.sleep(1)
  91 t = r_arm(['-i', 'statistics'], want_stdo=False)
  92 time.sleep(1)
  93
  94 print("TEST: Bad argument checking...", end='')
  95 r_st(['-x'], normal=False, nofail=True, want_stdo=False, want_stde=True)
  96 print("PASS")
  97
  98 print("TEST: Set value...", end='')
  99 r_st(['-n', 'test', '-s', 'subsystem', b'42'], nofail=True, want_stdo=False)
 100 print("PASS")
 101
 102 print("TEST: Set another value...", end='')
 103 r_st(['-n', 'other', '-s', 'osystem', b'43'], nofail=True, want_stdo=False)
 104 print("PASS")
 105
 106 print("TEST: Viewing all stats...", end='')
 107 rc, stdo, stde = r_st([], nofail=True, want_stdo=True)
 108 if len(stdo.splitlines()) != 2:
 109     fail("FAIL: unexpected output:\n{}".format(stdo))
 110 print("PASS")
 111
 112 print("TEST: Viewing stats by name...", end='')
 113 rc, stdo, stde = r_st(['-n', 'other'], nofail=True, want_stdo=True)
 114 if len([x for x in stdo.splitlines() if re.search(b'43', x)]) != 1:
 115     fail("FAIL: unexpected output:\n{}".format(stdo))
 116 print("PASS")
 117
 118 print("TEST: Viewing stats by subsystem...", end='')
 119 rc, stdo, stde = r_st(['-s', 'subsystem'], nofail=True, want_stdo=True)
 120 if len([x for x in stdo.splitlines() if re.search(b'42', x)]) != 1:
 121     fail("FAIL: unexpected output:\n{}".format(stdo))
 122 print("PASS")
 123
 124 print("TEST: Set persistent value...", end='')
 125 rc, stdo, stde = r_st(['-n', 'lasting', '-s', 'subsystem', '40', '-p'], nofail=True, want_stdo=False)
 126 rc, stdo, stde = r_st([], nofail=True, want_stdo=True)
 127 if len([x for x in stdo.splitlines() if re.search(b'40', x)]) != 1:
 128     fail("FAIL: unexpected output:\n{}".format(stdo))
 129 print("PASS")
 130
 131 restart()
 132
 133 print("TEST: Checking persistence...", end='')
 134 rc, stdo, stde = r_st([], nofail=True, want_stdo=True)
 135 if len([x for x in stdo.splitlines() if re.search(b'40', x)]) != 1:
 136     fail("FAIL: unexpected output:\n{}".format(stdo))
 137 print("PASS")
 138
 139 print("TEST: Removing persistence...", end='')
 140 rc, stdo, stde = r_st(['-n', 'lasting', '-s', 'subsystem', '40'], nofail=True, want_stdo=False)
 141 rc, stdo, stde = r_st([], nofail=True, want_stdo=True)
 142 if len([x for x in stdo.splitlines() if re.search(b'!', x)]) != 0:
 143     fail("FAIL: unexpected output:\n{}".format(stdo))
 144 print("PASS")
 145
 146 restart()
 147
 148 print("TEST: Checking removed persistence...", end='')
 149 rc, stdo, stde = r_st([], nofail=True, want_stdo=True)
 150 if len([x for x in stdo.splitlines() if re.search(b'40', x)]) != 0:
 151     fail("FAIL: unexpected output:\n{}".format(stdo))
 152 print("PASS")
 153
 154 print("Stopping service...")
 155 t = r_arm(['-e'], want_stdo=False)
 156 time.sleep(1)
 157
 158 cleanup()