Merge pull request vyos#3692 from jestabro/revise-migration

T6007: revise migration system

Makefile

@@ -78,7 +78,7 @@ vyshim:
 	$(MAKE) -C $(SHIM_DIR)
 
 .PHONY: all
-all: clean interface_definitions op_mode_definitions test j2lint vyshim check_migration_scripts_executable generate-configd-include-json
+all: clean interface_definitions op_mode_definitions test j2lint vyshim generate-configd-include-json
 
 .PHONY: clean
 clean:

python/vyos/base.py

@@ -63,3 +63,10 @@ def __init__(self, message):
         message = fill(message, width=72)
         # Call the base class constructor with the parameters it needs
         super().__init__(message)
+
+class MigrationError(Exception):
+    def __init__(self, message):
+        # Reformat the message and trim it to 72 characters in length
+        message = fill(message, width=72)
+        # Call the base class constructor with the parameters it needs
+        super().__init__(message)

python/vyos/component_version.py

@@ -1,4 +1,4 @@
-# Copyright 2022 VyOS maintainers and contributors <[email protected]>
+# Copyright 2022-2024 VyOS maintainers and contributors <[email protected]>
 #
 # This library is free software; you can redistribute it and/or
 # modify it under the terms of the GNU Lesser General Public
@@ -35,133 +35,173 @@
 import os
 import re
 import sys
-import fileinput
+from dataclasses import dataclass
+from dataclasses import replace
+from typing import Optional
 
 from vyos.xml_ref import component_version
+from vyos.utils.file import write_file
 from vyos.version import get_version
 from vyos.defaults import directories
 
 DEFAULT_CONFIG_PATH = os.path.join(directories['config'], 'config.boot')
 
-def from_string(string_line, vintage='vyos'):
-    """
-    Get component version dictionary from string.
-    Return empty dictionary if string contains no config information
-    or raise error if component version string malformed.
-    """
-    version_dict = {}
-
-    if vintage == 'vyos':
-        if re.match(r'// vyos-config-version:.+', string_line):
-            if not re.match(r'// vyos-config-version:\s+"([\w,-]+@\d+:)+([\w,-]+@\d+)"\s*', string_line):
-                raise ValueError(f"malformed configuration string: {string_line}")
+REGEX_WARN_VYOS = r'(// Warning: Do not remove the following line.)'
+REGEX_WARN_VYATTA = r'(/\* Warning: Do not remove the following line. \*/)'
+REGEX_COMPONENT_VERSION_VYOS = r'// vyos-config-version:\s+"([\w@:-]+)"\s*'
+REGEX_COMPONENT_VERSION_VYATTA = r'/\* === vyatta-config-version:\s+"([\w@:-]+)"\s+=== \*/'
+REGEX_RELEASE_VERSION_VYOS = r'// Release version:\s+(\S*)\s*'
+REGEX_RELEASE_VERSION_VYATTA = r'/\* Release version:\s+(\S*)\s*\*/'
 
-            for pair in re.findall(r'([\w,-]+)@(\d+)', string_line):
-                version_dict[pair[0]] = int(pair[1])
-
-    elif vintage == 'vyatta':
-        if re.match(r'/\* === vyatta-config-version:.+=== \*/$', string_line):
-            if not re.match(r'/\* === vyatta-config-version:\s+"([\w,-]+@\d+:)+([\w,-]+@\d+)"\s+=== \*/$', string_line):
-                raise ValueError(f"malformed configuration string: {string_line}")
+CONFIG_FILE_VERSION = """\
+// Warning: Do not remove the following line.
+// vyos-config-version: "{}"
+// Release version: {}
+"""
 
