WIP: Improve config parsing, make --verbose and --config-file global options

Bonus: Generalize and reuse cmd_ssh()

Signed-off-by: InsanePrawn <insane.prawny@gmail.com>

boot.py

@@ -1,18 +1,14 @@
 import os
 import urllib.request
 from image import get_device_and_flavour, get_image_name, dump_bootimg, dump_lk2nd
-from logger import setup_logging, verbose_option
 from fastboot import fastboot_boot, fastboot_erase_dtbo
 from constants import BOOT_STRATEGIES, FASTBOOT, JUMPDRIVE, LK2ND, JUMPDRIVE_VERSION
 import click
 
 
 @click.command(name='boot')
-@verbose_option
 @click.argument('type', required=False)
-def cmd_boot(verbose, type):
-    setup_logging(verbose)
-
+def cmd_boot(type):
     device, flavour = get_device_and_flavour()
     image_name = get_image_name(device, flavour)
     strategy = BOOT_STRATEGIES[device]

cache.py

@@ -1,5 +1,4 @@
 import shutil
-from logger import setup_logging, verbose_option
 import click
 import os
 
@@ -10,10 +9,7 @@ def cmd_cache():
 
 
 @click.command(name='clean')
-@verbose_option
-def cmd_clean(verbose):
-    setup_logging(verbose)
-
+def cmd_clean():
     for dir in ['/chroot', '/var/cache/pacman/pkg', '/var/cache/jumpdrive']:
         for file in os.listdir(dir):
             path = os.path.join(dir, file)

config.py

@@ -3,6 +3,7 @@ import os
 import toml
 import logging
 from copy import deepcopy
+import click
 
 CONFIG_DEFAULT_PATH = os.path.join(appdirs.user_config_dir('kupfer'), 'kupferbootstrap.toml')
 
@@ -23,33 +24,28 @@ CONFIG_DEFAULTS = {
     }
 }
 
-
-class ConfigParserException(Exception):
-    pass
-
-
-def load_config(config_file=None, merge_defaults=True):
+def parse_file(config_file: str, base: dict=CONFIG_DEFAULTS) -> dict:
+    """
+    Parse the toml contents of `config_file`, validating keys against `CONFIG_DEFAULTS`.
+    The parsed results are semantically merged into `base` before returning.
+    `base` itself is NOT checked for invalid keys.
+    """
     _conf_file = config_file if config_file != None else CONFIG_DEFAULT_PATH
     loaded_conf = toml.load(_conf_file)
-
-    if merge_defaults:
-        # Selectively merge known keys in loaded_conf with CONFIG_DEFAULTS
-        parsed = deepcopy(CONFIG_DEFAULTS)
-    else:
-        parsed = {}
+    parsed = deepcopy(base)
 
     for outer_name, outer_conf in loaded_conf.items():
         # only handle known config sections
         if outer_name not in CONFIG_DEFAULTS.keys():
-            logging.warning('Removed unknown config section', outer_name)
+            logging.warning(f'Skipped unknown config section "{outer_name}"')
             continue
         logging.debug(f'Working on outer section "{outer_name}"')
         # check if outer_conf is a dict
         if not isinstance(outer_conf, dict):
             parsed[outer_name] = outer_conf
         else:
-            if not merge_defaults:
             # init section
+            if outer_name not in parsed:
                 parsed[outer_name] = {}
 
             # profiles need special handling:
@@ -57,11 +53,11 @@ def load_config(config_file=None, merge_defaults=True):
             # 2. A profile's subkeys must be compared against PROFILE_DEFAULTS.keys()
             if outer_name == 'profiles':
                 if 'default' not in outer_conf.keys():
-                    logging.warning('Default profile is not in profiles')
+                    logging.warning('Default profile is not defined in config file')
 
                 for profile_name, profile_conf in outer_conf.items():
-                    #  init profile; don't accidentally overwrite the default profile when merging
-                    if not (merge_defaults and profile_name == 'default'):
+                    #  init profile
+                    if profile_name not in parsed[outer_name]:
                         parsed[outer_name][profile_name] = {}
 
                     for key, val in profile_conf.items():
@@ -80,13 +76,79 @@ def load_config(config_file=None, merge_defaults=True):
 
     return parsed
 
