diff --git a/tests/test_ntp.py b/tests/test_ntp.py index 6a3cacf25bfee93986c38fad48207a80dc0b30f9..ce85bec5c6e936023ae32670a3183b78af8fa0ae 100755 --- a/tests/test_ntp.py +++ b/tests/test_ntp.py @@ -5,66 +5,73 @@ Criticality: Low Checks that the server is synchronized with the configured NTP server. ''' -import imp +from pathlib import Path import os import re import subprocess import sys -YELLOW = '\033[93m' -GREEN = '\033[92m' -RED = '\033[91m' -DEF = '\033[0m' +sys.path.append(str(Path(__file__).parents[1].resolve())) -# Check that ntpd is synced -if os.path.isfile('/usr/bin/ntpq'): - cmd = 'LANG=C ntpq -pd' - expected = 'remote' - ntpconf = '/etc/ntp.conf' - ntpconf_expected = r'^(?:server|pool)\s([a-zA-Z0-9\._-]+)(?:\s*iburst)?$' -else: - cmd = 'LANG=C timedatectl' - expected = 'NTP synchronized' - ntpconf = '/etc/systemd/timesyncd.conf' - ntpconf_expected = r'^NTP=(.*)$' +# pylint: disable=wrong-import-position +from envsetup import utils as u # noqa: E402 -print('Running %s' % cmd) -status = subprocess.getoutput(cmd) -if expected not in status: - print('%sNTP not working: %s%s' % (RED, status, DEF)) - sys.exit(1) -print('%sSystem is NTP synchronized.%s' % (GREEN, DEF)) -os.chdir(os.path.dirname(__file__)) +def main(): + # Check that ntpd is synced + if os.path.isfile('/usr/bin/ntpq'): + cmd = 'LANG=C ntpq -pd' + expected = 'remote' + ntpconf = '/etc/ntp.conf' + ntpconf_expected = r'^(?:server|pool)\s([a-zA-Z0-9\._-]+)(?:\s*iburst)?$' + else: + cmd = 'LANG=C timedatectl' + expected = 'NTP synchronized' + ntpconf = '/etc/systemd/timesyncd.conf' + ntpconf_expected = r'^NTP=(.*)$' -print('Checking NTP server conforms to conf...') -if not os.path.isfile('../utils.py'): - print('%sCould not find envsetup conf file or not running from expected location.%s' % (RED, DEF)) - sys.exit(1) + u.log('Running %s' % cmd) + status = subprocess.getoutput(cmd) + if expected not in status: + u.error('NTP not working: %s' % status) + return 1 + u.success('System is NTP synchronized.') -es_utils = imp.load_source('es_utils', '../utils.py') -conf = es_utils.load_conf() + os.chdir(os.path.dirname(__file__)) -expected_servers = None -if conf.get('NTP_SERVER'): - expected_servers = [s.strip() for s in conf['NTP_SERVER'].split(',')] -if not expected_servers: - if 'Ubuntu' in subprocess.getoutput('lsb_release -a'): - expected_servers = ['ntp.ubuntu.com'] - else: - expected_servers = ['0.debian.pool.ntp.org'] + u.log('Checking NTP server conforms to conf...') + if not os.path.isfile('../utils.py'): + u.error('Could not find envsetup conf file or not running from expected location.') + return 1 -with open(ntpconf, 'r') as f: - content = f.read() -servers = list() -for l in content.split('\n'): - m = re.match(ntpconf_expected, l) - if m: - servers.append(m.groups()[0].strip()) -for expected_server in expected_servers: - if expected_server not in servers: - print('%sWarning: Expected NTP server %s not found in %s, found %s instead.%s' % (YELLOW, expected_server, ntpconf, ', '.join(servers), DEF)) - sys.exit(3) - else: - print('Expected NTP server %s found in configuration (total servers: %s).' % (expected_server, len(servers))) -print('%sNTP OK.%s' % (GREEN, DEF)) + conf = u.load_conf() + + expected_servers = None + if conf.get('NTP_SERVER'): + expected_servers = [s.strip() for s in conf['NTP_SERVER'].split(',')] + if not expected_servers: + if 'Ubuntu' in subprocess.getoutput('lsb_release -a'): + expected_servers = ['ntp.ubuntu.com'] + else: + expected_servers = ['0.debian.pool.ntp.org'] + + with open(ntpconf, 'r') as fo: + content = fo.read() + servers = list() + for line in content.split('\n'): + m = re.match(ntpconf_expected, line) + if m: + servers.append(m.groups()[0].strip()) + for expected_server in expected_servers: + if expected_server not in servers: + u.warning('Warning: Expected NTP server %s not found in %s, found %s instead.%s' % (expected_server, ntpconf, ', '.join(servers))) + return 3 + else: + u.log('Expected NTP server %s found in configuration (total servers: %s).' % (expected_server, len(servers))) + u.success('NTP OK.') + return 0 + + +if __name__ == '__main__': + code = main() + sys.exit(code)