-            for pair in re.findall(r'([\w,-]+)@(\d+)', string_line):
-                version_dict[pair[0]] = int(pair[1])
+warn_filter_vyos = re.compile(REGEX_WARN_VYOS)
+warn_filter_vyatta = re.compile(REGEX_WARN_VYATTA)
+
+regex_filter = { 'vyos': dict(zip(['component', 'release'],
+                                  [re.compile(REGEX_COMPONENT_VERSION_VYOS),
+                                   re.compile(REGEX_RELEASE_VERSION_VYOS)])),
+                 'vyatta': dict(zip(['component', 'release'],
+                                    [re.compile(REGEX_COMPONENT_VERSION_VYATTA),
+                                     re.compile(REGEX_RELEASE_VERSION_VYATTA)])) }
+
+@dataclass
+class VersionInfo:
+    component: Optional[dict[str,int]] = None
+    release: str = get_version()
+    vintage: str = 'vyos'
+    config_body: Optional[str] = None
+    footer_lines: Optional[list[str]] = None
+
+    def component_is_none(self) -> bool:
+        return bool(self.component is None)
+
+    def config_body_is_none(self) -> bool:
+        return bool(self.config_body is None)
+
+    def update_footer(self):
+        f = CONFIG_FILE_VERSION.format(component_to_string(self.component),
+                                       self.release)
+        self.footer_lines = f.splitlines()
+
+    def update_syntax(self):
+        self.vintage = 'vyos'
+        self.update_footer()
+
+    def update_release(self, release: str):
+        self.release = release
+        self.update_footer()
+
+    def update_component(self, key: str, version: int):
+        if not isinstance(version, int):
+            raise ValueError('version must be int')
+        if self.component is None:
+            self.component = {}
+        self.component[key] = version
+        self.component = dict(sorted(self.component.items(), key=lambda x: x[0]))
+        self.update_footer()
+
+    def update_config_body(self, config_str: str):
+        self.config_body = config_str
+
+    def write_string(self) -> str:
+        config_body = '' if self.config_body is None else self.config_body
+        footer_lines = [] if self.footer_lines is None else self.footer_lines
+
+        return config_body + '\n' + '\n'.join(footer_lines) + '\n'
+
+    def write(self, config_file):
+        string = self.write_string()
+        try:
+            write_file(config_file, string)
+        except Exception as e:
+            raise ValueError(e) from e
+
+def component_to_string(component: dict) -> str:
+    l = [f'{k}@{v}' for k, v in sorted(component.items(), key=lambda x: x[0])]
+    return ':'.join(l)
+
+def component_from_string(string: str) -> dict:
+    return {k: int(v) for k, v in re.findall(r'([\w,-]+)@(\d+)', string)}
+
+def version_info_from_file(config_file) -> VersionInfo:
+    version_info = VersionInfo()
+    try:
+        with open(config_file) as f:
+            config_str = f.read()
+    except OSError:
+        return None
+
+    if len(parts := warn_filter_vyos.split(config_str)) > 1:
+        vintage = 'vyos'
+    elif len(parts := warn_filter_vyatta.split(config_str)) > 1:
+        vintage = 'vyatta'
     else:
-        raise ValueError("Unknown config string vintage")
+        version_info.config_body = parts[0] if parts else None
+        return version_info
 
-    return version_dict
+    version_info.vintage = vintage
+    version_info.config_body = parts[0]
+    version_lines = ''.join(parts[1:]).splitlines()
+    version_lines = [k for k in version_lines if k]
+    if len(version_lines) != 3:
+        raise ValueError(f'Malformed version strings: {version_lines}')
 
-def from_file(config_file_name=DEFAULT_CONFIG_PATH, vintage='vyos'):
-    """
-    Get component version dictionary parsing config file line by line
-    """
-    with open(config_file_name, 'r') as f:
-        for line_in_config in f:
-            version_dict = from_string(line_in_config, vintage=vintage)
-            if version_dict:
-                return version_dict
+    m = regex_filter[vintage]['component'].match(version_lines[1])
+    if not m:
+        raise ValueError(f'Malformed component string: {version_lines[1]}')
+    version_info.component = component_from_string(m.group(1))
 
-    # no version information
-    return {}
+    m = regex_filter[vintage]['release'].match(version_lines[2])
+    if not m:
+        raise ValueError(f'Malformed component string: {version_lines[2]}')
+    version_info.release = m.group(1)
 
-def from_system():
-    """
-    Get system component version dict.
-    """
-    return component_version()
+    version_info.footer_lines = version_lines
 
-def format_string(ver: dict) -> str:
-    """
-    Version dict to string.
-    """
-    keys = list(ver)
-    keys.sort()
-    l = []
-    for k in keys:
-        v = ver[k]
-        l.append(f'{k}@{v}')
-    sep = ':'
-    return sep.join(l)
-
-def version_footer(ver: dict, vintage='vyos') -> str:
-    """
-    Version footer as string.
-    """
-    ver_str = format_string(ver)
-    release = get_version()
-    if vintage == 'vyos':
-        ret_str = (f'// Warning: Do not remove the following line.\n'
-                +  f'// vyos-config-version: "{ver_str}"\n'
-                +  f'// Release version: {release}\n')
-    elif vintage == 'vyatta':
-        ret_str = (f'/* Warning: Do not remove the following line. */\n'
-                +  f'/* === vyatta-config-version: "{ver_str}" === */\n'
-                +  f'/* Release version: {release} */\n')
-    else:
-        raise ValueError("Unknown config string vintage")
+    return version_info
 