+class ConfigLoadException(Exception):
+    inner = None
+    def __init__(self, extra_msg='', inner_exception: Exception = None):
+        msg = ['Config load failed!']
+        if extra_msg:
+            msg[0].append(':')
+            msg.append(extra_msg)
+        if inner_exception:
+            self.inner = inner_exception
+            msg.append(str(inner_exception))
+        super().__init__(self, ' '.join(msg))
+
+class ConfigStateHolder:
+    class ConfigLoadState:
+        load_finished = False
+        exception = None
+    file_state = ConfigLoadState()
+
+    # config options that are persisted to file
+    file: dict = {}
+    # runtime config not persisted anywhere
+    runtime: dict = {'verbose': False}
+
+    def __init__(self, runtime_conf = {}, file_conf_path: str = None, file_conf_base: dict = {}):
+        """init a stateholder, optionally loading `file_conf_path`"""
+        self.runtime.update(runtime_conf)
+        self.file.update(file_conf_base)
+        if file_conf_path:
+            self.try_load_file(file_conf_path)
+
+    def try_load_file(self, config_file=None, base=CONFIG_DEFAULTS):
+        _conf_file = config_file if config_file != None else CONFIG_DEFAULT_PATH
+        try:
+            self.file = parse_file(config_file=_conf_file, base=base)
+        except Exception as ex:
+            self.file_state.exception = ex
+        self.file_state.load_finished = True
+
+    def is_loaded(self):
+        return self.file_state.load_finished and self.file_state.exception == None
+
+    def enforce_config_loaded(self):
+        if not self.file_state.load_finished:
+            raise ConfigLoadException(Exception("Config file wasn't even parsed yet. This is probably a bug in kupferbootstrap :O"))
+        ex = self.file_state.exception
+        if ex:
+            msg = ''
+            if type(ex) == FileNotFoundError:
+                msg = "File doesn't exist. Try running `kupferbootstrap config init` first?"
+            raise ConfigLoadException(extra_msg=msg, inner_exception=ex)
+
+config = ConfigStateHolder(file_conf_base=CONFIG_DEFAULTS)
+
+
+config_option = click.option(
+    '-C',
+    '--config',
+    'config_file',
+    help='Override path to config file',
+)
 
 # temporary demo
 if __name__ == '__main__':
+    print('vanilla:')
+    print(toml.dumps(config.file))
+    print('\n\n-----------------------------\n\n')
+
     try:
-        conf = load_config()
-    except FileNotFoundError as ex:
-        logging.warning(f'Error reading toml file "{ex.filename}": {ex.strerror}')
+        config.try_load_file()
+        config.enforce_config_loaded()
+        conf = config.file
+    except ConfigLoadException as ex:
+        logging.fatal(str(ex))
         conf = deepcopy(CONFIG_DEFAULTS)
     conf['profiles']['pinephone'] = {'hostname': 'slowphone', 'pkgs_include': ['zsh', 'tmux', 'mpv', 'firefox']}
     print(toml.dumps(conf))

flash.py

@@ -7,16 +7,13 @@ import os
 import subprocess
 import click
 import tempfile
-from logger import logging, setup_logging, verbose_option
+from logger import logging
 
 
 @click.command(name='flash')
-@verbose_option
 @click.argument('what')
 @click.argument('location', required=False)
-def cmd_flash(verbose, what, location):
-    setup_logging(verbose)
-
+def cmd_flash(what, location):
     device, flavour = get_device_and_flavour()
     image_name = get_image_name(device, flavour)
 

forwarding.py

@@ -1,13 +1,11 @@
 import click
 import subprocess
-from logger import logging, setup_logging, verbose_option
+from logger import logging
+from ssh import cmd_ssh
 
 
 @click.command(name='forwarding')
