Skip to content
Snippets Groups Projects
Commit e391769f authored by Stéphane Diemer's avatar Stéphane Diemer
Browse files

Avoid imp usage in NTP test

parent 3ea31f5c
No related branches found
No related tags found
No related merge requests found
...@@ -5,66 +5,73 @@ ...@@ -5,66 +5,73 @@
Criticality: Low Criticality: Low
Checks that the server is synchronized with the configured NTP server. Checks that the server is synchronized with the configured NTP server.
''' '''
import imp from pathlib import Path
import os import os
import re import re
import subprocess import subprocess
import sys import sys
YELLOW = '\033[93m' sys.path.append(str(Path(__file__).parents[1].resolve()))
GREEN = '\033[92m'
RED = '\033[91m'
DEF = '\033[0m'
# Check that ntpd is synced # pylint: disable=wrong-import-position
if os.path.isfile('/usr/bin/ntpq'): from envsetup import utils as u # noqa: E402
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('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...') u.log('Running %s' % cmd)
if not os.path.isfile('../utils.py'): status = subprocess.getoutput(cmd)
print('%sCould not find envsetup conf file or not running from expected location.%s' % (RED, DEF)) if expected not in status:
sys.exit(1) u.error('NTP not working: %s' % status)
return 1
u.success('System is NTP synchronized.')
es_utils = imp.load_source('es_utils', '../utils.py') os.chdir(os.path.dirname(__file__))
conf = es_utils.load_conf()
expected_servers = None u.log('Checking NTP server conforms to conf...')
if conf.get('NTP_SERVER'): if not os.path.isfile('../utils.py'):
expected_servers = [s.strip() for s in conf['NTP_SERVER'].split(',')] u.error('Could not find envsetup conf file or not running from expected location.')
if not expected_servers: return 1
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 f: conf = u.load_conf()
content = f.read()
servers = list() expected_servers = None
for l in content.split('\n'): if conf.get('NTP_SERVER'):
m = re.match(ntpconf_expected, l) expected_servers = [s.strip() for s in conf['NTP_SERVER'].split(',')]
if m: if not expected_servers:
servers.append(m.groups()[0].strip()) if 'Ubuntu' in subprocess.getoutput('lsb_release -a'):
for expected_server in expected_servers: expected_servers = ['ntp.ubuntu.com']
if expected_server not in servers: else:
print('%sWarning: Expected NTP server %s not found in %s, found %s instead.%s' % (YELLOW, expected_server, ntpconf, ', '.join(servers), DEF)) expected_servers = ['0.debian.pool.ntp.org']
sys.exit(3)
else: with open(ntpconf, 'r') as fo:
print('Expected NTP server %s found in configuration (total servers: %s).' % (expected_server, len(servers))) content = fo.read()
print('%sNTP OK.%s' % (GREEN, DEF)) 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)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment