Single JSON schema validation in early boot (#1175)

Package a single JSON schema file for user-data validation at
cloudinit/config/cloud-init-schema.json.

Perform validate_cloudconfig_schema call to just after the
user-data is consumed. This will allow single validation of all
user-data against the full schema instead of
repetitive validatation calls against each cloud-config module
(cloudinit.config.cc_*) sub-schemas.

This branch defines the simple apt_pipelining schema and
migrates existing cc_apk_configure into cloud-init-schema.json.

The expectation will be additional branches to migrate from legacy
"schema" attributes inside each cloud-config module toward unique
cc_<module_name> definitions in the global shema file under "$defs"
of cloud-init-schema-X.Y..json.

Before legacy sub-schema definitions are migrated the following
funcs grew support to read sub-schemas from both static
cloud-init-schema.json and the individual cloud-config module
"schema" attributes:

- get_schema: source base schema file from cloud-init-schema.json
  and supplement with all legacy cloud-config module "schema" defs
- get_meta_doc: optional schema param so cloud-config modules
  no longer provide the own local sub-schemas
- _get_property_doc: render only documentation of sub-schema based
  on meta['id'] provided
- validate_cloudconfig_schema: allow optional schema param


Additionally, fix two minor bugs in _schemapath_for_cloudconfig:
- `cloud-init devel schema --annotate` which results in a Traceback
  if two keys at the same indent level have invalid types.
- exit early on empty cloud-config to avoid a Traceback on the CLI

18 files changed, 558 insertions, 254 deletions

diff --git a/cloudinit/cmd/main.py b/cloudinit/cmd/main.py
index e67edbc3..c9be41b3 100644
--- a/cloudinit/cmd/main.py
+++ b/cloudinit/cmd/main.py
@@ -22,6 +22,7 @@ from cloudinit import patcher
 
 patcher.patch_logging()
 
+from cloudinit.config.schema import validate_cloudconfig_schema
 from cloudinit import log as logging
 from cloudinit import netinfo
 from cloudinit import signal_handler
@@ -474,6 +475,12 @@ def main_init(name, args):
         util.logexc(LOG, "Consuming user data failed!")
         return (init.datasource, ["Consuming user data failed!"])
 
+    # Validate user-data adheres to schema definition
+    if os.path.exists(init.paths.get_ipath_cur("userdata_raw")):
+        validate_cloudconfig_schema(config=init.cfg, strict=False)
+    else:
+        LOG.debug("Skipping user-data validation. No user-data found.")
+
     apply_reporting_cfg(init.cfg)
 
     # Stage 8 - re-read and apply relevant cloud-config to include user-data
diff --git a/cloudinit/config/cc_apk_configure.py b/cloudinit/config/cc_apk_configure.py
index a615c814..2cb2dad1 100644
--- a/cloudinit/config/cc_apk_configure.py
+++ b/cloudinit/config/cc_apk_configure.py
@@ -10,7 +10,7 @@ from textwrap import dedent
 
 from cloudinit import log as logging
 from cloudinit import temp_utils, templater, util
-from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
+from cloudinit.config.schema import get_meta_doc
 from cloudinit.settings import PER_INSTANCE
 
 LOG = logging.getLogger(__name__)
@@ -102,104 +102,7 @@ meta = {
     "frequency": frequency,
 }
 
-schema = {
-    "type": "object",
-    "properties": {
-        "apk_repos": {
-            "type": "object",
-            "properties": {
-                "preserve_repositories": {
-                    "type": "boolean",
-                    "default": False,
-                    "description": dedent(
-                        """\
-                        By default, cloud-init will generate a new repositories
-                        file ``/etc/apk/repositories`` based on any valid
-                        configuration settings specified within a apk_repos
-                        section of cloud config. To disable this behavior and
-                        preserve the repositories file from the pristine image,
-                        set ``preserve_repositories`` to ``true``.
-
-                        The ``preserve_repositories`` option overrides
-                        all other config keys that would alter
-                        ``/etc/apk/repositories``.
-                    """
-                    ),
-                },
-                "alpine_repo": {
-                    "type": ["object", "null"],
-                    "properties": {
-                        "base_url": {
-                            "type": "string",
-                            "default": DEFAULT_MIRROR,
-                            "description": dedent(
-                                """\
-                                The base URL of an Alpine repository, or
-                                mirror, to download official packages from.
-                                If not specified then it defaults to ``{}``
-                            """.format(
-                                    DEFAULT_MIRROR
-                                )
-                            ),
-                        },
-                        "community_enabled": {
-                            "type": "boolean",
-                            "default": False,
-                            "description": dedent(
-                                """\
-                                Whether to add the Community repo to the
-                                repositories file. By default the Community
-                                repo is not included.
-                            """
-                            ),
-                        },
-                        "testing_enabled": {
-                            "type": "boolean",
-                            "default": False,
-                            "description": dedent(
-                                """\
-                                Whether to add the Testing repo to the
-                                repositories file. By default the Testing
-                                repo is not included. It is only recommended
-                                to use the Testing repo on a machine running
-                                the ``Edge`` version of Alpine as packages
-                                installed from Testing may have dependancies
-                                that conflict with those in non-Edge Main or
-                                Community repos."
-                            """
-                            ),
-                        },
-                        "version": {
-                            "type": "string",
-                            "description": dedent(
-                                """\
-                                The Alpine version to use (e.g. ``v3.12`` or
-                                ``edge``)
-                            """
-                            ),
-                        },
-                    },
-                    "required": ["version"],
-                    "minProperties": 1,
-                    "additionalProperties": False,
-                },
-                "local_repo_base_url": {
-                    "type": "string",
-                    "description": dedent(
-                        """\
-                        The base URL of an Alpine repository containing
-                        unofficial packages
-                    """
-                    ),
-                },
-            },
-            "minProperties": 1,  # Either preserve_repositories or alpine_repo
-            "additionalProperties": False,
-        }
-    },
-}
-
-__doc__ = get_meta_doc(meta, schema)
+__doc__ = get_meta_doc(meta)
 
 
 def handle(name, cfg, cloud, log, _args):
@@ -222,8 +125,6 @@ def handle(name, cfg, cloud, log, _args):
         )
         return
 
-    validate_cloudconfig_schema(cfg, schema)
-
     # If "preserve_repositories" is explicitly set to True in
     # the configuration do nothing.
     if util.get_cfg_option_bool(apk_section, "preserve_repositories", False):
diff --git a/cloudinit/config/cc_apt_pipelining.py b/cloudinit/config/cc_apt_pipelining.py
index 569849d1..34b6ac0e 100644
--- a/cloudinit/config/cc_apt_pipelining.py
+++ b/cloudinit/config/cc_apt_pipelining.py
@@ -4,54 +4,59 @@
 #
 # This file is part of cloud-init. See LICENSE file for license information.
 
-"""
-Apt Pipelining
---------------
-**Summary:** configure apt pipelining
+"""Apt Pipelining: configure apt pipelining."""
 
-This module configures apt's ``Acquite::http::Pipeline-Depth`` option, which
-controls how apt handles HTTP pipelining. It may be useful for pipelining to be
-disabled, because some web servers, such as S3 do not pipeline properly (LP:
-#948461). The ``apt_pipelining`` config key may be set to ``false`` to disable
-pipelining altogether. This is the default behavior. If it is set to ``none``,
-``unchanged``, or ``os``, no change will be made to apt configuration and the
-default setting for the distro will be used. The pipeline depth can also be
-manually specified by setting ``apt_pipelining`` to a number. However, this is
-not recommended.
-
-**Internal name:** ``cc_apt_pipelining``
-
-**Module frequency:** per instance
-
-**Supported distros:** ubuntu, debian
-
-**Config keys**::
-    apt_pipelining: <false/none/unchanged/os/number>
-"""
+from textwrap import dedent
 
 from cloudinit import util
