testcases/basic_func_tests/tc_006_ssh_test_ext_ntp.py

   1 import sys
   2 import os
   3 from decorators_for_robot_functionalities import *
   4 from robot.api import logger
   5 from robot.libraries.BuiltIn import BuiltIn
   6 from test_constants import *
   7
   8 sys.path.append(os.path.join(os.path.dirname(__file__), '..', '../libraries/common'))
   9 import common_utils  # noqa
  10
  11 ex = BuiltIn().get_library_instance('execute_command')
  12 stack_infos = BuiltIn().get_library_instance('stack_infos')
  13
  14
  15 def tc_006_ssh_test_ext_ntp():
  16     steps = ['step1_check_ntpd_service_and_ext_ntp_ip_on_crf_nodes']
  17     common_utils.keyword_runner(steps)
  18
  19
  20 def step1_check_ntpd_service_and_ext_ntp_ip_on_crf_nodes():
  21     crf_nodes = stack_infos.get_crf_nodes()
  22     check_ntpd_status(crf_nodes)
  23     check_if_nokia_ntp_server_address_set_on_crf_node(crf_nodes)
  24
  25
  26 @robot_log
  27 def check_ntpd_status(nodes):
  28     if not nodes:
  29         logger.info("Nodes dictionary is empty, nothing to check.")
  30         return
  31     command = 'systemctl status ntpd.service | grep --color=no "Active"'
  32     for node in nodes:
  33         logger.console("\nCheck ntpd status " + node + " " + nodes[node])
  34         stdout = ex.execute_unix_command_on_remote_as_user(command, nodes[node])
  35         if "running" not in stdout:
  36             raise Exception("ntpd.service is not running!")
  37
  38
  39 @robot_log
  40 def get_ext_ntp_ips_from_node():
  41     return stack_infos.get_inventory()["all"]["vars"]["time"]["ntp_servers"]
  42
  43
  44 @robot_log
  45 def filter_valid_ntp_servers(ntp_servers):
  46     valid_servers = []
  47     for server in ntp_servers:
  48         stdout = ex.execute_unix_command("ntpdate -q {}".format(server), fail_on_non_zero_rc=False)
  49         if "no server suitable for synchronization found" not in stdout:
  50             valid_servers.append(server)
  51     return valid_servers
  52
  53
  54 @robot_log
  55 def is_ntp_server_set_on_node(server_ip, node):
  56     command = 'ntpq -pn | grep -w --color=no ' + server_ip
  57     stdout = ex.execute_unix_command_on_remote_as_user(command, node, {}, fail_on_non_zero_rc=False)
  58     return server_ip in str(stdout)
  59
  60
  61 @robot_log
  62 def check_if_nokia_ntp_server_address_set_on_crf_node(nodes):
  63     ext_ntp_server_ips = get_ext_ntp_ips_from_node()
  64     valid_servers = filter_valid_ntp_servers(ext_ntp_server_ips)
  65     logger.info("The following ntp_servers will be tested:")
  66     logger.info(valid_servers)
  67     is_ip_set = True
  68     for node in nodes:
  69         for ntp_serv_ip in valid_servers:
  70             if not is_ntp_server_set_on_node(ntp_serv_ip, node):
  71                 is_ip_set = False
  72     if not is_ip_set:
  73         raise Exception("Wrong or no NTP server address set!")