def setup_suite():
  client_exec_location = runtime.get_active_config('client_exec_path')

  #Clients and server exec can also be in remote location.
  #if so specify the client_exec as <Remoteserver>:<path>
  #In that case you can also path a temp_scratch config for the sftp 
  #else /tmp folder used as default
  if (":" in client_exec_location):
    client_exec = client_exec_location
  else:
    client_exec = os.path.join(os.path.dirname(os.path.abspath(__file__)),
      client_exec_location)

  global client_deployer
  client_deployer = adhoc_deployer.SSHDeployer("AdditionClient",
      {'pid_keyword': "AdditionClient",
       'executable': client_exec,
       'start_command': runtime.get_active_config('client_start_command')})
  runtime.set_deployer("AdditionClient", client_deployer)

  client_deployer.install("client1",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('client_install_path') + 'client1'})

  client_deployer.install("client2",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('client_install_path') + 'client2'})

  server_exec_location = runtime.get_active_config('server_exec_path')


  if (":" in server_exec_location):
    server_exec = server_exec_location  
  else:
    server_exec = os.path.join(os.path.dirname(os.path.abspath(__file__)),
    runtime.get_active_config('server_exec_path'))

  global server_deployer
  server_deployer = adhoc_deployer.SSHDeployer("AdditionServer",
      {'pid_keyword': "AdditionServer",
       'executable': server_exec,
       'start_command': runtime.get_active_config('server_start_command')})
  runtime.set_deployer("AdditionServer", server_deployer)

  server_deployer.deploy("server1",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('server_install_path') + 'server1',
       "args": "localhost 8000".split()})

  server_deployer.deploy("server2",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('server_install_path') + 'server2',
       "args": "localhost 8001".split()})

  server_deployer.deploy("server3",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('server_install_path') + 'server3',
       "args": "localhost 8002".split()})

 def setup_suite(self):
   if os.path.isdir("/tmp/ztestsute"):
     shutil.rmtree("/tmp/ztestsuite")
   if not os.path.isdir(runtime.get_active_config("LOGS_DIRECTORY")):
     os.makedirs(runtime.get_active_config("LOGS_DIRECTORY"))
   if not os.path.isdir(runtime.get_active_config("OUTPUT_DIRECTORY")):
     os.makedirs(runtime.get_active_config("OUTPUT_DIRECTORY"))
   sample_code = os.path.join(TEST_DIRECTORY, "samples/trivial_program_with_timing")
   self.ssh.connect("localhost")
   self.sample_code_in, stdout, stderr = self.ssh.exec_command("python {0}".format(sample_code))

 def teardown_suite(self):
   if self.sample_code_in is not None:
     self.sample_code_in.write("quit\n")
     self.sample_code_in.flush()
   if self.ssh is not None:
     with self.ssh.open_sftp() as ftp:
       input_csv = os.path.join(runtime.get_active_config("LOGS_DIRECTORY"), "output.csv")
       input_log = os.path.join(runtime.get_active_config("LOGS_DIRECTORY"), "output.log")
       ftp.get("/tmp/trivial_timed_output.csv", input_csv)
       ftp.get("/tmp/trivial_timed_output", input_log)
       ftp.remove("/tmp/trivial_timed_output.csv")
       ftp.remove("/tmp/trivial_timed_output")
     self.ssh.close()