-@verbose_option
-def cmd_forwarding(verbose):
-    setup_logging(verbose)
-
+def cmd_forwarding():
     result = subprocess.run([
         'sysctl',
         'net.ipv4.ip_forward=1',
@@ -41,18 +39,7 @@ def cmd_forwarding(verbose):
         logging.fatal(f'Failed set iptables rule')
         exit(1)
 
-    result = subprocess.run([
-        'ssh',
-        '-o',
-        'GlobalKnownHostsFile=/dev/null',
-        '-o',
-        'UserKnownHostsFile=/dev/null',
-        '-o',
-        'StrictHostKeyChecking=no',
-        '-t',
-        'kupfer@172.16.42.1',
-        'sudo route add default gw 172.16.42.2',
-    ])
+    result = cmd_ssh(cmd=['sudo route add default gw 172.16.42.2'])
     if result.returncode != 0:
         logging.fatal(f'Failed to add gateway over ssh')
         exit(1)

image.py

@@ -2,7 +2,7 @@ import atexit
 import os
 import subprocess
 import click
-from logger import logging, setup_logging, verbose_option
+from logger import logging
 from chroot import create_chroot, create_chroot_user
 from constants import DEVICES, FLAVOURS
 
@@ -109,11 +109,8 @@ def cmd_image():
 
 
 @click.command(name='device')
-@verbose_option
 @click.argument('device')
-def cmd_device(verbose, device):
-    setup_logging(verbose)
-
+def cmd_device(device):
     for key in DEVICES.keys():
         if '-'.join(key.split('-')[1:]) == device:
             device = key
@@ -130,11 +127,8 @@ def cmd_device(verbose, device):
 
 
 @click.command(name='flavour')
-@verbose_option
 @click.argument('flavour')
-def cmd_flavour(verbose, flavour):
-    setup_logging(verbose)
-
+def cmd_flavour(flavour):
     if flavour not in FLAVOURS:
         logging.fatal(f'Unknown flavour {flavour}. Pick one from:\n{", ".join(FLAVOURS.keys())}')
         exit(1)
@@ -146,10 +140,7 @@ def cmd_flavour(verbose, flavour):
 
 
 @click.command(name='build')
-@verbose_option
-def cmd_build(verbose):
-    setup_logging(verbose)
-
+def cmd_build():
     device, flavour = get_device_and_flavour()
     image_name = get_image_name(device, flavour)
 
@@ -209,10 +200,7 @@ This doesn't work, because the mount isn't passed through to the real host
 """
 """
 @click.command(name='inspect')
-@verbose_option
-def cmd_inspect(verbose):
-    setup_logging(verbose)
-
+def cmd_inspect():
     device, flavour = get_device_and_flavour()
     image_name = get_image_name(device, flavour)
 

logger.py

@@ -2,17 +2,15 @@ import click
 import logging
 import sys
 
-
 def setup_logging(verbose: bool):
-    level = logging.INFO
-    if verbose:
-        level = logging.DEBUG
+    level = logging.DEBUG if verbose else logging.INFO
     logging.basicConfig(
         stream=sys.stdout,
         format='%(asctime)s %(levelname)s: %(message)s',
         datefmt='%m/%d/%Y %H:%M:%S',
         level=level,
     )
+    logging.debug('Logging set up.')
 
 
 verbose_option = click.option(

main.py

@@ -6,13 +6,21 @@ from flash import cmd_flash
 from ssh import cmd_ssh
 from forwarding import cmd_forwarding
 from telnet import cmd_telnet
+from logger import setup_logging, verbose_option
 import click
+from config import config, config_option
 
 
 @click.group()
-def cli():
-    pass
+@verbose_option
+@config_option
+def cli(verbose: bool = False, config_file: str = None):
+    setup_logging(verbose)
+    config.runtime['verbose'] = verbose
+    config.try_load_file(config_file)
 
+def main():
+    return cli(prog_name='kupferbootstrap')
 
 cli.add_command(cmd_cache)
 cli.add_command(cmd_packages)

packages.py

@@ -1,5 +1,4 @@
 from constants import REPOSITORIES
-from logger import setup_logging, verbose_option
 import atexit
 import click
 import logging
@@ -463,11 +462,8 @@ def cmd_packages():
 
 
 @click.command(name='build')
-@verbose_option
 @click.argument('paths', nargs=-1)
-def cmd_build(verbose, paths):
-    setup_logging(verbose)
-
+def cmd_build(paths):
     check_prebuilts()
 
     paths = list(paths)
@@ -492,9 +488,7 @@ def cmd_build(verbose, paths):
 
 
 @click.command(name='clean')
-@verbose_option
-def cmd_clean(verbose):
-    setup_logging(verbose)
+def cmd_clean():
     result = subprocess.run([
         'git',
         'clean',
@@ -506,11 +500,8 @@ def cmd_clean(verbose):
 
 
 @click.command(name='check')
-@verbose_option
 @click.argument('paths', nargs=-1)
-def cmd_check(verbose, paths):
-    setup_logging(verbose)
-
+def cmd_check(paths):
     paths = list(paths)
     packages = discover_packages(paths)
 

ssh.py

@@ -1,14 +1,10 @@
 import subprocess
 import click
-from logger import setup_logging, verbose_option
 
 
 @click.command(name='ssh')
-@verbose_option
-def cmd_ssh(verbose):
-    setup_logging(verbose)
-
-    subprocess.run([
+def cmd_ssh(cmd: list[str] = [], host: str = '172.16.42.1', user: str = 'kupfer', port: int = 22):
+    return subprocess.run([
         'ssh',
         '-o',
         'GlobalKnownHostsFile=/dev/null',
@@ -16,5 +12,8 @@ def cmd_ssh(verbose):
         'UserKnownHostsFile=/dev/null',
         '-o',
         'StrictHostKeyChecking=no',
-        'kupfer@172.16.42.1',
-    ])
+        '-p',
+        str(port),
+        f'{user}@{host}',
+        '--',
+    ] + cmd)

telnet.py

@@ -1,14 +1,10 @@
 import subprocess
 import click
-from logger import setup_logging, verbose_option
 
 
 @click.command(name='telnet')
-@verbose_option
-def cmd_telnet(verbose):
-    setup_logging(verbose)
-
+def cmd_telnet(hostname: str = '172.16.42.1'):
     subprocess.run([
         'telnet',
-        '172.16.42.1',
+        hostname,
     ])

wrapper.py

@@ -6,8 +6,8 @@ import appdirs
 import uuid
 
 if os.getenv('KUPFERBOOTSTRAP_DOCKER') == '1':
-    from main import cli
-    cli(prog_name='kupferbootstrap')
+    from main import main
+    main()
 else:
     script_path = os.path.dirname(os.path.abspath(__file__))
     with open(os.path.join(script_path, 'version.txt')) as version_file: