Merge ~jocave/plainbox-provider-checkbox:netplan-try-finally into plainbox-provider-checkbox:master

[Jonathan Cave]

Changes made when trying to isolate the problem with these tests on the core20 snap 20211124. Needed some kind of persistent logging and ability to attempt to automatically restore the netplan config.

Unmerged commits

e411366... by Jonathan Cave on 2021-11-10

Change: run netplan test in try/finally

diff --git a/bin/wifi_client_test_netplan.py b/bin/wifi_client_test_netplan.py
index d984e2e..21cbb94 100755
--- a/bin/wifi_client_test_netplan.py
+++ b/bin/wifi_client_test_netplan.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-# Copyright 2018-2019 Canonical Ltd.
+# Copyright 2018-2021 Canonical Ltd.
 # All rights reserved.
 #
 # Written by:
@@ -14,7 +14,6 @@ To see how to use, please run "./wifi_client_test_netplan.py --help"
 
 import argparse
 import datetime
-import functools
 import glob
 import os
 import subprocess as sp
@@ -22,18 +21,32 @@ import textwrap
 import time
 import shutil
 import sys
+import logging
+import logging.handlers
 
 from gateway_ping_test import ping
 
-print = functools.partial(print, flush=True)
 
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(message)s',
+    handlers=[
+        logging.FileHandler("/var/tmp/checkbox-netplan.log"),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
 
-def print_head(txt):
-    print("##", txt)
 
+def info_header(txt):
+    logging.info("## %s", txt)
 
-def print_cmd(cmd):
-    print("+", cmd)
+
+def info_cmd(cmd):
+    logging.info("+ %s", cmd)
+
+
+def info_line(line=""):
+    logging.info("%s", line)
 
 
 # Configuration file path
@@ -48,31 +61,31 @@ NETPLAN_TEST_CFG = "/etc/netplan/99-CREATED-BY-CHECKBOX.yaml"
 
 
 def netplan_config_backup():
-    print_head("Backup any existing netplan configuration files")
+    info_header("Backup any existing netplan configuration files")
     if os.path.exists(TMP_PATH):
-        print("Clear backup location")
+        info_line("Clear backup location")
         shutil.rmtree(TMP_PATH)
     for basedir in NETPLAN_CFG_PATHS:
-        print("Checking in {}".format(basedir))
+        info_line("Checking in {}".format(basedir))
         files = glob.glob(os.path.join(basedir, '*.yaml'))
         if files:
             backup_loc = os.path.join(TMP_PATH, *basedir.split('/'))
             os.makedirs(backup_loc)
             for f in files:
-                print(" ", f)
+                info_line(" {}".format(f))
                 shutil.copy(f, backup_loc)
-    print()
+    info_line()
 
 
 def netplan_config_wipe():
-    print_head("Delete any existing netplan configuration files")
+    info_header("Delete any existing netplan configuration files")
     # NOTE: this removes not just configs for wifis, but for all device types
     #  (ethernets, bridges) which could be dangerous
     for basedir in NETPLAN_CFG_PATHS:
-        print("Wiping {}".format(basedir))
+        info_line("Wiping {}".format(basedir))
         files = glob.glob(os.path.join(basedir, '*.yaml'))
         for f in files:
-            print(" ", f)
+            info_line(" {}".format(f))
             os.remove(f)
 
     # If there's any file left in configuration folder then there's something
@@ -80,22 +93,22 @@ def netplan_config_wipe():
     for basedir in NETPLAN_CFG_PATHS:
         files = glob.glob(os.path.join(basedir, "*.yaml"))
         if files:
-            print("ERROR: Failed to wipe netplan config files:")
+            logging.critical("ERROR: Failed to wipe netplan config files:")
             for f in files:
-                print(" ", f)
+                info_line(" {}".format(f))
             netplan_config_restore()
             raise SystemExit("Configuration file restored, exiting...")
-    print()
+    info_line()
 
 
 def netplan_config_restore():
-    print_head("Restore configuration files")
+    info_header("Restore configuration files")
     files = glob.glob("{}/**/*.yaml".format(TMP_PATH), recursive=True)
     if files:
         print("Restoring:")
         for f in files:
             restore_loc = f[len(TMP_PATH):]
-            print(" ", restore_loc)
+            info_line(" {}".format(restore_loc))
             try:
                 shutil.move(f, restore_loc)
             except shutil.Error:
@@ -168,33 +181,30 @@ def generate_test_config(interface, ssid, psk, address, dhcp):
 
 
 def write_test_config(config):
-    print_head("Write the test netplan config file to disk")
+    info_header("Write the test netplan config file to disk")
     with open(NETPLAN_TEST_CFG, "w", encoding="utf-8") as f:
         f.write(config)
-    print()
+    info_line()
 
 
 def delete_test_config():
-    print_head("Delete the test file")
+    info_header("Delete the test file")
     os.remove(NETPLAN_TEST_CFG)
-    print()
+    info_line()
 
 
 def netplan_apply_config():
     cmd = "netplan --debug apply"
-    print_cmd(cmd)
+    info_cmd(cmd)
     # Make sure the python env used by netplan is from the base snap
     env = os.environ
     env.pop('PYTHONHOME', None)
     env.pop('PYTHONPATH', None)
     env.pop('PYTHONUSERBASE', None)
-    retcode = sp.call(cmd, shell=True, env=env)
-    if retcode != 0:
-        print("ERROR: failed netplan apply call")
-        print()
-        return False
-    print()
-    return True
+    output = sp.check_output(cmd, shell=True, stderr=sp.STDOUT, env=env)
+    for line in output.decode(sys.stdout.encoding).splitlines():
+        info_line(line)
+    info_line()
 
 
 def _get_networkctl_state(interface):
@@ -207,45 +217,46 @@ def _get_networkctl_state(interface):
 
 
 def wait_for_routable(interface, max_wait=30):
-    routable = False
     attempts = 0
-    while not routable and attempts < max_wait:
+    while attempts < max_wait:
         state = _get_networkctl_state(interface)
-        print(state)
+        info_line(state)
         if "routable" in state:
-            routable = True
-            break
+            info_line("Reached routable state")
+            info_line()
+            return
         time.sleep(1)
         attempts += 1
-    if routable:
-        print("Reached routable state")
+
+    if "degraded" in state:
+        raise RuntimeError("ERROR: degraded state, no IP address assigned")
     else:
-        if "degraded" in state:
-            print("ERROR: degraded state, no IP address assigned")
-        else:
-            print("ERROR: did not reach routable state")
-    print()
-    return routable
+        raise RuntimeError("ERROR: did not reach routable state")
 
 
 def print_address_info(interface):
     cmd = 'ip address show dev {}'.format(interface)
-    print_cmd(cmd)
-    sp.call(cmd, shell=True)
-    print()
+    info_cmd(cmd)
+    output = sp.check_output(cmd, shell=True, stderr=sp.STDOUT)
+    for line in output.decode(sys.stdout.encoding).splitlines():
+        info_line(line)
+    info_line()
 
 
 def print_route_info():
     cmd = 'ip route'
-    print_cmd(cmd)
-    sp.call(cmd, shell=True)
-    print()
+    info_cmd(cmd)
+    output = sp.check_output(cmd, shell=True, stderr=sp.STDOUT)
+    for line in output.decode(sys.stdout.encoding).splitlines():
+        info_line(line)
+    info_line()
 
 
 def perform_ping_test(interface):
     target = None
     cmd = 'networkctl status --no-pager --no-legend {}'.format(interface)
-    print_cmd(cmd)
+    info_cmd(cmd)
+    # get ip to ping
     output = sp.check_output(cmd, shell=True)
     for line in output.decode(sys.stdout.encoding).splitlines():
         vals = line.strip().split(' ')
@@ -253,29 +264,31 @@ def perform_ping_test(interface):
             if vals[0] == 'Gateway:':
                 target = vals[1]
                 print('Got gateway address: {}'.format(target))
-
+    # do the test
     if target:
         count = 5
         result = ping(target, interface, count, 10, True)
         if result['received'] == count:
-            return True
-
-    return False
+            info_line("Connection test passed")
+            info_line()
+            return
+    raise RuntimeError("Connection test failed")
 
 
 def print_journal_entries(start):
-    print_head("Journal Entries")
+    info_header("Journal Entries")
     cmd = ('journalctl -q --no-pager '
            '-u systemd-networkd.service '
            '-u wpa_supplicant.service '
            ' -u netplan-* '
            '--since "{}" '.format(start.strftime('%Y-%m-%d %H:%M:%S')))
-    print_cmd(cmd)
-    sp.call(cmd, shell=True)
+    info_cmd(cmd)
+    output = sp.check_output(cmd, shell=True)
+    for line in output.decode(sys.stdout.encoding).splitlines():
+        info_line(line)
 
 
 def main():
-    # Read arguments
     parser = argparse.ArgumentParser(
         description=("This script will test wireless network with netplan"
                      " in client mode."))
@@ -305,57 +318,55 @@ def main():
 
     start_time = datetime.datetime.now()
 
-    netplan_config_backup()
-    netplan_config_wipe()
-
     # Create wireless network test configuration file
-    print_head("Generate a test netplan configuration")
+    info_header("Generate a test netplan configuration")
     config_data = generate_test_config(**vars(args))
-    print(config_data)
-    print()
+    info_line(config_data)
+    info_line()
 
-    write_test_config(config_data)
+    netplan_config_backup()
+    netplan_config_wipe()
 
-    # Bring up the interface
-    print_head("Apply the test configuration")
-    if not netplan_apply_config():
-        delete_test_config()
-        netplan_config_restore()
-        print_journal_entries(start_time)
-        raise SystemExit(1)
-    time.sleep(20)
+    try:
+        write_test_config(config_data)
 
-    print_head("Wait for interface to be routable")
-    reached_routable = wait_for_routable(args.interface)
+        # Bring up the interface
+        info_header("Apply the test configuration")
+        netplan_apply_config()
+        time.sleep(20)
 
-    test_result = False
-    if reached_routable:
-        print_head("Display address")
+        info_header("Wait for interface to be routable")
+        wait_for_routable(args.interface)
+
+        info_header("Display address")
         print_address_info(args.interface)
 
-        print_head("Display route table")
+        info_header("Display route table")
         print_route_info()
 
         # Check connection by ping or link status
-        print_head("Perform a ping test")
-        test_result = perform_ping_test(args.interface)
-        if test_result:
-            print("Connection test passed\n")
-        else:
-            print("Connection test failed\n")
-
-    delete_test_config()
-    netplan_config_restore()
-
-    if not netplan_apply_config():
+        info_header("Perform a ping test")
+        perform_ping_test(args.interface)
+    except sp.CalledProcessError as cpe:
+        for line in cpe.output.decode(sys.stdout.encoding).splitlines():
+            logging.critical("%s", line)
+        info_line()
         print_journal_entries(start_time)
-        raise SystemExit("ERROR: failed to apply restored config")
-
-    if not test_result:
+        info_line()
+        raise SystemExit("FAILED")
+    except RuntimeError as re:
+        logging.critical(str(re))
         print_journal_entries(start_time)
-        raise SystemExit(1)
+        info_line()
+        raise SystemExit("FAILED")
+    finally:
+        delete_test_config()
+        netplan_config_restore()
+        netplan_apply_config()
 
     print_journal_entries(start_time)
+    info_line()
+    info_line("PASSED")
 
 
 if __name__ == "__main__":