def get_kafka_client(num_retries=20, retry_sleep=1):
  """
  Returns a KafkaClient based off of the kafka_hosts and kafka_port configs set 
  in the active runtime.
  """
  kafka_hosts = runtime.get_active_config('kafka_hosts').values()
  kafka_port = runtime.get_active_config('kafka_port')
  assert len(kafka_hosts) > 0, 'Missing required configuration: kafka_hosts'
  connect_string = ','.join(map(lambda h: h + ':{0},'.format(kafka_port), kafka_hosts)).rstrip(',')
  # wait for at least one broker to come up
  if not wait_for_server(kafka_hosts[0], kafka_port, 30):
    raise Exception('Unable to connect to Kafka broker: {0}:{1}'.format(kafka_hosts[0], kafka_port))
  return KafkaClient(connect_string)

 def run_test_command(test):
   while (test.current_iteration < test.total_number_iterations):
     test.current_iteration = test.current_iteration + 1
     #verify if the test has previously failed. If so then don't try to run again
     #unless the config asks for it
     if (  (test.result != constants.FAILED)
       or (runtime.get_active_config("consecutive_failures_per_test",0) > test.consecutive_failures)
      ):
       self._run_and_verify_test(test)
     #if each test is run for number of required iterations before moving to next test
     #test.total_number_iterations can be 4 if TEST_ITER for test module is set to 2 and loop_all_test is 2
     #in that case each test will be run twice before moving to next test and the whole suite twice
     if ((test.current_iteration % (test.total_number_iterations/int(runtime.get_active_config("loop_all_tests",1))))== 0):
       break

  def _run_and_verify_test(self,test):
    """
    Runs a test and performs validation
    :param test:
    :return:
    """
    if(test.total_number_iterations > 1):
      logger.debug("Executing iteration:" + str(test.current_iteration))
    try:
      test.func_start_time = time.time()
      test.function()
      test.func_end_time = time.time()
      test.iteration_results[test.current_iteration] = constants.PASSED
      #The final iteration result. Useful to make sure the tests recover in case of error injection
      test.result = constants.PASSED
    except BaseException as e:
      test.result = constants.FAILED
      test.iteration_results[test.current_iteration] = constants.FAILED
      test.exception = e
      test.message = traceback.format_exc()
    else:
      #If verify_after_each_test flag is set we can verify after each test even for single iteration
      if ((test.total_number_iterations > 1) or (runtime.get_active_config("verify_after_each_test",False))):
        test.end_time = time.time()
        self._copy_logs()
        self._execute_singletest_verification(test)

    if (test.result == constants.FAILED):
      test.consecutive_failures += 1
    else:
      test.consecutive_failures = 0

  def _execute_run(self, config, naarad_obj):
    """
    Executes tests for a single config
    """
    failure_handler = FailureHandler(config.mapping.get("max_failures_per_suite_before_abort"))
    loop_all_tests = int(runtime.get_active_config("loop_all_tests",1))

    self.compute_total_iterations_per_test()

    #iterate through the test_suite based on config settings
    for i in xrange(loop_all_tests):
      for tests in self.tests:
        if not isinstance(tests, list) or len(tests) == 1:
          if isinstance(tests, list):
            test = tests[0]
          else:
            test = tests
          self._execute_single_test(config, failure_handler, naarad_obj, test)
        else:
          self._execute_parallel_tests(config, failure_handler, naarad_obj, tests)
          
    self._copy_logs()
    if not self.master_config.mapping.get("no_perf", False):
      naarad_obj.signal_stop(config.naarad_id)
      self._execute_performance(naarad_obj)
    self._execute_verification()

def validate_zookeeper_fault_tolerance():
  """
  Validate that we can still connect to zookeeper instance 2 to read the node
  """
  zk2 = KazooClient(hosts=str(runtime.get_active_config('zookeeper_host') + ':2182'))
  zk2.start()
  assert zk2.exists("/my/zookeeper_errorinjection/"), "zookeeper_errorinjection node not found"

  zk2.stop()

  def kill_all_process(self):
    """ Terminates all the running processes. By default it is set to false.
    Users can set to true in config once the method to get_pid is done deterministically
    either using pid_file or an accurate keyword

    """
    if (runtime.get_active_config("cleanup_pending_process",False)):
      for process in self.get_processes():
        self.terminate(process.unique_id)

def zookeeper_ephemeral_node(name):
  zk = KazooClient(hosts=str(runtime.get_active_config('zookeeper_host') + ':2181'))
  zk.start()
  zk.create("/my/zookeeper_test/node1", b"process1 running", ephemeral=True)
  #At 10 validate that ephemeral node exist that is the process is still running
  time.sleep(10)
  assert zk.exists("/my/zookeeper_test/node1"), "process node is not found at 10 s when it is still running"

  time.sleep(20)
  zk.stop()

def test_ping_host1():
    print "==> ping example.com (machine1)"
    pyhk_deployer = runtime.get_deployer("pyhk")

    pyhk_deployer.start(
        "machine1",
        configs={
            "start_command": runtime.get_active_config('ping_cmd'),
            "sync": True
            })