+from cloudinit.config.schema import get_meta_doc
 from cloudinit.settings import PER_INSTANCE
 
 frequency = PER_INSTANCE
-
 distros = ["ubuntu", "debian"]
-
 DEFAULT_FILE = "/etc/apt/apt.conf.d/90cloud-init-pipelining"
-
 APT_PIPE_TPL = (
     "//Written by cloud-init per 'apt_pipelining'\n"
     'Acquire::http::Pipeline-Depth "%s";\n'
 )
-
 # Acquire::http::Pipeline-Depth can be a value
 # from 0 to 5 indicating how many outstanding requests APT should send.
 # A value of zero MUST be specified if the remote host does not properly linger
 # on TCP connections - otherwise data corruption will occur.
 
+meta = {
+    "id": "cc_apt_pipelining",
+    "name": "Apt Pipelining",
+    "title": "Configure apt pipelining",
+    "description": dedent(
+        """\
+        This module configures apt's ``Acquite::http::Pipeline-Depth`` option,
+        which controls how apt handles HTTP pipelining. It may be useful for
+        pipelining to be disabled, because some web servers, such as S3 do not
+        pipeline properly (LP: #948461).
+
+        Value configuration options for this module are:
+
+        * ``false`` (Default): disable pipelining altogether
+        * ``none``, ``unchanged``, or ``os``: use distro default
+        * ``<number>``: Manually specify pipeline depth. This is not recommended."""  # noqa: E501
+    ),
+    "distros": distros,
+    "frequency": frequency,
+    "examples": [
+        "apt_pipelining: false",
+        "apt_pipelining: none",
+        "apt_pipelining: unchanged",
+        "apt_pipelining: os",
+        "apt_pipelining: 3",
+    ],
+}
+
+__doc__ = get_meta_doc(meta)
 
-def handle(_name, cfg, _cloud, log, _args):
 
-    apt_pipe_value = util.get_cfg_option_str(cfg, "apt_pipelining", "os")
+def handle(_name, cfg, _cloud, log, _args):
+    apt_pipe_value = cfg.get("apt_pipelining", "os")
     apt_pipe_value_s = str(apt_pipe_value).lower().strip()
 
     if apt_pipe_value_s == "false":
diff --git a/cloudinit/config/cloud-init-schema.json b/cloudinit/config/cloud-init-schema.json
new file mode 100644
index 00000000..afaed285
--- /dev/null
+++ b/cloudinit/config/cloud-init-schema.json
@@ -0,0 +1,69 @@
+{
+  "$schema": "http://json-schema.org/draft-04/schema#",
+  "$defs": {
+    "cc_apk_configure": {
+      "type": "object",
+      "properties": {
+        "apk_repos": {
+          "type": "object",
+          "properties": {
+            "preserve_repositories": {
+              "type": "boolean",
+              "default": false,
+              "description": "By default, cloud-init will generate a new repositories file ``/etc/apk/repositories`` based on any valid configuration settings specified within a apk_repos section of cloud config. To disable this behavior and preserve the repositories file from the pristine image, set ``preserve_repositories`` to ``true``.\n\n The ``preserve_repositories`` option overrides all other config keys that would alter ``/etc/apk/repositories``."
+            },
+            "alpine_repo": {
+              "type": ["object", "null"],
+              "properties": {
+                "base_url": {
+                  "type": "string",
+                  "default": "https://alpine.global.ssl.fastly.net/alpine",
+                  "description": "The base URL of an Alpine repository, or mirror, to download official packages from. If not specified then it defaults to ``https://alpine.global.ssl.fastly.net/alpine``"
+                },
+                "community_enabled": {
+                  "type": "boolean",
+                  "default": false,
+                  "description": "Whether to add the Community repo to the repositories file. By default the Community repo is not included."
+                },
+                "testing_enabled": {
+                  "type": "boolean",
+                  "default": false,
+                  "description": "Whether to add the Testing repo to the repositories file. By default the Testing repo is not included. It is only recommended to use the Testing repo on a machine running the ``Edge`` version of Alpine as packages installed from Testing may have dependancies that conflict with those in non-Edge Main or Community repos."
+                },
+                "version": {
+                  "type": "string",
+                  "description": "The Alpine version to use (e.g. ``v3.12`` or ``edge``)"
+                }
+              },
+              "required": ["version"],
+              "minProperties": 1,
+              "additionalProperties": false
+            },
+            "local_repo_base_url": {
+              "type": "string",
+                "description": "The base URL of an Alpine repository containing unofficial packages"
+            }
+          },
+          "minProperties": 1,
+          "additionalProperties": false
+        }
+      }
+    },
+    "cc_apt_pipelining": {
+      "type": "object",
+      "properties": {
+        "apt_pipelining": {
+          "oneOf": [
+            {"type": "integer"},
+            {"type": "boolean"},
+            {"type": "string", "enum": ["none", "unchanged", "os"]}
+          ]
+        }
+      }
+    }
+  },
+  "allOf": [
+    { "$ref": "#/$defs/cc_apk_configure" },
+    { "$ref": "#/$defs/cc_apt_pipelining" }
+  ]
+}
diff --git a/cloudinit/config/schema.py b/cloudinit/config/schema.py
index f54cf18f..1f969c97 100644
--- a/cloudinit/config/schema.py
+++ b/cloudinit/config/schema.py
@@ -2,6 +2,7 @@
 """schema.py: Set of module functions for processing cloud-config schema."""
 
 import argparse
+import json
 import logging
 import os
 import re
@@ -166,14 +167,18 @@ def validate_cloudconfig_metaschema(validator, schema: dict, throw=True):
 
 
 def validate_cloudconfig_schema(
-    config: dict, schema: dict, strict=False, strict_metaschema=False
+    config: dict,
+    schema: dict = None,
+    strict: bool = False,
+    strict_metaschema: bool = False,
 ):
     """Validate provided config meets the schema definition.
 
     @param config: Dict of cloud configuration settings validated against
         schema. Ignored if strict_metaschema=True
     @param schema: jsonschema dict describing the supported schema definition
-       for the cloud config module (config.cc_*).
+       for the cloud config module (config.cc_*). If None, validate against
+       global schema.
     @param strict: Boolean, when True raise SchemaValidationErrors instead of
        logging warnings.
     @param strict_metaschema: Boolean, when True validates schema using strict
@@ -183,6 +188,8 @@ def validate_cloudconfig_schema(
         against the provided schema.
     @raises: RuntimeError when provided config sourced from YAML is not a dict.
     """
+    if schema is None:
+        schema = get_schema()
     try:
         (cloudinitValidator, FormatChecker) = get_jsonschema_validator()
         if strict_metaschema:
