lttng_ivc/utils/utils.py

   1 import signal
   2 import hashlib
   3 import os
   4 import time
   5 import socket
   6
   7 from contextlib import closing
   8
   9 def line_count(file_path):
  10     line_count = 0
  11     with open(file_path) as f:
  12         for line in f:
  13             line_count += 1
  14     return line_count
  15
  16
  17 def sha256_checksum(filename, block_size=65536):
  18     sha256 = hashlib.sha256()
  19     with open(filename, 'rb') as f:
  20         for block in iter(lambda: f.read(block_size), b''):
  21             sha256.update(block)
  22     return sha256.hexdigest()
  23
  24
  25 # TODO: timeout as a parameter or Settings
  26 # TODO: Custom exception
  27 def wait_for_file(path):
  28     i = 0
  29     timeout = 60
  30     while not os.path.exists(path):
  31         time.sleep(1)
  32         i = i + 1
  33         if i > timeout:
  34             raise Exception("File still does not exists. Timeout expired")
  35
  36
  37 # TODO: find better exception
  38 def create_empty_file(path):
  39     if os.path.exists(path):
  40         raise Exception("Path already exist")
  41     open(path, 'w').close()
  42
  43
  44 def __dummy_sigusr1_handler():
  45     pass
  46
  47
  48 def sessiond_spawn(runtime):
  49     agent_port = find_free_port()
  50     previous_handler = signal.signal(signal.SIGUSR1, __dummy_sigusr1_handler)
  51     sessiond = runtime.spawn_subprocess("lttng-sessiond -vvv -S --agent-tcp-port {}".format(agent_port))
  52     signal.sigtimedwait({signal.SIGUSR1}, 60)
  53     previous_handler = signal.signal(signal.SIGUSR1, previous_handler)
  54     return sessiond
  55
  56
  57 def find_free_port():
  58     # There is no guarantee that the port will be free at runtime but should be
  59     # good enough
  60     with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
  61         s.bind(('', 0))
  62         return s.getsockname()[1]
  63
  64
  65 def file_contains(stderr_file, list_of_string):
  66     with open(stderr_file, 'r') as stderr:
  67         for line in stderr:
  68             for s in list_of_string:
  69                 if s in line:
  70                     return True
  71
  72
  73 def find_dir(root, name):
  74     """
  75     Returns the absolute path or None.
  76     """
  77     abs_path = None
  78     for base, dirs, files in os.walk(root):
  79         for tmp in dirs:
  80             print(tmp)
  81             if tmp.endswith(name):
  82                 abs_path = os.path.abspath(os.path.join(base, tmp))
  83     return abs_path