-    return ret_str
-
-def system_footer(vintage='vyos') -> str:
+def version_info_from_system() -> VersionInfo:
     """
-    System version footer as string.
+    Return system component versions.
     """
-    ver_d = from_system()
-    return version_footer(ver_d, vintage=vintage)
+    d = component_version()
+    sort_d = dict(sorted(d.items(), key=lambda x: x[0]))
+    version_info = VersionInfo(
+        component = sort_d,
+        release =  get_version(),
+        vintage = 'vyos'
+    )
+
+    return version_info
 
-def write_version_footer(ver: dict, file_name, vintage='vyos'):
+def version_info_copy(v: VersionInfo) -> VersionInfo:
     """
-    Write version footer to file.
+    Make a copy of dataclass.
     """
-    footer = version_footer(ver=ver, vintage=vintage)
-    if file_name:
-        with open(file_name, 'a') as f:
-            f.write(footer)
-    else:
-        sys.stdout.write(footer)
+    return replace(v)
 
-def write_system_footer(file_name, vintage='vyos'):
+def version_info_prune_component(x: VersionInfo, y: VersionInfo) -> VersionInfo:
     """
-    Write system version footer to file.
+    In place pruning of component keys of x not in y.
     """
-    ver_d = from_system()
-    return write_version_footer(ver_d, file_name=file_name, vintage=vintage)
+    x.component = { k: v for k,v in x.component.items() if k in y.component }
 
-def remove_footer(file_name):
+def add_system_version(config_str: str = None, out_file: str = None):
     """
-    Remove old version footer.
+    Wrap config string with system version and write to out_file.
+    For convenience, calling with no argument will write system version
+    string to stdout, for use in bash scripts.
     """
-    for line in fileinput.input(file_name, inplace=True):
-        if re.match(r'/\* Warning:.+ \*/$', line):
-            continue
-        if re.match(r'/\* === vyatta-config-version:.+=== \*/$', line):
-            continue
-        if re.match(r'/\* Release version:.+ \*/$', line):
-            continue
-        if re.match('// vyos-config-version:.+', line):
-            continue
-        if re.match('// Warning:.+', line):
-            continue
-        if re.match('// Release version:.+', line):
-            continue
-        sys.stdout.write(line)
+    version_info = version_info_from_system()
+    if config_str is not None:
+        version_info.update_config_body(config_str)
+    version_info.update_footer()
+    if out_file is not None:
+        version_info.write(out_file)
+    else:
+        sys.stdout.write(version_info.write_string())

python/vyos/compose_config.py

@@ -17,12 +17,13 @@
 config.
 """
 
+import traceback
 from pathlib import Path
 from typing import TypeAlias, Union, Callable
 
 from vyos.configtree import ConfigTree
 from vyos.configtree import deep_copy as ct_deep_copy
-from vyos.utils.system import load_as_module
+from vyos.utils.system import load_as_module_source
 
 ConfigObj: TypeAlias = Union[str, ConfigTree]
 
@@ -54,23 +55,26 @@ def apply_func(self, func: Callable):
         try:
             func(self.config_tree)
         except Exception as e:
-            self.config_tree = self.checkpoint
+            if self.checkpoint_file is not None:
+                self.config_tree = self.checkpoint
             raise ComposeConfigError(e) from e
 
     def apply_file(self, func_file: str, func_name: str):
         """Apply named function from file.
         """
         try:
             mod_name = Path(func_file).stem.replace('-', '_')
-            mod = load_as_module(mod_name, func_file)
+            mod = load_as_module_source(mod_name, func_file)
             func = getattr(mod, func_name)
         except Exception as e:
             raise ComposeConfigError(f'Error with {func_file}: {e}') from e
 
         try:
             self.apply_func(func)
         except ComposeConfigError as e:
-            raise ComposeConfigError(f'Error in {func_file}: {e}') from e
+            msg = str(e)
+            tb = f'{traceback.format_exc()}'
+            raise ComposeConfigError(f'Error in {func_file}: {msg}\n{tb}') from e
 
     def to_string(self, with_version=False) -> str:
         """Return the rendered config tree.