@@ -203,7 +210,9 @@ def validate_cloudconfig_schema(
             raise SchemaValidationError(errors)
         else:
             messages = ["{0}: {1}".format(k, msg) for k, msg in errors]
-            LOG.warning("Invalid config:\n%s", "\n".join(messages))
+            LOG.warning(
+                "Invalid cloud-config provided:\n%s", "\n".join(messages)
+            )
 
 
 def annotated_cloudconfig_file(cloudconfig, original_content, schema_errors):
@@ -347,12 +356,14 @@ def _schemapath_for_cloudconfig(config, original_content):
     @param config: The yaml.loaded config dictionary of a cloud-config file.
     @param original_content: The simple file content of the cloud-config file
     """
-    # FIXME Doesn't handle multi-line lists or multi-line strings
+    # TODO( handle multi-line lists or multi-line strings, inline dicts)
     content_lines = original_content.decode().split("\n")
     schema_line_numbers = {}
     list_index = 0
     RE_YAML_INDENT = r"^(\s*)"
     scopes = []
+    if not config:
+        return {}  # No YAML config dict, no schemapaths to annotate
     for line_number, line in enumerate(content_lines, 1):
         indent_depth = len(re.match(RE_YAML_INDENT, line).groups()[0])
         line = line.strip()
@@ -369,7 +380,6 @@ def _schemapath_for_cloudconfig(config, original_content):
             if path_prefix and path_prefix.endswith(previous_list_idx):
                 path_prefix = path_prefix[: -len(previous_list_idx)]
             key = str(list_index)
-            schema_line_numbers[key] = line_number
             item_indent = len(re.match(RE_YAML_INDENT, line[1:]).groups()[0])
             item_indent += 1  # For the leading '-' character
             previous_depth = indent_depth
@@ -380,7 +390,7 @@ def _schemapath_for_cloudconfig(config, original_content):
             # Process non-list lines setting value if present
             list_index = 0
             key, value = line.split(":", 1)
-        if path_prefix:
+        if path_prefix and indent_depth > previous_depth:
             # Append any existing path_prefix for a fully-pathed key
             key = path_prefix + "." + key
         while indent_depth <= previous_depth:
@@ -409,10 +419,17 @@ def _get_property_type(property_dict: dict) -> str:
     jsonschema.
     """
     property_type = property_dict.get("type")
-    if property_type is None and property_dict.get("enum"):
-        property_type = [
-            str(_YAML_MAP.get(k, k)) for k in property_dict["enum"]
-        ]
+    if property_type is None:
+        if property_dict.get("enum"):
+            property_type = [
+                str(_YAML_MAP.get(k, k)) for k in property_dict["enum"]
+            ]
+        elif property_dict.get("oneOf"):
+            property_type = [
+                subschema["type"]
+                for subschema in property_dict.get("oneOf")
+                if subschema.get("type")
+            ]
     if isinstance(property_type, list):
         property_type = "/".join(property_type)
     items = property_dict.get("items", {})
@@ -449,7 +466,7 @@ def _parse_description(description, prefix) -> str:
     return description
 
 
-def _get_property_doc(schema: dict, prefix="    ") -> str:
+def _get_property_doc(schema: dict, defs: dict, prefix="    ") -> str:
     """Return restructured text describing the supported schema properties."""
     new_prefix = prefix + "    "
     properties = []
@@ -460,6 +477,10 @@ def _get_property_doc(schema: dict, prefix="    ") -> str:
 
     for props in property_keys:
         for prop_key, prop_config in props.items():
+            if "$ref" in prop_config:
+                # Update the defined references in subschema for doc rendering
+                ref = defs[prop_config["$ref"].replace("#/$defs/", "")]
+                prop_config.update(ref)
             # Define prop_name and description for SCHEMA_PROPERTY_TMPL
             description = prop_config.get("description", "")
 
@@ -478,7 +499,9 @@ def _get_property_doc(schema: dict, prefix="    ") -> str:
                 if isinstance(items, list):
                     for item in items:
                         properties.append(
-                            _get_property_doc(item, prefix=new_prefix)
+                            _get_property_doc(
+                                item, defs=defs, prefix=new_prefix
+                            )
                         )
                 elif isinstance(items, dict) and (
                     items.get("properties") or items.get("patternProperties")
@@ -490,14 +513,16 @@ def _get_property_doc(schema: dict, prefix="    ") -> str:
                     )
                     new_prefix += "    "
                     properties.append(
-                        _get_property_doc(items, prefix=new_prefix)
+                        _get_property_doc(items, defs=defs, prefix=new_prefix)
                     )
             if (
                 "properties" in prop_config
                 or "patternProperties" in prop_config
             ):
                 properties.append(
-                    _get_property_doc(prop_config, prefix=new_prefix)
+                    _get_property_doc(
+                        prop_config, defs=defs, prefix=new_prefix
+                    )
                 )
     return "\n\n".join(properties)
 
@@ -520,15 +545,18 @@ def _get_examples(meta: MetaSchema) -> str:
     return rst_content
 
 
-def get_meta_doc(meta: MetaSchema, schema: dict) -> str:
+def get_meta_doc(meta: MetaSchema, schema: dict = None) -> str:
     """Return reStructured text rendering the provided metadata.
 
     @param meta: Dict of metadata to render.
+    @param schema: Optional module schema, if absent, read global schema.
     @raise KeyError: If metadata lacks an expected key.
     """
 
+    if schema is None:
+        schema = get_schema()
     if not meta or not schema:
-        raise ValueError("Expected meta and schema")
+        raise ValueError("Expected non-empty meta and schema")
     keys = set(meta.keys())
     expected = set(
         {
@@ -557,8 +585,11 @@ def get_meta_doc(meta: MetaSchema, schema: dict) -> str:
 
     # cast away type annotation
     meta_copy = dict(deepcopy(meta))
+    defs = schema.get("$defs", {})
+    if defs.get(meta["id"]):
+        schema = defs.get(meta["id"])
     try:
-        meta_copy["property_doc"] = _get_property_doc(schema)
+        meta_copy["property_doc"] = _get_property_doc(schema, defs=defs)
     except AttributeError:
         LOG.warning("Unable to render property_doc due to invalid schema")
         meta_copy["property_doc"] = ""
@@ -593,7 +624,7 @@ def load_doc(requested_modules: list) -> str:
     for mod_name in all_modules:
         if "all" in requested_modules or mod_name in requested_modules:
             (mod_locs, _) = importer.find_module(
-                mod_name, ["cloudinit.config"], ["schema"]
+                mod_name, ["cloudinit.config"], ["meta"]
             )
             if mod_locs:
                 mod = importer.import_module(mod_locs[0])
@@ -602,14 +633,34 @@ def load_doc(requested_modules: list) -> str:
 
 
 def get_schema() -> dict:
-    """Return jsonschema coalesced from all cc_* cloud-config module."""
-    full_schema = {
-        "$schema": "http://json-schema.org/draft-04/schema#",
-        "id": "cloud-config-schema",
-        "allOf": [],
-    }
+    """Return jsonschema coalesced from all cc_* cloud-config modules."""
+    schema_file = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)), "cloud-init-schema.json"
+    )
+    full_schema = None
+    try:
+        full_schema = json.loads(load_file(schema_file))
+    except Exception as e:
+        LOG.warning("Cannot parse JSON schema file %s. %s", schema_file, e)
+    if not full_schema:
+        LOG.warning(
+            "No base JSON schema files found at %s."
+            " Setting default empty schema",
+            schema_file,
+        )
+        full_schema = {
+            "$defs": {},
+            "$schema": "http://json-schema.org/draft-04/schema#",
+            "allOf": [],
+        }
 
+    # TODO( Drop the get_modules loop when all legacy cc_* schema migrates )
+    # Supplement base_schema with any legacy modules which still contain a
+    # "schema" attribute. Legacy cc_* modules will be migrated to use the
+    # store module schema in the composite cloud-init-schema-<version>.json
+    # and will drop "schema" at that point.
     for (_, mod_name) in get_modules().items():
+        # All cc_* modules need a "meta" attribute to represent schema defs
         (mod_locs, _) = importer.find_module(
             mod_name, ["cloudinit.config"], ["schema"]
         )