def validate_zookeeper_process_tracking():
  """
  Verify if process register node correctly with zookeeper and zookeeper deletes it when process terminates
  """
  zk = KazooClient(hosts=str(runtime.get_active_config('zookeeper_host') + ':2181'))
  zk.start()

  #At 60 validate that process has terminated by looking at the ephemeral node
  time.sleep(60)
  assert not zk.exists("/my/zookeeper_test/node1"), "process node  not found at 60 s when it should have terminated"

  zk.stop()

 def _copy_logs(self):
   """
   Copy logs from remote machines to local destination
   """
   should_fetch_logs = runtime.get_active_config("should_fetch_logs", True)
   if should_fetch_logs:
    for deployer in runtime.get_deployers():
       for process in deployer.get_processes():
         logs = self.dynamic_config_module.process_logs( process.servicename) or []
         logs += self.dynamic_config_module.machine_logs( process.unique_id)
         logs += self.dynamic_config_module.naarad_logs( process.unique_id)
         pattern = self.dynamic_config_module.log_patterns(process.unique_id) or constants.FILTER_NAME_ALLOW_ALL
         #now copy logs filtered on given pattern to local machine:
         deployer.fetch_logs(process.unique_id, logs, self._logs_dir, pattern)

def test_zookeeper_fault_tolerance():
  """
  Kill zookeeper1 and see if other zookeeper instances are in quorum
  """
  zookeper_deployer = runtime.get_deployer("zookeeper")
  kazoo_connection_url = str(runtime.get_active_config('zookeeper_host') + ':2181')
  zkclient = KazooClient(hosts=kazoo_connection_url)

  zkclient.start()

  zkclient.ensure_path("/my/zookeeper_errorinjection")
  # kill the Zookeeper1 instance
  print "killing zoookeeper instance1"
  zookeper_deployer.kill("zookeeper1")
  time.sleep(20)
  zkclient.stop()

  def compute_total_iterations_per_test(self):
    """
    Factor in loop_all_tests config into iteration count of each test
    Each test has an tests_iteration associated with them from the test module.
    The loop_all_tests is set in config that repeats the entire suite after each
    tests necessary iterations is repeated

    :return:
    """
    loop_all_tests = int(runtime.get_active_config("loop_all_tests",1))
    if (loop_all_tests <= 1):
      return
    else:
      for tests in self.tests:
        if isinstance(tests, list):
          for test in  tests:
            test.total_number_iterations = test.total_number_iterations * loop_all_tests
        else:
          tests.total_number_iterations = tests.total_number_iterations * loop_all_tests

  def _execute_verification(self):
    """
    Executes verification methods for the tests

    :return:
    """
    tests = [test for test in self.tests if not isinstance(test, list)] +\
            [individual_test for test in self.tests if isinstance(test, list) for individual_test in test]
    for test in tests:
      if (test.result != constants.SKIPPED
              and test.validation_function is not None
              and (test.total_number_iterations <= 1)
              and not (runtime.get_active_config("verify_after_each_test",False))
              and hasattr(test.validation_function, '__call__')):
        try:
          test.validation_function()
        except BaseException as e:
          test.result = constants.FAILED
          test.exception = e

def test_zookeeper_process_tracking():
  """
  Tests if process register node correctly with zookeeper and zookeeper deletes it when process terminates
  """
  #Wait for zookeeper to start so that kazoo client can connect correctly
  time.sleep(5)
  #"connecting to esnure /my/zookeeper_test"

  kazoo_connection_url = str(runtime.get_active_config('zookeeper_host') + ':2181')
  zkclient = KazooClient(hosts=kazoo_connection_url)

  zkclient.start()

  zkclient.ensure_path("/my/zookeeper_test")
  #spawn a python multiprocess which creates an ephermeral node
  #once the process ends the node will be deleted.
  p = Process(target=zookeeper_ephemeral_node, args=("process1",))
  p.start()
  zkclient.stop()

  def setup_suite(self):
    print "Starting zookeeper"
    env_dict = {}

    if "localhost" not in runtime.get_active_config('zookeeper_host'):
      env_dict = {'JAVA_HOME':'/export/apps/jdk/current'}

    zookeeper_deployer = adhoc_deployer.SSHDeployer("zookeeper",
        {'pid_keyword': "zookeeper",
         'executable': runtime.get_active_config('zookeeper_exec_location'),
         'env':env_dict,
         'extract': True,
         'post_install_cmds':runtime.get_active_config('zookeeper_post_install_cmds'),
         'stop_command':runtime.get_active_config('zookeeper_stop_command'),
         'start_command': runtime.get_active_config('zookeeper_start_command')})
    runtime.set_deployer("zookeeper", zookeeper_deployer)

    zookeeper_deployer.install("zookeeper",
        {"hostname": runtime.get_active_config('zookeeper_host'),
         "install_path": "/tmp/zookeeper_test"})
    zookeeper_deployer.start("zookeeper",configs={"sync": True})

def setup_suite():
  print "Starting zookeeper quorum"
  global zookeper_deployer
  env_dict = {}

  if "localhost" not in runtime.get_active_config('zookeeper_host'):
    env_dict = {'JAVA_HOME':'/export/apps/jdk/current'}

  zookeper_deployer = adhoc_deployer.SSHDeployer("zookeeper",
      {'pid_keyword': "zookeeper",
       'executable': runtime.get_active_config('zookeeper_exec_location'),
       'env':env_dict,
       'extract': True,
       'stop_command':runtime.get_active_config('zookeeper_stop_command'),
       'start_command': runtime.get_active_config('zookeeper_start_command')})
  runtime.set_deployer("zookeeper", zookeper_deployer)

  # Deploy Zookeeper1
  print "Deploy Zookeeper1"
  zookeper_deployer.install("zookeeper1",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('zookeeper1_install_path'),
       "pid_file": runtime.get_active_config('zookeeper1_pid_file'),       
       'post_install_cmds':runtime.get_active_config('zookeeper1_post_install_cmds')})

  zookeper_deployer.start("zookeeper1",configs={"sync": True})

  # Deploy Zookeeper2
  print "Deploy Zookeeper2"
  zookeper_deployer.install("zookeeper2",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('zookeeper2_install_path'),
       "pid_file": runtime.get_active_config('zookeeper2_pid_file'),       
       'post_install_cmds':runtime.get_active_config('zookeeper2_post_install_cmds')})

  zookeper_deployer.start("zookeeper2",configs={"sync": True})

  # Deploy Zookeeper3
  print "Deploy Zookeeper3"
  zookeper_deployer.install("zookeeper3",
      {"hostname": "localhost",
       "install_path": runtime.get_active_config('zookeeper3_install_path'),
       "pid_file": runtime.get_active_config('zookeeper3_pid_file'),
       'post_install_cmds':runtime.get_active_config('zookeeper3_post_install_cmds')})

  zookeper_deployer.start("zookeeper3",configs={"sync": True})

def setup_suite():
    print "==> Starting tests for PYHK."

    runtime.set_user('root', '')

    global pyhk_deployer
    global tcp_server

    #  Set the server deployer.
    tcp_server = adhoc_deployer.SSHDeployer(
        "server",
        {'executable': runtime.get_active_config('pyhk_exec'),
         'extract': True,
         'start_command': runtime.get_active_config('tcp_server_cmd'),
         'stop_command': "ps ax | grep '[p]ython server' | awk '{print $1}' | xargs kill -9"})
    runtime.set_deployer("server", tcp_server)

    #  Provisioning the server.
    tcp_server.install("server1",
        {"hostname": "10.0.1.23",
         "install_path": runtime.get_active_config('pyhk_install')})

    #  Set the client deployer
    tcp_client = adhoc_deployer.SSHDeployer(
        "client",
        {'executable': runtime.get_active_config('pyhk_exec'),
         'extract': True,
         'start_command': runtime.get_active_config('tcp_client_cmd')})
    runtime.set_deployer("client", tcp_client)

    #  Provisioning the client.
    tcp_client.install("client1",
        {"hostname": "10.0.1.24",
         "install_path": runtime.get_active_config('pyhk_install')})

    #  Set general deployer.
    pyhk_deployer = adhoc_deployer.SSHDeployer(
        "pyhk",
        {'executable': runtime.get_active_config('pyhk_exec'),
         'extract': True,
         'start_command': runtime.get_active_config('pyhk_cmd')})
    runtime.set_deployer("pyhk", pyhk_deployer)

    #  Hostname 1
    pyhk_deployer.install(
        "machine1",
        {"hostname": runtime.get_active_config('pyhk_hostname1'),
         "install_path": runtime.get_active_config('pyhk_install')})

    #  Hostname 2
    pyhk_deployer.install(
        "machine2",
        {"hostname": runtime.get_active_config('pyhk_hostname2'),
         "install_path": runtime.get_active_config('pyhk_install')})