update test dns

tests/test_dns_records.py

 #!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-# Copyright 2017, Florent Thiery
+
 """
 Criticality: Normal
 Checks that DNS records are provided by the customer servers are correctly set
 """
+
+from pathlib import Path
 import subprocess
-import os
 import sys
-import imp
 
-GREEN = "\033[92m"
-RED = "\033[91m"
-DEF = "\033[0m"
+sys.path.append(str(Path(__file__).parents[1].resolve()))
+
+# pylint: disable=wrong-import-position
+from envsetup import utils as u  # noqa: E402
 
 
-def get_dns_servers():
+def get_dns_servers() -> list:
     servers = list()
     with open("/etc/resolv.conf", "r") as f:
         d = f.read().strip()
@@ -24,96 +24,112 @@ def get_dns_servers():
                 servers.append(l.split("nameserver ")[1])
     if servers == ["127.0.1.1"]:
         # NetworkManager
-        status, output = subprocess.getstatusoutput(
+        _, output = subprocess.getstatusoutput(
             "nmcli -f all device show | grep IP4.DNS | awk '{ print $2 }'"
         )
         servers.extend(output.split("\n"))
-    return servers
-
 
-resolvers = get_dns_servers()
+    return servers
 
 
-def get_result(output):
+def get_result(output: str) -> str:
     for line in output.split("\n"):
         if "has address " in line:
             return line.split("has address ")[1]
 
 
-def check_dns(hostname, expected_ip):
-    all_ok = True
+def check_dns(hostname: str, expected_ip: str, resolvers: list) -> tuple:
+    warnings = 0
+    errors = 0
+
     for resolver in resolvers:
         status, output = subprocess.getstatusoutput(
-            'host "%s" "%s"' % (hostname, resolver)
+            "host {} {}".format(hostname, resolver)
         )
         if status == 0:
-            color = GREEN
             address = get_result(output)
-            if address == "127.0.0.1" and resolver == "127.0.0.53":
-                pass
-            elif address != expected_ip:
-                print('Expected ip was: "%s", got "%s".' % (expected_ip, address))
-                color = RED
-                all_ok = False
-        else:
-            color = RED
-            all_ok = False
-            address = "FAIL"
-        print(
-            '%sDNS resolution of "%s" on server "%s" returned "%s".%s'
-            % (color, hostname, resolver, address, DEF)
-        )
-    return all_ok
-
-
-os.chdir(os.path.dirname(__file__))
-if not os.path.isfile("../utils.py"):
-    print("conf.sh not found")
-    sys.exit(1)
-
-es_utils = imp.load_source("es_utils", "../utils.py")
-conf = es_utils.load_conf()
-
-all_ok = True
-conf_resolvers_keys = ("NETWORK_DNS1", "NETWORK_DNS2")
-for conf_resolver_key in conf_resolvers_keys:
-    conf_resolver = conf.get(conf_resolver_key)
-    if conf_resolver and conf_resolver != "0" and conf_resolver not in resolvers:
-        print("Resolver %s not configured on the system." % conf_resolver)
-        all_ok = False
-
-ip = conf.get("NETWORK_IP_NAT")
-if not ip or ip == "0":
-    ip = conf.get("NETWORK_IP")
-if not ip or ip == "0":
-    if not all_ok:
-        sys.exit(1)
-    print("No IP set in configuration file, unable to test DNS.")
-    sys.exit(2)
-
-services_info = (
-    ("MS_SERVER_NAME", "mediaserver", "python3-mediaserver"),
-    ("MONITOR_SERVER_NAME", "monitor", "python3-mediaserver-monitor"),
-    ("CM_SERVER_NAME", "mirismanager", "skyreach"),
-)
-
-for conf_name, default_domain, package in services_info:
-    domain = conf.get(conf_name)
-    if domain and domain not in ("localhost", default_domain):
-        # check that the service is installed on this system
-        status, output = subprocess.getstatusoutput("dpkg -s %s" % package)
-        if status == 0:
-            print('Checking IP of "%s".' % domain)
-            ok = check_dns(domain, ip)
-            if not ok:
-                all_ok = False
+            if address == expected_ip:
+                u.success("dns({}): {} -> {}".format(resolver, hostname, address))
+            elif address == "127.0.0.1" and resolver == "127.0.0.53":
+                u.success("dns({}): {} -> {}".format(resolver, hostname, address))
+            else:
+                u.error(
+                    "dns({}): {} -> {} (should be {})".format(
+                        resolver, hostname, address, expected_ip
+                    )
+                )
+                errors += 1
         else:
-            print(
-                'Package %s is not installed, IP of "%s" will not be checked.'
-                % (package, domain)
-            )
-
-if not all_ok:
-    sys.exit(1)
-else:
-    sys.exit(0)
+            u.error("dns({}): cannot resolve {}".format(resolver, hostname))
+            errors += 1
+
+    return warnings, errors
+
+
+def check_resolver(conf: dict, resolvers: list, ip: str) -> tuple:
+    warnings = 0
+    errors = 0
+
+    conf_resolvers_keys = ("NETWORK_DNS1", "NETWORK_DNS2")
+    for conf_resolver_key in conf_resolvers_keys:
+        conf_resolver = conf.get(conf_resolver_key)
+        if conf_resolver and conf_resolver not in resolvers:
+            u.error("resolver {} not configured".format(conf_resolver))
+            errors += 1
+
+    if not ip and (not errors):
+        u.info("no IP set in configuration , unable to test DNS")
+        exit(2)
+
+    return warnings, errors
+
+
+def main():
+    print("Check DNS settings:")
+
+    warnings = 0
+    errors = 0
+    conf = u.load_conf()
+    resolvers = get_dns_servers()
+    ip = conf.get("NETWORK_IP_NAT") or conf.get("NETWORK_IP")
+
+    check_resolver_warn, check_resolver_err = check_resolver(conf, resolvers, ip)
+    if check_resolver_err:
+        errors += check_resolver_err
+    if check_resolver_warn:
+        warnings += check_resolver_warn
+
+    services_info = (
+        ("MS_SERVER_NAME", "mediaserver", "python3-mediaserver"),
+        ("MONITOR_SERVER_NAME", "monitor", "python3-mediaserver-monitor"),
+        ("CM_SERVER_NAME", "mirismanager", "skyreach"),
+    )
+
+    for conf_name, default_domain, package in services_info:
+        domain = conf.get(conf_name)
+        resolution_ignored = conf.get("TESTER_DNS_RESOLUTION_IGNORED", "").split(",")
+        if (
+            domain
+            and domain not in ("localhost", default_domain)
+            and domain not in resolution_ignored
+        ):
+            # check that the service is installed on this system
+            status, _ = subprocess.getstatusoutput("dpkg -s {}".format(package))
+            if status == 0:
+                u.info("- checking IP of {}".format(domain))
+                check_dns_warn, check_dns_err = check_dns(domain, ip, resolvers)
+                if check_dns_err:
+                    errors += check_dns_err
+                if check_dns_warn:
+                    warnings += check_dns_warn
+            else:
+                u.info("{} not installed, skip {}".format(package, domain))
+
+    if errors:
+        exit(1)
+    elif warnings:
+        exit(3)
+
+
+if __name__ == "__main__":
+    main()