diff --git a/setup.py b/setup.py
index c98a4703..a9132d2c 100755
--- a/setup.py
+++ b/setup.py
@@ -332,6 +332,9 @@ setuptools.setup(
     author="Scott Moser",
     author_email="scott.moser@canonical.com",
     url="http://launchpad.net/cloud-init/",
+    package_data={
+        "": ["*.json"],
+    },
     packages=setuptools.find_packages(exclude=["tests.*", "tests"]),
     scripts=["tools/cloud-init-per"],
     license="Dual-licensed under GPLv3 or Apache 2.0",
diff --git a/tests/integration_tests/modules/test_cli.py b/tests/integration_tests/modules/test_cli.py
index 97bfe52d..81a5e7a8 100644
--- a/tests/integration_tests/modules/test_cli.py
+++ b/tests/integration_tests/modules/test_cli.py
@@ -13,11 +13,18 @@ runcmd:
   - echo 'hi' > /var/tmp/test
 """
 
-INVALID_USER_DATA = """\
+INVALID_USER_DATA_HEADER = """\
 runcmd:
   - echo 'hi' > /var/tmp/test
 """
 
+INVALID_USER_DATA_SCHEMA = """\
+#cloud-config
+updates:
+ notnetwork: -1
+apt_pipelining: bogus
+"""
+
 
 @pytest.mark.sru_2020_11
 @pytest.mark.user_data(VALID_USER_DATA)
@@ -29,10 +36,15 @@ def test_valid_userdata(client: IntegrationInstance):
     result = client.execute("cloud-init devel schema --system")
     assert result.ok
     assert "Valid cloud-config: system userdata" == result.stdout.strip()
+    result = client.execute("cloud-init status --long")
+    if not result.ok:
+        raise AssertionError(
+            f"Unexpected error from cloud-init status: {result}"
+        )
 
 
 @pytest.mark.sru_2020_11
-@pytest.mark.user_data(INVALID_USER_DATA)
+@pytest.mark.user_data(INVALID_USER_DATA_HEADER)
 def test_invalid_userdata(client: IntegrationInstance):
     """Test `cloud-init devel schema` with invalid userdata.
 
@@ -42,3 +54,30 @@ def test_invalid_userdata(client: IntegrationInstance):
     assert not result.ok
     assert "Cloud config schema errors" in result.stderr
     assert 'needs to begin with "#cloud-config"' in result.stderr
+    result = client.execute("cloud-init status --long")
+    if not result.ok:
+        raise AssertionError(
+            f"Unexpected error from cloud-init status: {result}"
+        )
+
+
+@pytest.mark.user_data(INVALID_USER_DATA_SCHEMA)
+def test_invalid_userdata_schema(client: IntegrationInstance):
+    """Test invalid schema represented as Warnings, not fatal
+
+    PR #1175
+    """
+    result = client.execute("cloud-init status --long")
+    assert result.ok
+    log = client.read_from_file("/var/log/cloud-init.log")
+    warning = (
+        "[WARNING]: Invalid cloud-config provided:\napt_pipelining: 'bogus'"
+        " is not valid under any of the given schemas\nupdates: Additional"
+        " properties are not allowed ('notnetwork' was unexpected)"
+    )
+    assert warning in log
+    result = client.execute("cloud-init status --long")
+    if not result.ok:
+        raise AssertionError(
+            f"Unexpected error from cloud-init status: {result}"
+        )
diff --git a/tests/unittests/config/test_cc_apk_configure.py b/tests/unittests/config/test_cc_apk_configure.py
index 6fbc3dec..85dd028f 100644
--- a/tests/unittests/config/test_cc_apk_configure.py
+++ b/tests/unittests/config/test_cc_apk_configure.py
@@ -6,11 +6,23 @@ Test creation of repositories file
 
 import logging
 import os
+import re
 import textwrap
 
+import pytest
+
 from cloudinit import cloud, helpers, util
 from cloudinit.config import cc_apk_configure
-from tests.unittests.helpers import FilesystemMockingTestCase, mock
+from cloudinit.config.schema import (
+    SchemaValidationError,
+    get_schema,
+    validate_cloudconfig_schema,
+)
+from tests.unittests.helpers import (
+    FilesystemMockingTestCase,
+    mock,
+    skipUnlessJsonSchema,
+)
 
 REPO_FILE = "/etc/apk/repositories"
 DEFAULT_MIRROR_URL = "https://alpine.global.ssl.fastly.net/alpine"
@@ -322,4 +334,77 @@ class TestConfig(FilesystemMockingTestCase):
         self.assertEqual(expected_content, util.load_file(REPO_FILE))
 
 
+class TestApkConfigureSchema:
+    @pytest.mark.parametrize(
+        "config, error_msg",
+        (
+            # Valid schemas
+            ({"apk_repos": {"preserve_repositories": True}}, None),
+            ({"apk_repos": {"alpine_repo": None}}, None),
+            ({"apk_repos": {"alpine_repo": {"version": "v3.21"}}}, None),
+            (
+                {
+                    "apk_repos": {
+                        "alpine_repo": {
+                            "base_url": "http://yep",
+                            "community_enabled": True,
+                            "testing_enabled": True,
+                            "version": "v3.21",
+                        }
+                    }
+                },
+                None,
+            ),
+            ({"apk_repos": {"local_repo_base_url": "http://some"}}, None),
+            # Invalid schemas
+            (
+                {"apk_repos": {"alpine_repo": {"version": False}}},
+                "apk_repos.alpine_repo.version: False is not of type"
+                " 'string'",
+            ),
+            (
+                {
+                    "apk_repos": {
+                        "alpine_repo": {"version": "v3.12", "bogus": 1}
+                    }
+                },
+                re.escape(
+                    "apk_repos.alpine_repo: Additional properties are not"
+                    " allowed ('bogus' was unexpected)"
+                ),
+            ),
+            (
+                {"apk_repos": {"alpine_repo": {}}},
+                "apk_repos.alpine_repo: 'version' is a required property,"
+                " apk_repos.alpine_repo: {} does not have enough properties",
+            ),
+            (
+                {"apk_repos": {"alpine_repo": True}},
+                "apk_repos.alpine_repo: True is not of type 'object', 'null'",
+            ),
+            (
+                {"apk_repos": {"preserve_repositories": "wrongtype"}},
+                "apk_repos.preserve_repositories: 'wrongtype' is not of type"
+                " 'boolean'",
+            ),
+            (
+                {"apk_repos": {}},
+                "apk_repos: {} does not have enough properties",
+            ),
+            (
+                {"apk_repos": {"local_repo_base_url": None}},
+                "apk_repos.local_repo_base_url: None is not of type 'string'",
+            ),
+        ),
+    )
+    @skipUnlessJsonSchema()
+    def test_schema_validation(self, config, error_msg):
+        schema = get_schema()
+        if error_msg is None:
+            validate_cloudconfig_schema(config, schema, strict=True)
+        else:
+            with pytest.raises(SchemaValidationError, match=error_msg):
+                validate_cloudconfig_schema(config, schema, strict=True)
+
+
 # vi: ts=4 expandtab
diff --git a/tests/unittests/config/test_cc_apt_pipelining.py b/tests/unittests/config/test_cc_apt_pipelining.py
index b4497156..0f72d32b 100644
--- a/tests/unittests/config/test_cc_apt_pipelining.py
+++ b/tests/unittests/config/test_cc_apt_pipelining.py
@@ -2,16 +2,23 @@
 
 """Tests cc_apt_pipelining handler"""
 
+import pytest
+
 import cloudinit.config.cc_apt_pipelining as cc_apt_pipelining
-from tests.unittests.helpers import CiTestCase, mock
+from cloudinit.config.schema import (
+    SchemaValidationError,
+    get_schema,
+    validate_cloudconfig_schema,
+)
+from tests.unittests.helpers import mock, skipUnlessJsonSchema
 
 
-class TestAptPipelining(CiTestCase):
+class TestAptPipelining:
     @mock.patch("cloudinit.config.cc_apt_pipelining.util.write_file")
     def test_not_disabled_by_default(self, m_write_file):
         """ensure that default behaviour is to not disable pipelining"""
         cc_apt_pipelining.handle("foo", {}, None, mock.MagicMock(), None)
-        self.assertEqual(0, m_write_file.call_count)
+        assert 0 == m_write_file.call_count
 
     @mock.patch("cloudinit.config.cc_apt_pipelining.util.write_file")
     def test_false_disables_pipelining(self, m_write_file):
@@ -19,10 +26,40 @@ class TestAptPipelining(CiTestCase):
         cc_apt_pipelining.handle(
             "foo", {"apt_pipelining": "false"}, None, mock.MagicMock(), None
         )
-        self.assertEqual(1, m_write_file.call_count)
+        assert 1 == m_write_file.call_count
         args, _ = m_write_file.call_args
-        self.assertEqual(cc_apt_pipelining.DEFAULT_FILE, args[0])
-        self.assertIn('Pipeline-Depth "0"', args[1])
+        assert cc_apt_pipelining.DEFAULT_FILE == args[0]
+        assert 'Pipeline-Depth "0"' in args[1]
+
+    @pytest.mark.parametrize(
+        "config, error_msg",
+        (
+            # Valid schemas
+            ({}, None),
+            ({"apt_pipelining": 1}, None),
+            ({"apt_pipelining": True}, None),
+            ({"apt_pipelining": False}, None),
+            ({"apt_pipelining": "none"}, None),
+            ({"apt_pipelining": "unchanged"}, None),
+            ({"apt_pipelining": "os"}, None),
+            # Invalid schemas
+            (
+                {"apt_pipelining": "bogus"},
+                "Cloud config schema errors: apt_pipelining: 'bogus' is not"
+                " valid under any of the given schema",
+            ),
+        ),
+    )
+    @skipUnlessJsonSchema()
+    def test_schema_validation(self, config, error_msg):
+        """Assert expected schema validation and error messages."""
+        # New-style schema $defs exist in config/cloud-init-schema*.json
+        schema = get_schema()
+        if error_msg is None:
+            validate_cloudconfig_schema(config, schema, strict=True)
+        else:
+            with pytest.raises(SchemaValidationError, match=error_msg):
+                validate_cloudconfig_schema(config, schema, strict=True)
 
 
 # vi: ts=4 expandtab
diff --git a/tests/unittests/config/test_cc_bootcmd.py b/tests/unittests/config/test_cc_bootcmd.py
index 6d8793b9..17033596 100644
--- a/tests/unittests/config/test_cc_bootcmd.py
+++ b/tests/unittests/config/test_cc_bootcmd.py
@@ -77,7 +77,8 @@ class TestBootcmd(CiTestCase):
         with self.assertRaises(TypeError):
             handle("cc_bootcmd", invalid_config, cc, LOG, [])
         self.assertIn(
-            "Invalid config:\nbootcmd: 1 is not of type 'array'",
+            "Invalid cloud-config provided:\nbootcmd: 1 is not of type"
+            " 'array'",
             self.logs.getvalue(),
         )
         self.assertIn("Failed to shellify", self.logs.getvalue())
diff --git a/tests/unittests/config/test_cc_ntp.py b/tests/unittests/config/test_cc_ntp.py
index 7da82cee..fba141aa 100644
--- a/tests/unittests/config/test_cc_ntp.py
+++ b/tests/unittests/config/test_cc_ntp.py
@@ -385,7 +385,9 @@ class TestNtp(FilesystemMockingTestCase):
                         "servers []\npools {0}\n".format(pools),
                         util.load_file(confpath),
                     )
-            self.assertNotIn("Invalid config:", self.logs.getvalue())
+            self.assertNotIn(
+                "Invalid cloud-config provided:", self.logs.getvalue()
+            )
 
     @skipUnlessJsonSchema()
     @mock.patch("cloudinit.config.cc_ntp.select_ntp_client")
@@ -404,8 +406,8 @@ class TestNtp(FilesystemMockingTestCase):
             m_sel.return_value = ntpconfig
             cc_ntp.handle("cc_ntp", invalid_config, mycloud, None, [])
             self.assertIn(
-                "Invalid config:\nntp.pools.0: 123 is not of type 'string'\n"
-                "ntp.servers.1: None is not of type 'string'",
+                "Invalid cloud-config provided:\nntp.pools.0: 123 is not of"
+                " type 'string'\nntp.servers.1: None is not of type 'string'",
                 self.logs.getvalue(),
             )
             self.assertEqual(
@@ -431,8 +433,8 @@ class TestNtp(FilesystemMockingTestCase):
             m_select.return_value = ntpconfig
             cc_ntp.handle("cc_ntp", invalid_config, mycloud, None, [])
             self.assertIn(
-                "Invalid config:\nntp.pools: 123 is not of type 'array'\n"
-                "ntp.servers: 'non-array' is not of type 'array'",
+                "Invalid cloud-config provided:\nntp.pools: 123 is not of type"
+                " 'array'\nntp.servers: 'non-array' is not of type 'array'",
                 self.logs.getvalue(),
             )
             self.assertEqual(
@@ -459,8 +461,9 @@ class TestNtp(FilesystemMockingTestCase):
                 m_select.return_value = ntpconfig
                 cc_ntp.handle("cc_ntp", invalid_config, mycloud, None, [])
                 self.assertIn(
-                    "Invalid config:\nntp: Additional properties are not "
-                    "allowed ('invalidkey' was unexpected)",
+                    "Invalid cloud-config provided:\nntp: Additional"
+                    " properties are not allowed ('invalidkey' was"
+                    " unexpected)",
                     self.logs.getvalue(),
                 )
                 self.assertEqual(
@@ -488,8 +491,8 @@ class TestNtp(FilesystemMockingTestCase):
             m_select.return_value = ntpconfig
             cc_ntp.handle("cc_ntp", invalid_config, mycloud, None, [])
             self.assertIn(
-                "Invalid config:\nntp.pools: ['0.mypool.org', '0.mypool.org']"
-                " has non-unique elements\nntp.servers: "
+                "Invalid cloud-config provided:\nntp.pools: ['0.mypool.org',"
+                " '0.mypool.org'] has non-unique elements\nntp.servers: "
                 "['10.0.0.1', '10.0.0.1'] has non-unique elements",
                 self.logs.getvalue(),
             )
diff --git a/tests/unittests/config/test_cc_resizefs.py b/tests/unittests/config/test_cc_resizefs.py
index 228f1e45..9981dcea 100644
--- a/tests/unittests/config/test_cc_resizefs.py
+++ b/tests/unittests/config/test_cc_resizefs.py
@@ -92,8 +92,8 @@ class TestResizefs(CiTestCase):
         handle("cc_resizefs", cfg, _cloud=None, log=LOG, args=[])
         logs = self.logs.getvalue()
         self.assertIn(
-            "WARNING: Invalid config:\nresize_rootfs: 'junk' is not one of"
-            " [True, False, 'noblock']",
+            "WARNING: Invalid cloud-config provided:\nresize_rootfs: 'junk' is"
+            " not one of [True, False, 'noblock']",
             logs,
         )
         self.assertIn(
@@ -108,7 +108,9 @@ class TestResizefs(CiTestCase):
         cfg = {"resize_rootfs": True}
         handle("cc_resizefs", cfg, _cloud=None, log=LOG, args=[])
         logs = self.logs.getvalue()
-        self.assertNotIn("WARNING: Invalid config:\nresize_rootfs:", logs)
+        self.assertNotIn(
+            "WARNING: Invalid cloud-config provided:\nresize_rootfs:", logs
+        )
         self.assertIn(
             "WARNING: Could not determine filesystem type of /\n", logs
         )
diff --git a/tests/unittests/config/test_cc_runcmd.py b/tests/unittests/config/test_cc_runcmd.py
index 34b3fb77..59490d67 100644
--- a/tests/unittests/config/test_cc_runcmd.py
+++ b/tests/unittests/config/test_cc_runcmd.py
@@ -73,7 +73,7 @@ class TestRuncmd(FilesystemMockingTestCase):
         with self.assertRaises(TypeError) as cm:
             handle("cc_runcmd", invalid_config, cc, LOG, [])
         self.assertIn(
-            "Invalid config:\nruncmd: 1 is not of type 'array'",
+            "Invalid cloud-config provided:\nruncmd: 1 is not of type 'array'",
             self.logs.getvalue(),
         )
         self.assertIn("Failed to shellify", str(cm.exception))
diff --git a/tests/unittests/config/test_cc_snap.py b/tests/unittests/config/test_cc_snap.py
index f7e66ad2..1632676d 100644
--- a/tests/unittests/config/test_cc_snap.py
+++ b/tests/unittests/config/test_cc_snap.py
@@ -284,8 +284,8 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         """If the snap configuration is not a dict, emit a warning."""
         validate_cloudconfig_schema({"snap": "wrong type"}, schema)
         self.assertEqual(
-            "WARNING: Invalid config:\nsnap: 'wrong type' is not of type"
-            " 'object'\n",
+            "WARNING: Invalid cloud-config provided:\nsnap: 'wrong type'"
+            " is not of type 'object'\n",
             self.logs.getvalue(),
         )
 
@@ -296,8 +296,8 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             {"snap": {"commands": ["ls"], "invalid-key": ""}}, schema
         )
         self.assertIn(
-            "WARNING: Invalid config:\nsnap: Additional properties are not"
-            " allowed ('invalid-key' was unexpected)",
+            "WARNING: Invalid cloud-config provided:\nsnap: Additional"
+            " properties are not allowed ('invalid-key' was unexpected)",
             self.logs.getvalue(),
         )
 
@@ -305,8 +305,8 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         """Warn when snap configuration lacks both commands and assertions."""
         validate_cloudconfig_schema({"snap": {}}, schema)
         self.assertIn(
-            "WARNING: Invalid config:\nsnap: {} does not have enough"
-            " properties",
+            "WARNING: Invalid cloud-config provided:\nsnap: {} does not"
+            " have enough properties",
             self.logs.getvalue(),
         )
 
@@ -315,8 +315,8 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         """Warn when snap:commands config is not a list or dict."""
         validate_cloudconfig_schema({"snap": {"commands": "broken"}}, schema)
         self.assertEqual(
-            "WARNING: Invalid config:\nsnap.commands: 'broken' is not of type"
-            " 'object', 'array'\n",
+            "WARNING: Invalid cloud-config provided:\nsnap.commands: 'broken'"
+            " is not of type 'object', 'array'\n",
             self.logs.getvalue(),
         )
 
@@ -326,9 +326,9 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         validate_cloudconfig_schema({"snap": {"commands": []}}, schema)
         validate_cloudconfig_schema({"snap": {"commands": {}}}, schema)
         self.assertEqual(
-            "WARNING: Invalid config:\nsnap.commands: [] is too short\n"
-            "WARNING: Invalid config:\nsnap.commands: {} does not have enough"
-            " properties\n",
+            "WARNING: Invalid cloud-config provided:\nsnap.commands: [] is"
+            " too short\nWARNING: Invalid cloud-config provided:\n"
+            "snap.commands: {} does not have enough properties\n",
             self.logs.getvalue(),
         )
 
@@ -349,10 +349,10 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             {"snap": {"commands": {"01": 123}}}, schema
         )
         self.assertEqual(
-            "WARNING: Invalid config:\n"
+            "WARNING: Invalid cloud-config provided:\n"
             "snap.commands.0: 123 is not valid under any of the given"
             " schemas\n"
-            "WARNING: Invalid config:\n"
+            "WARNING: Invalid cloud-config provided:\n"
             "snap.commands.01: 123 is not valid under any of the given"
             " schemas\n",
             self.logs.getvalue(),
@@ -368,10 +368,10 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             {"snap": {"commands": {"01": ["snap", "install", 123]}}}, schema
         )
         self.assertEqual(
-            "WARNING: Invalid config:\n"
+            "WARNING: Invalid cloud-config provided:\n"
             "snap.commands.0: ['snap', 'install', 123] is not valid under any"
             " of the given schemas\n",
-            "WARNING: Invalid config:\n"
+            "WARNING: Invalid cloud-config provided:\n"
             "snap.commands.0: ['snap', 'install', 123] is not valid under any"
             " of the given schemas\n",
             self.logs.getvalue(),
@@ -385,9 +385,9 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             {"snap": {"assertions": {"01": 123}}}, schema
         )
         self.assertEqual(
-            "WARNING: Invalid config:\n"
+            "WARNING: Invalid cloud-config provided:\n"
             "snap.assertions.0: 123 is not of type 'string'\n"
-            "WARNING: Invalid config:\n"
+            "WARNING: Invalid cloud-config provided:\n"
             "snap.assertions.01: 123 is not of type 'string'\n",
             self.logs.getvalue(),
         )
@@ -397,8 +397,8 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         """Warn when snap:assertions config is not a list or dict."""
         validate_cloudconfig_schema({"snap": {"assertions": "broken"}}, schema)
         self.assertEqual(
-            "WARNING: Invalid config:\nsnap.assertions: 'broken' is not of"
-            " type 'object', 'array'\n",
+            "WARNING: Invalid cloud-config provided:\nsnap.assertions:"
+            " 'broken' is not of type 'object', 'array'\n",
             self.logs.getvalue(),
         )
 
@@ -408,9 +408,10 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         validate_cloudconfig_schema({"snap": {"assertions": []}}, schema)
         validate_cloudconfig_schema({"snap": {"assertions": {}}}, schema)
         self.assertEqual(
-            "WARNING: Invalid config:\nsnap.assertions: [] is too short\n"
-            "WARNING: Invalid config:\nsnap.assertions: {} does not have"
-            " enough properties\n",
+            "WARNING: Invalid cloud-config provided:\nsnap.assertions: []"
+            " is too short\n"
+            "WARNING: Invalid cloud-config provided:\nsnap.assertions: {}"
+            " does not have enough properties\n",
             self.logs.getvalue(),
         )
 
@@ -574,8 +575,8 @@ class TestHandle(CiTestCase):
             args=None,
         )
         self.assertEqual(
-            "WARNING: Invalid config:\nsnap: Additional properties are not"
-            " allowed ('invalid' was unexpected)\n",
+            "WARNING: Invalid cloud-config provided:\nsnap: Additional"
+            " properties are not allowed ('invalid' was unexpected)\n",
             self.logs.getvalue(),
         )
 
diff --git a/tests/unittests/config/test_cc_ubuntu_advantage.py b/tests/unittests/config/test_cc_ubuntu_advantage.py
index c39e421f..2037c5ed 100644
--- a/tests/unittests/config/test_cc_ubuntu_advantage.py
+++ b/tests/unittests/config/test_cc_ubuntu_advantage.py
@@ -184,8 +184,8 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
         """If ubuntu_advantage configuration is not a dict, emit a warning."""
         validate_cloudconfig_schema({"ubuntu_advantage": "wrong type"}, schema)
         self.assertEqual(
-            "WARNING: Invalid config:\nubuntu_advantage: 'wrong type' is not"
-            " of type 'object'\n",
+            "WARNING: Invalid cloud-config provided:\nubuntu_advantage:"
+            " 'wrong type' is not of type 'object'\n",
             self.logs.getvalue(),
         )
 
@@ -198,8 +198,9 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             schema,
         )
         self.assertIn(
-            "WARNING: Invalid config:\nubuntu_advantage: Additional properties"
-            " are not allowed ('invalid-key' was unexpected)",
+            "WARNING: Invalid cloud-config provided:\nubuntu_advantage:"
+            " Additional properties are not allowed ('invalid-key' was"
+            " unexpected)",
             self.logs.getvalue(),
         )
 
@@ -211,7 +212,7 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             {"ubuntu_advantage": {"enable": ["esm"]}}, schema
         )
         self.assertEqual(
-            "WARNING: Invalid config:\nubuntu_advantage:"
+            "WARNING: Invalid cloud-config provided:\nubuntu_advantage:"
             " 'token' is a required property\n",
             self.logs.getvalue(),
         )
@@ -224,9 +225,9 @@ class TestSchema(CiTestCase, SchemaTestCaseMixin):
             {"ubuntu_advantage": {"enable": "needslist"}}, schema
         )
         self.assertEqual(
-            "WARNING: Invalid config:\nubuntu_advantage: 'token' is a"
-            " required property\nubuntu_advantage.enable: 'needslist'"
-            " is not of type 'array'\n",
+            "WARNING: Invalid cloud-config provided:\nubuntu_advantage:"
+            " 'token' is a required property\nubuntu_advantage.enable:"
+            " 'needslist' is not of type 'array'\n",
             self.logs.getvalue(),
         )
 
diff --git a/tests/unittests/config/test_cc_write_files.py b/tests/unittests/config/test_cc_write_files.py
index 7eea99d3..faea5885 100644
--- a/tests/unittests/config/test_cc_write_files.py
+++ b/tests/unittests/config/test_cc_write_files.py
@@ -79,9 +79,11 @@ class TestWriteFilesSchema(CiTestCase):
         cc = self.tmp_cloud("ubuntu")
         valid_config = {"write_files": [{"path": "/some/path"}]}
         handle("cc_write_file", valid_config, cc, LOG, [])
-        self.assertNotIn("Invalid config:", self.logs.getvalue())
+        self.assertNotIn(
+            "Invalid cloud-config provided:", self.logs.getvalue()
+        )
         handle("cc_write_file", INVALID_SCHEMA, cc, LOG, [])
-        self.assertIn("Invalid config:", self.logs.getvalue())
+        self.assertIn("Invalid cloud-config provided:", self.logs.getvalue())
         self.assertIn("'path' is a required property", self.logs.getvalue())
 
     def test_schema_validation_warns_non_string_type_for_files(
@@ -105,7 +107,7 @@ class TestWriteFilesSchema(CiTestCase):
                 "write_files.0.%s: 1 is not of type '%s'" % (key, key_type),
                 self.logs.getvalue(),
             )
-        self.assertIn("Invalid config:", self.logs.getvalue())
+        self.assertIn("Invalid cloud-config provided:", self.logs.getvalue())
 
     def test_schema_validation_warns_on_additional_undefined_propertes(
         self, m_write_files
@@ -116,8 +118,8 @@ class TestWriteFilesSchema(CiTestCase):
         invalid_config["write_files"][0]["bogus"] = "value"
         handle("cc_write_file", invalid_config, cc, LOG, [])
         self.assertIn(
-            "Invalid config:\nwrite_files.0: Additional properties"
-            " are not allowed ('bogus' was unexpected)",
+            "Invalid cloud-config provided:\nwrite_files.0: Additional"
+            " properties are not allowed ('bogus' was unexpected)",
             self.logs.getvalue(),
         )
 
@@ -139,7 +141,8 @@ class TestWriteFiles(FilesystemMockingTestCase):
         with self.assertRaises(TypeError):
             handle("cc_write_file", invalid_config, cc, LOG, [])
         self.assertIn(
-            "Invalid config:\nwrite_files: 1 is not of type 'array'",
+            "Invalid cloud-config provided:\nwrite_files: 1 is not of type"
+            " 'array'",
             self.logs.getvalue(),
         )
 
diff --git a/tests/unittests/config/test_cc_write_files_deferred.py b/tests/unittests/config/test_cc_write_files_deferred.py
index 3faac1bf..17203233 100644
--- a/tests/unittests/config/test_cc_write_files_deferred.py
+++ b/tests/unittests/config/test_cc_write_files_deferred.py
@@ -43,9 +43,11 @@ class TestWriteFilesDeferredSchema(CiTestCase):
 
         cc = self.tmp_cloud("ubuntu")
         handle("cc_write_files_deferred", valid_config, cc, LOG, [])
-        self.assertNotIn("Invalid config:", self.logs.getvalue())
+        self.assertNotIn(
+            "Invalid cloud-config provided:", self.logs.getvalue()
+        )
         handle("cc_write_files_deferred", invalid_config, cc, LOG, [])
-        self.assertIn("Invalid config:", self.logs.getvalue())
+        self.assertIn("Invalid cloud-config provided:", self.logs.getvalue())
         self.assertIn(
             "defer: 'no' is not of type 'boolean'", self.logs.getvalue()
         )
diff --git a/tests/unittests/config/test_schema.py b/tests/unittests/config/test_schema.py
index 93206bdd..5cb00c5d 100644
--- a/tests/unittests/config/test_schema.py
+++ b/tests/unittests/config/test_schema.py
@@ -11,16 +11,19 @@ from pathlib import Path
 from textwrap import dedent
 
 import pytest
+import yaml
 from yaml import safe_load
 
 from cloudinit.config.schema import (
     CLOUD_CONFIG_HEADER,
     MetaSchema,
     SchemaValidationError,
+    _schemapath_for_cloudconfig,
     annotated_cloudconfig_file,
     get_jsonschema_validator,
     get_meta_doc,
     get_schema,
+    load_doc,
     main,
     validate_cloudconfig_file,
     validate_cloudconfig_metaschema,
@@ -36,7 +39,7 @@ from tests.unittests.helpers import (
 
 
 def get_schemas() -> dict:
-    """Return all module schemas
+    """Return all legacy module schemas
 
     Assumes that module schemas have the variable name "schema"
     """
@@ -78,14 +81,15 @@ def get_module_variable(var_name) -> dict:
     return schemas
 
 
-class GetSchemaTest(CiTestCase):
+class TestGetSchema:
     def test_get_schema_coalesces_known_schema(self):
         """Every cloudconfig module with schema is listed in allOf keyword."""
         schema = get_schema()
-        self.assertCountEqual(
+        assert sorted(
             [
                 "cc_apk_configure",
                 "cc_apt_configure",
+                "cc_apt_pipelining",
                 "cc_bootcmd",
                 "cc_keyboard",
                 "cc_locale",
@@ -99,14 +103,88 @@ class GetSchemaTest(CiTestCase):
                 "cc_zypper_add_repo",
                 "cc_chef",
                 "cc_install_hotplug",
-            ],
-            [meta["id"] for meta in get_metas().values() if meta is not None],
-        )
-        self.assertEqual("cloud-config-schema", schema["id"])
-        self.assertEqual(
-            "http://json-schema.org/draft-04/schema#", schema["$schema"]
+            ]
+        ) == sorted(
+            [meta["id"] for meta in get_metas().values() if meta is not None]
+        )
+        assert "http://json-schema.org/draft-04/schema#" == schema["$schema"]
+        assert ["$defs", "$schema", "allOf"] == sorted(list(schema.keys()))
+        # New style schema should be defined in static schema file in $defs
+        expected_subschema_defs = [
+            {"$ref": "#/$defs/cc_apk_configure"},
+            {"$ref": "#/$defs/cc_apt_pipelining"},
+        ]
+        found_subschema_defs = []
+        legacy_schema_keys = []
+        for subschema in schema["allOf"]:
+            if "$ref" in subschema:
+                found_subschema_defs.append(subschema)
+            else:  # Legacy subschema sourced from cc_* module 'schema' attr
+                legacy_schema_keys.extend(subschema["properties"].keys())
+
+        assert expected_subschema_defs == found_subschema_defs
+        # This list will dwindle as we move legacy schema to new $defs
+        assert [
+            "apt",
+            "bootcmd",
+            "chef",
+            "drivers",
+            "keyboard",
+            "locale",
+            "locale_configfile",
+            "ntp",
+            "resize_rootfs",
+            "runcmd",
+            "snap",
+            "ubuntu_advantage",
+            "updates",
+            "write_files",
+            "write_files",
+            "zypper",
+        ] == sorted(legacy_schema_keys)
+
+
+class TestLoadDoc:
+
+    docs = get_module_variable("__doc__")
+
+    # TODO( Drop legacy test when all sub-schemas in cloud-init-schema.json )
+    @pytest.mark.parametrize(
+        "module_name",
+        (
+            "cc_apt_pipelining",  # new style composite schema file
+            "cc_bootcmd",  # legacy sub-schema defined in module
+        ),
+    )
+    def test_report_docs_for_legacy_and_consolidated_schema(self, module_name):
+        doc = load_doc([module_name])
+        assert doc, "Unexpected empty docs for {}".format(module_name)
+        assert self.docs[module_name] == doc
+
+
+class Test_SchemapathForCloudconfig:
+    """Coverage tests for supported YAML formats."""
+
+    @pytest.mark.parametrize(
+        "source_content, expected",
+        (
+            (b"{}", {}),  # assert empty config handled
+            # Multiple keys account for comments and whitespace lines
+            (b"#\na: va\n  \nb: vb\n#\nc: vc", {"a": 2, "b": 4, "c": 6}),
+            # List items represented on correct line number
+            (b"a:\n - a1\n\n - a2\n", {"a": 1, "a.0": 2, "a.1": 4}),
+            # Nested dicts represented on correct line number
+            (b"a:\n a1:\n\n  aa1: aa1v\n", {"a": 1, "a.a1": 2, "a.a1.aa1": 4}),
+        ),
+    )
+    def test_schemapaths_representatative_of_source_yaml(
+        self, source_content, expected
+    ):
+        """Validate schemapaths dict accurately represents source YAML line."""
+        cfg = yaml.safe_load(source_content)
+        assert expected == _schemapath_for_cloudconfig(
+            config=cfg, original_content=source_content
         )
-        self.assertCountEqual(["id", "$schema", "allOf"], get_schema().keys())
 
 
 class SchemaValidationErrorTest(CiTestCase):
@@ -129,69 +207,87 @@ class SchemaValidationErrorTest(CiTestCase):
         self.assertTrue(isinstance(exception, ValueError))
 
 
-class ValidateCloudConfigSchemaTest(CiTestCase):
+class TestValidateCloudConfigSchema:
     """Tests for validate_cloudconfig_schema."""
 
     with_logs = True
 
+    @pytest.mark.parametrize(
+        "schema, call_count",
+        ((None, 1), ({"properties": {"p1": {"type": "string"}}}, 0)),
+    )
     @skipUnlessJsonSchema()
-    def test_validateconfig_schema_non_strict_emits_warnings(self):
+    @mock.patch("cloudinit.config.schema.get_schema")
+    def test_validateconfig_schema_use_full_schema_when_no_schema_param(
+        self, get_schema, schema, call_count
+    ):
+        """Use full schema when schema param is absent."""
+        get_schema.return_value = {"properties": {"p1": {"type": "string"}}}
+        kwargs = {"config": {"p1": "valid"}}
+        if schema:
+            kwargs["schema"] = schema
+        validate_cloudconfig_schema(**kwargs)
+        assert call_count == get_schema.call_count
+
+    @skipUnlessJsonSchema()
+    def test_validateconfig_schema_non_strict_emits_warnings(self, caplog):
         """When strict is False validate_cloudconfig_schema emits warnings."""
         schema = {"properties": {"p1": {"type": "string"}}}
         validate_cloudconfig_schema({"p1": -1}, schema, strict=False)
-        self.assertIn(
-            "Invalid config:\np1: -1 is not of type 'string'\n",
-            self.logs.getvalue(),
+        assert (
+            "Invalid cloud-config provided:\np1: -1 is not of type 'string'\n"
+            in (caplog.text)
         )
 
     @skipUnlessJsonSchema()
-    def test_validateconfig_schema_emits_warning_on_missing_jsonschema(self):
+    def test_validateconfig_schema_emits_warning_on_missing_jsonschema(
+        self, caplog
+    ):
         """Warning from validate_cloudconfig_schema when missing jsonschema."""
         schema = {"properties": {"p1": {"type": "string"}}}
         with mock.patch.dict("sys.modules", **{"jsonschema": ImportError()}):
             validate_cloudconfig_schema({"p1": -1}, schema, strict=True)
-        self.assertIn(
-            "Ignoring schema validation. jsonschema is not present",
-            self.logs.getvalue(),
+        assert "Ignoring schema validation. jsonschema is not present" in (
+            caplog.text
         )
 
     @skipUnlessJsonSchema()
     def test_validateconfig_schema_strict_raises_errors(self):
         """When strict is True validate_cloudconfig_schema raises errors."""
         schema = {"properties": {"p1": {"type": "string"}}}
-        with self.assertRaises(SchemaValidationError) as context_mgr:
+        with pytest.raises(SchemaValidationError) as context_mgr:
             validate_cloudconfig_schema({"p1": -1}, schema, strict=True)
-        self.assertEqual(
-            "Cloud config schema errors: p1: -1 is not of type 'string'",
-            str(context_mgr.exception),
+        assert (
+            "Cloud config schema errors: p1: -1 is not of type 'string'"
+            == (str(context_mgr.value))
         )
 
     @skipUnlessJsonSchema()
     def test_validateconfig_schema_honors_formats(self):
         """With strict True, validate_cloudconfig_schema errors on format."""
         schema = {"properties": {"p1": {"type": "string", "format": "email"}}}
-        with self.assertRaises(SchemaValidationError) as context_mgr:
+        with pytest.raises(SchemaValidationError) as context_mgr:
             validate_cloudconfig_schema({"p1": "-1"}, schema, strict=True)
-        self.assertEqual(
-            "Cloud config schema errors: p1: '-1' is not a 'email'",
-            str(context_mgr.exception),
+        assert "Cloud config schema errors: p1: '-1' is not a 'email'" == (
+            str(context_mgr.value)
         )
 
     @skipUnlessJsonSchema()
     def test_validateconfig_schema_honors_formats_strict_metaschema(self):
         """With strict and strict_metaschema True, ensure errors on format"""
         schema = {"properties": {"p1": {"type": "string", "format": "email"}}}
-        with self.assertRaises(SchemaValidationError) as context_mgr:
+        with pytest.raises(SchemaValidationError) as context_mgr:
             validate_cloudconfig_schema(
                 {"p1": "-1"}, schema, strict=True, strict_metaschema=True
             )
-        self.assertEqual(
-            "Cloud config schema errors: p1: '-1' is not a 'email'",
-            str(context_mgr.exception),
+        assert "Cloud config schema errors: p1: '-1' is not a 'email'" == str(
+            context_mgr.value
         )
 
     @skipUnlessJsonSchema()
-    def test_validateconfig_strict_metaschema_do_not_raise_exception(self):
+    def test_validateconfig_strict_metaschema_do_not_raise_exception(
+        self, caplog
+    ):
         """With strict_metaschema=True, do not raise exceptions.
 
         This flag is currently unused, but is intended for run-time validation.
@@ -203,12 +299,11 @@ class ValidateCloudConfigSchemaTest(CiTestCase):
         )
         assert (
             "Meta-schema validation failed, attempting to validate config"
-            in self.logs.getvalue()
+            in caplog.text
         )
 
 
 class TestCloudConfigExamples:
-    schema = get_schemas()
     metas = get_metas()
     params = [
         (meta["id"], example)
@@ -223,10 +318,9 @@ class TestCloudConfigExamples:
         """For a given example in a config module we test if it is valid
         according to the unified schema of all config modules
         """
+        schema = get_schema()
         config_load = safe_load(example)
-        validate_cloudconfig_schema(
-            config_load, self.schema[schema_id], strict=True
-        )
+        validate_cloudconfig_schema(config_load, schema, strict=True)
 
 
 class ValidateCloudConfigFileTest(CiTestCase):