chore: update netplan import semantics and related tests

cloudinit/config/schema.py

@@ -50,6 +50,13 @@
 except ImportError:
     ValidationError = Exception  # type: ignore
 
+try:
+    from netplan import NetplanParserException, Parser  # type: ignore
+
+    LIBNETPLAN_AVAILABLE = True
+except ImportError:
+    LIBNETPLAN_AVAILABLE = False
+
 
 LOG = logging.getLogger(__name__)
 
@@ -615,14 +622,13 @@ def netplan_validate_network_schema(
     @raises: SchemaValidationError when netplan's parser raises
         NetplanParserExceptions.
     """
-    try:
-        from netplan import NetplanParserException, Parser  # type: ignore
-    except ImportError:
+    if LIBNETPLAN_AVAILABLE:
+        LOG.debug("Validating network-config with netplan API")
+    else:
         LOG.debug(
             "Skipping netplan schema validation. No netplan API available"
         )
         return False
-
     # netplan Parser looks at all *.yaml files in the target directory underA
     # /etc/netplan. cloud-init should only validate schema of the
     # network-config it generates, so create a <tmp_dir>/etc/netplan

tests/unittests/config/test_schema.py

@@ -430,12 +430,11 @@ class TestNetplanValidateNetworkSchema:
         ),
     )
     def test_network_config_schema_validation_false_when_skipped(
-        self, config, expected_log, caplog
+        self, config, expected_log, caplog, mocker
     ):
         """netplan_validate_network_schema returns false when skipped."""
-        with mock.patch.dict("sys.modules"):
-            sys.modules.pop("netplan", None)
-            assert False is netplan_validate_network_schema(config)
+        mocker.patch(f"{M_PATH}LIBNETPLAN_AVAILABLE", False)
+        assert False is netplan_validate_network_schema(config)
         assert expected_log in caplog.text
 
     @pytest.mark.parametrize(
@@ -456,9 +455,8 @@ def test_network_config_schema_validation_false_when_skipped(
         ),
     )
     def test_network_config_schema_validation(
-        self, error, error_log, caplog, tmpdir
+        self, error, error_log, caplog, tmpdir, mocker
     ):
-
         fake_tmpdir = tmpdir.join("mkdtmp")
 
         class FakeParser:
@@ -469,21 +467,21 @@ def load_yaml_hierarchy(self, parse_dir):
                     raise error
 
         # Mock expected imports
-        with mock.patch.dict(
-            "sys.modules",
-            netplan=mock.MagicMock(
-                NetplanParserException=FakeNetplanParserException,
-                Parser=FakeParser,
-            ),
-        ):
-            with mock.patch(
-                "cloudinit.config.schema.mkdtemp",
-                return_value=fake_tmpdir.strpath,
-            ):
-                with caplog.at_level(logging.WARNING):
-                    assert netplan_validate_network_schema({"version": 2})
-            if error_log:
-                assert re.match(error_log, caplog.records[0].msg, re.DOTALL)
+        mocker.patch(f"{M_PATH}LIBNETPLAN_AVAILABLE", True)
+        mocker.patch(
+            f"{M_PATH}NetplanParserException",
+            FakeNetplanParserException,
+            create=True,
+        )
+        mocker.patch(f"{M_PATH}Parser", FakeParser, create=True)
+        mocker.patch(
+            "cloudinit.config.schema.mkdtemp",
+            return_value=fake_tmpdir.strpath,
+        )
+        with caplog.at_level(logging.WARNING):
+            assert netplan_validate_network_schema({"version": 2})
+        if error_log:
+            assert re.match(error_log, caplog.records[0].msg, re.DOTALL)
 
 
 class TestValidateCloudConfigSchema:
@@ -1995,9 +1993,10 @@ def test_main_validates_config_file(
         expected,
         tmpdir,
         capsys,
-        caplog,
+        mocker,
     ):
         """When --config-file parameter is provided, main validates schema."""
+        mocker.patch(f"{M_PATH}LIBNETPLAN_AVAILABLE", False)
         myyaml = tmpdir.join("my.yaml")
         myargs = ["mycmd", "--config-file", myyaml.strpath]
         if schema_type:
@@ -2172,6 +2171,7 @@ def test_main_validates_system_userdata_vendordata_and_network_config(
         paths,
     ):
         """When --system is provided, main validates all config userdata."""
+        mocker.patch(f"{M_PATH}LIBNETPLAN_AVAILABLE", False)
         paths.get_ipath = paths.get_ipath_cur
         read_cfg_paths.return_value = paths
         cloud_config_file = paths.get_ipath_cur("cloud_config")
@@ -2562,7 +2562,9 @@ def test_network_schema(
         expectation,
         log,
         caplog,
+        mocker,
     ):
+        mocker.patch(f"{M_PATH}LIBNETPLAN_AVAILABLE", False)
         net_schema = get_schema(schema_type=schema_type_version)
         with expectation:
             validate_cloudconfig_schema(