netplan_cli: add new --ignore-errors flag to generate and get commands

netplan_cli/cli/commands/generate.py

@@ -39,6 +39,8 @@ def run(self):
                                  help='Search for and generate configuration files in this root directory instead of /')
         self.parser.add_argument('--mapping',
                                  help='Display the netplan device ID/backend/interface name mapping and exit.')
+        self.parser.add_argument('--ignore-errors', action='store_true',
+                                 help='Ignores files and/or interfaces that fail parsing.')
 
         self.func = self.command_generate
 
@@ -80,6 +82,8 @@ def command_generate(self):
             argv += ['--root-dir', self.root_dir]
         if self.mapping:
             argv += ['--mapping', self.mapping]
+        if self.ignore_errors:
+            argv += ['--ignore-errors']
         logging.debug('command generate: running %s', argv)
         # FIXME: os.execv(argv[0], argv) would be better but fails coverage
         sys.exit(subprocess.call(argv))

netplan_cli/cli/commands/get.py

@@ -31,12 +31,14 @@ def run(self):
         self.parser.add_argument('key', type=str, nargs='?', default='all', help='The nested key in dotted format')
         self.parser.add_argument('--root-dir', default='/',
                                  help='Read configuration files from this root directory instead of /')
+        self.parser.add_argument('--ignore-errors', action='store_true',
+                                 help='Ignores files and/or interfaces that fail parsing.')
 
         self.func = self.command_get
 
         self.parse_args()
         self.run_command()
 
     def command_get(self):
-        state_data = NetplanConfigState(self.key, self.root_dir)
+        state_data = NetplanConfigState(self.key, self.root_dir, self.ignore_errors)
         print(state_data, end='')

netplan_cli/cli/state.py

@@ -576,9 +576,9 @@ def get_data(self) -> dict:
 class NetplanConfigState():
     ''' Collects the Netplan's network configuration '''
 
-    def __init__(self, subtree='all', rootdir='/'):
+    def __init__(self, subtree='all', rootdir='/', ignore_errors=False):
 
-        parser = netplan.Parser()
+        parser = netplan.Parser(ignore_errors=ignore_errors)
         parser.load_yaml_hierarchy(rootdir)
 
         np_state = netplan.State()

python-cffi/netplan/_build_cffi.py

@@ -68,6 +68,7 @@
     // Parser
     NetplanParser* netplan_parser_new();
     void netplan_parser_clear(NetplanParser **npp);
+    void netplan_parser_set_flags(NetplanParser *npp, int flags);
     gboolean netplan_parser_load_yaml(NetplanParser* npp, const char* filename, NetplanError** error);
     gboolean netplan_parser_load_yaml_from_fd(NetplanParser* npp, int input_fd, NetplanError** error);
     gboolean netplan_parser_load_yaml_hierarchy(NetplanParser* npp, const char* rootdir, NetplanError** error);

python-cffi/netplan/parser.py

@@ -20,9 +20,12 @@
 
 
 class Parser():
-    def __init__(self):
+    def __init__(self, ignore_errors=False):
         self._ptr = lib.netplan_parser_new()
 
+        if ignore_errors:
+            lib.netplan_parser_set_flags(self._ptr, 1)
+
     def __del__(self):
         ref = ffi.new('NetplanParser **', self._ptr)
         lib.netplan_parser_clear(ref)

tests/cli_legacy.py

@@ -116,6 +116,23 @@ def test_with_config(self):
         self.assertEqual(os.listdir(os.path.join(self.workdir.name, 'run', 'systemd', 'network')),
                          ['10-netplan-enlol.network'])
 
+    def test_with_ignore_errors(self):
+        ''' Check that the --ignore-errors works. A file for the ethbad interface should not be generated '''
+        c = os.path.join(self.workdir.name, 'etc', 'netplan')
+        os.makedirs(c)
+        with open(os.path.join(c, 'a.yaml'), 'w') as f:
+            f.write('''network:
+  version: 2
+  ethernets:
+    ethbad:
+      dhcp4: yesplease
+    enlol:
+      dhcp4: true''')
+        out = subprocess.check_output(exe_cli + ['generate', '--ignore-errors', '--root-dir', self.workdir.name])
+        self.assertEqual(out, b'')
+        self.assertEqual(os.listdir(os.path.join(self.workdir.name, 'run', 'systemd', 'network')),
+                         ['10-netplan-enlol.network'])
+
     def test_mapping_for_unknown_iface(self):
         os.environ.setdefault('NETPLAN_GENERATE_PATH', os.path.join(rootdir, 'generate'))
         c = os.path.join(self.workdir.name, 'etc', 'netplan')