[cli] Add config bundle CLI command for remote compilation (#13791)

This commit is contained in:
J. Nick Koston
2026-04-07 10:37:19 -10:00
committed by GitHub
parent c6c743e2bb
commit ef6c65c7ec
21 changed files with 2390 additions and 7 deletions
+61
View File
@@ -1242,6 +1242,38 @@ def command_clean(args: ArgsProtocol, config: ConfigType) -> int | None:
return 0
def command_bundle(args: ArgsProtocol, config: ConfigType) -> int | None:
from esphome.bundle import BUNDLE_EXTENSION, ConfigBundleCreator
creator = ConfigBundleCreator(config)
if args.list_only:
files = creator.discover_files()
for bf in sorted(files, key=lambda f: f.path):
safe_print(f" {bf.path}")
_LOGGER.info("Found %d files", len(files))
return 0
result = creator.create_bundle()
if args.output:
output_path = Path(args.output)
else:
stem = CORE.config_path.stem
output_path = CORE.config_dir / f"{stem}{BUNDLE_EXTENSION}"
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_bytes(result.data)
_LOGGER.info(
"Bundle created: %s (%d files, %.1f KB)",
output_path,
len(result.files),
len(result.data) / 1024,
)
return 0
def command_dashboard(args: ArgsProtocol) -> int | None:
from esphome.dashboard import dashboard
@@ -1517,6 +1549,7 @@ POST_CONFIG_ACTIONS = {
"rename": command_rename,
"discover": command_discover,
"analyze-memory": command_analyze_memory,
"bundle": command_bundle,
}
SIMPLE_CONFIG_ACTIONS = [
@@ -1818,6 +1851,24 @@ def parse_args(argv):
"configuration", help="Your YAML configuration file(s).", nargs="+"
)
parser_bundle = subparsers.add_parser(
"bundle",
help="Create a self-contained config bundle for remote compilation.",
)
parser_bundle.add_argument(
"configuration", help="Your YAML configuration file(s).", nargs="+"
)
parser_bundle.add_argument(
"-o",
"--output",
help="Output path for the bundle archive.",
)
parser_bundle.add_argument(
"--list-only",
help="List discovered files without creating the archive.",
action="store_true",
)
# Keep backward compatibility with the old command line format of
# esphome <config> <command>.
#
@@ -1896,6 +1947,16 @@ def run_esphome(argv):
_LOGGER.warning("Skipping secrets file %s", conf_path)
return 0
# Bundle support: if the configuration is a .esphomebundle, extract it
# and rewrite conf_path to the extracted YAML config.
from esphome.bundle import is_bundle_path, prepare_bundle_for_compile
if is_bundle_path(conf_path):
_LOGGER.info("Extracting config bundle %s...", conf_path)
conf_path = prepare_bundle_for_compile(conf_path)
# Update the argument so downstream code sees the extracted path
args.configuration[0] = str(conf_path)
CORE.config_path = conf_path
CORE.dashboard = args.dashboard
+699
View File
File diff suppressed because it is too large Load Diff
+4 -2
View File
@@ -71,9 +71,11 @@ def _validate_load_certificate(value):
def validate_certificate(value):
# _validate_load_certificate already calls cv.file_() internally,
# but returns the parsed certificate object. We re-call cv.file_()
# to get the resolved path string that the bundle walker can discover.
_validate_load_certificate(value)
# Validation result should be the path, not the loaded certificate
return value
return str(cv.file_(value))
def _validate_load_private_key(key, cert_pw):
+28 -5
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
from collections.abc import Callable
from contextlib import suppress
from collections.abc import Callable, Generator
from contextlib import contextmanager, suppress
import functools
import inspect
from io import BytesIO, TextIOBase, TextIOWrapper
@@ -44,6 +44,27 @@ _LOGGER = logging.getLogger(__name__)
SECRET_YAML = "secrets.yaml"
_SECRET_CACHE = {}
_SECRET_VALUES = {}
# Not thread-safe — config processing is single-threaded today.
_load_listeners: list[Callable[[Path], None]] = []
@contextmanager
def track_yaml_loads() -> Generator[list[Path]]:
"""Context manager that records every file loaded by the YAML loader.
Yields a list that is populated with resolved Path objects for every
file loaded through ``_load_yaml_internal`` while the context is active.
"""
loaded: list[Path] = []
def _on_load(fname: Path) -> None:
loaded.append(Path(fname).resolve())
_load_listeners.append(_on_load)
try:
yield loaded
finally:
_load_listeners.remove(_on_load)
class ESPHomeDataBase:
@@ -466,6 +487,8 @@ def load_yaml(fname: Path, clear_secrets: bool = True) -> Any:
def _load_yaml_internal(fname: Path) -> Any:
"""Load a YAML file."""
for listener in _load_listeners:
listener(fname)
try:
with fname.open(encoding="utf-8") as f_handle:
return parse_yaml(fname, f_handle)
@@ -473,10 +496,10 @@ def _load_yaml_internal(fname: Path) -> Any:
raise EsphomeError(f"Error reading file {fname}: {err}") from err
def parse_yaml(
file_name: Path, file_handle: TextIOWrapper, yaml_loader=_load_yaml_internal
) -> Any:
def parse_yaml(file_name: Path, file_handle: TextIOWrapper, yaml_loader=None) -> Any:
"""Parse a YAML file."""
if yaml_loader is None:
yaml_loader = _load_yaml_internal
try:
return _load_yaml_internal_with_type(
ESPHomeLoader, file_name, file_handle, yaml_loader
@@ -0,0 +1,18 @@
-----BEGIN CERTIFICATE-----
MIICzjCCAbagAwIBAgIUW3BzjtekVgMj12/oeXawSswGyXMwDQYJKoZIhvcNAQEL
BQAwITEfMB0GA1UEAwwWRVNQSG9tZSBCdW5kbGUgVGVzdCBDQTAeFw0yNjAyMDYx
MzMxMTZaFw0yNzAyMDYxMzMxMTZaMCExHzAdBgNVBAMMFkVTUEhvbWUgQnVuZGxl
IFRlc3QgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDG62vBFkGn
hEu54gh2A7b1ZwesVadZ6u0iaVO7GSWiI0o4nb6xv7ULZbGrgsKNIO6qCV4VSR3p
BfMhF5dFy8kkMzA8dKZMk16tygzocdNum2QQ8BHyIsATL7SGZ33si9Alp30gXv6h
XSlEKYDKHFavkDhWPFNa5+oeHbMS/MxjpOUXIpq32VaFpJr427d9Y9wGjuK8B7Gp
CI5Ub1g2dpC9xSHqQKD3JZokmtc70+mD74AcNWbyxWp0bkW9wOfNJJnAoiwhJxQ8
yfE37UsUIVc8014NhdhU1K/S0iQuOKfGX1L/GAshv8syQIcDfzJuJdX+5E/leAYD
UEKqRkcLT+D5AgMBAAEwDQYJKoZIhvcNAQELBQADggEBAF1HpJ6d+W5WrzOQrGej
41pxCDeJ9tSiSj/KtvJfjEVIpg0hMRTY7nSL7OAg9KGESfx4u1jMwVnyOv34br5B
DTlRl+wF2k7Ip8CNnyZfCC+1SVQZpUt1mVNz8BhIZZ9/a830wCILNQQrVKkSeNBk
SEc1qTt4mIhQZ+M422qAswluv4fz/FW1f4oB9KhCpzUCANjmyERnqTnImjnJu8h0
jbPNnNsN+G+Roju8UD/7atWYfAUmDjHx72Ci/5G9SzoM5fhgxxu43XYd5RW5wBzt
j4KdKdYlDtOL62mRPKWd40uGnJcieUjisU7noRn0ErMgbUlhLdbXT9X7aNborZcu
x6I=
-----END CERTIFICATE-----
@@ -0,0 +1,18 @@
-----BEGIN CERTIFICATE-----
MIICzjCCAbagAwIBAgIUW3BzjtekVgMj12/oeXawSswGyXMwDQYJKoZIhvcNAQEL
BQAwITEfMB0GA1UEAwwWRVNQSG9tZSBCdW5kbGUgVGVzdCBDQTAeFw0yNjAyMDYx
MzMxMTZaFw0yNzAyMDYxMzMxMTZaMCExHzAdBgNVBAMMFkVTUEhvbWUgQnVuZGxl
IFRlc3QgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDG62vBFkGn
hEu54gh2A7b1ZwesVadZ6u0iaVO7GSWiI0o4nb6xv7ULZbGrgsKNIO6qCV4VSR3p
BfMhF5dFy8kkMzA8dKZMk16tygzocdNum2QQ8BHyIsATL7SGZ33si9Alp30gXv6h
XSlEKYDKHFavkDhWPFNa5+oeHbMS/MxjpOUXIpq32VaFpJr427d9Y9wGjuK8B7Gp
CI5Ub1g2dpC9xSHqQKD3JZokmtc70+mD74AcNWbyxWp0bkW9wOfNJJnAoiwhJxQ8
yfE37UsUIVc8014NhdhU1K/S0iQuOKfGX1L/GAshv8syQIcDfzJuJdX+5E/leAYD
UEKqRkcLT+D5AgMBAAEwDQYJKoZIhvcNAQELBQADggEBAF1HpJ6d+W5WrzOQrGej
41pxCDeJ9tSiSj/KtvJfjEVIpg0hMRTY7nSL7OAg9KGESfx4u1jMwVnyOv34br5B
DTlRl+wF2k7Ip8CNnyZfCC+1SVQZpUt1mVNz8BhIZZ9/a830wCILNQQrVKkSeNBk
SEc1qTt4mIhQZ+M422qAswluv4fz/FW1f4oB9KhCpzUCANjmyERnqTnImjnJu8h0
jbPNnNsN+G+Roju8UD/7atWYfAUmDjHx72Ci/5G9SzoM5fhgxxu43XYd5RW5wBzt
j4KdKdYlDtOL62mRPKWd40uGnJcieUjisU7noRn0ErMgbUlhLdbXT9X7aNborZcu
x6I=
-----END CERTIFICATE-----
@@ -0,0 +1,27 @@
-----BEGIN RSA PRIVATE KEY-----
MIIEowIBAAKCAQEAxutrwRZBp4RLueIIdgO29WcHrFWnWertImlTuxkloiNKOJ2+
sb+1C2Wxq4LCjSDuqgleFUkd6QXzIReXRcvJJDMwPHSmTJNercoM6HHTbptkEPAR
8iLAEy+0hmd97IvQJad9IF7+oV0pRCmAyhxWr5A4VjxTWufqHh2zEvzMY6TlFyKa
t9lWhaSa+Nu3fWPcBo7ivAexqQiOVG9YNnaQvcUh6kCg9yWaJJrXO9Ppg++AHDVm
8sVqdG5FvcDnzSSZwKIsIScUPMnxN+1LFCFXPNNeDYXYVNSv0tIkLjinxl9S/xgL
Ib/LMkCHA38ybiXV/uRP5XgGA1BCqkZHC0/g+QIDAQABAoIBAEpsFwcJNCwf95MG
qcK5lhCPaRQFgdTG68ylmoGUIXvddy3ies+W2X33oLb5958ElLaCRbRyBCJEKxgU
8vBWk50bF69uty9MLa6YuyaWO5QUyCX8I8KzVKh4/zIP81F2Z7xGwy5CzEKED+Xk
Hz6+xoHt094TuN34iaOV2gM/GJsok4Wp/lzsuT3X6i3Nad9YGrV2yL/wv5c542bw
vrFDtYQ/+ADZZPW4+xK0ShiarSqV3iXB2cEjc4JX7yLX1hB4LY8VHRzl+Byjdl0/
lheiIesl5htl82SFxquZDimDsbilTm7TLW2bbm3b3/oC7DchTx6COBjp90VJqk3R
QrO5dicCgYEA80pyA7tCB0bGnJ7KWkteKddyOdakeYeM7Bpfv17qbCm9ciMw9nqt
KJVZPtAuqZGTpfSJseOCIyz9zloB79hVJ3mdWpGJVvmNM5H+BJyCciXpwfqp64QG
1gMqGlSy/MwsZHqNCsOIvrzH09GFN0LSPNKeXN7GNAtU1vI5s7Xf158CgYEA0U+Y
Qe1qJY4m597spHNFfkGznoFXAjHOoWYHv95902cH6JD4GnYPfwFXxgFsrJhFaFMC
jXlT0fRFAIe4NuUJhGD6TYSJqsFkH3xJkAepvKpfjM5qJ7+PQHRnED/E5OS2Nj0R
+cxBhTEWTw9YiOFBRbj6hlphkj8izVGJZ2pL4GcCgYEApsjiYKx/F33tqnExR7Vj
WEvagswi9S137mQmP4tSKdRzi0uUxWRUUP4RsH4HfzfNgHej7c+J55Nwa4ZIzaQA
vI8i0HP1MyrhIflzqrWgt6BGIDU3R7268fw5YNOv4J4X0Moy5q4lkJzaYNvB96BX
gFrjNceDGSqrfq+P3yNP0QECgYBNQfHTM8ygPA4EO/Zg5ONbrOidsuPovXWlgUGP
ApKy+y6iGxBYxAcIO/in71KrijDkRu+ERKo5rs3hWjcWnAedQyZggnFGA8fvDzMf
5JQ0PTazhGUOcthvVAfOqZsFWZ4f+v6tk0UD4pB3chSdwXcUQyjFeorVLlSsMFJl
R4jmNQKBgG38YFR2bqIc7jJItr+34POXdJ4te8Dm1jJHbo8xXsnjVSaxjc5PGs3p
OuJpwuMwzEuFEnE7XLkQxTJw54OBLMmDgK0XUOPDq6eLzrKkW5NlpejqaQV9Piyo
q1kqbJan20jfJQUGTcX7FXHMUThzqJltHILR1GTW6I9z4k8xdsDY
-----END RSA PRIVATE KEY-----
Binary file not shown.

After

Width:  |  Height:  |  Size: 9.5 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 685 B

@@ -0,0 +1,2 @@
/* Dummy CSS for bundle testing */
body { color: red; }
@@ -0,0 +1,2 @@
// Dummy JS for bundle testing
console.log("test");
@@ -0,0 +1,60 @@
esphome:
name: bundle-test
includes:
- includes/custom_sensor.h
esp32:
board: esp32dev
framework:
type: esp-idf
logger:
<<: !include common/base.yaml
wifi:
ssid: !secret wifi_ssid
password: !secret wifi_password
api:
ota:
- platform: esphome
password: !secret ota_password
web_server:
port: 80
css_include: assets/web/custom.css
js_include: assets/web/custom.js
i2c:
sda: GPIO21
scl: GPIO22
font:
- id: test_font
file: assets/fonts/test_font.ttf
size: 16
image:
- id: test_image
file: assets/images/logo.png
type: BINARY
resize: 16x16
animation:
- id: test_animation
file: assets/images/animation.gif
type: BINARY
resize: 16x16
display:
- platform: ssd1306_i2c
model: SSD1306_128X64
address: 0x3C
lambda: |-
it.image(0, 0, id(test_image));
external_components:
- source:
type: local
path: local_components
@@ -0,0 +1 @@
level: DEBUG
@@ -0,0 +1,3 @@
// Dummy custom sensor header for bundle testing
#pragma once
#include "esphome/core/component.h"
@@ -0,0 +1 @@
# Dummy local external component for bundle testing
@@ -0,0 +1,2 @@
// Dummy component header for bundle testing
#pragma once
@@ -0,0 +1,4 @@
wifi_ssid: "TestNetwork"
wifi_password: "TestPassword123"
api_key: "unused_secret_should_not_appear"
ota_password: "ota_test_password"
File diff suppressed because it is too large Load Diff
+196
View File
@@ -24,6 +24,7 @@ from esphome.__main__ import (
_make_crystal_freq_callback,
choose_upload_log_host,
command_analyze_memory,
command_bundle,
command_clean_all,
command_rename,
command_update_all,
@@ -47,6 +48,7 @@ from esphome.__main__ import (
upload_using_picotool,
upload_using_platformio,
)
from esphome.bundle import BUNDLE_EXTENSION, BundleFile, BundleResult
from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
from esphome.const import (
CONF_API,
@@ -1101,6 +1103,8 @@ class MockArgs:
name: str | None = None
dashboard: bool = False
reset: bool = False
list_only: bool = False
output: str | None = None
def test_upload_program_serial_esp32(
@@ -3765,6 +3769,198 @@ esp32:
assert "secrets.yaml" not in summary_section
# --- command_bundle tests ---
def test_command_bundle_list_only(
tmp_path: Path,
capsys: CaptureFixture[str],
) -> None:
"""Test command_bundle with --list-only prints files and returns 0."""
mock_files = [
BundleFile(path="device.yaml", source=tmp_path / "device.yaml"),
BundleFile(path="secrets.yaml", source=tmp_path / "secrets.yaml"),
BundleFile(path="common/base.yaml", source=tmp_path / "common" / "base.yaml"),
]
args = MockArgs(list_only=True)
config: dict[str, Any] = {}
mock_creator = MagicMock()
mock_creator.discover_files.return_value = mock_files
with patch("esphome.bundle.ConfigBundleCreator", return_value=mock_creator):
result = command_bundle(args, config)
assert result == 0
captured = capsys.readouterr()
# Files should be printed in sorted order
assert "common/base.yaml" in captured.out
assert "device.yaml" in captured.out
assert "secrets.yaml" in captured.out
def test_command_bundle_list_only_empty(
tmp_path: Path,
capsys: CaptureFixture[str],
) -> None:
"""Test command_bundle --list-only with no files discovered."""
args = MockArgs(list_only=True)
config: dict[str, Any] = {}
mock_creator = MagicMock()
mock_creator.discover_files.return_value = []
with patch("esphome.bundle.ConfigBundleCreator", return_value=mock_creator):
result = command_bundle(args, config)
assert result == 0
def test_command_bundle_creates_archive(tmp_path: Path) -> None:
"""Test command_bundle creates archive at default output path."""
CORE.config_path = tmp_path / "mydevice.yaml"
mock_result = BundleResult(
data=b"fake-tar-gz-data",
manifest={"manifest_version": 1},
files=[BundleFile(path="mydevice.yaml", source=tmp_path / "mydevice.yaml")],
)
args = MockArgs()
config: dict[str, Any] = {}
mock_creator = MagicMock()
mock_creator.create_bundle.return_value = mock_result
with patch("esphome.bundle.ConfigBundleCreator", return_value=mock_creator):
result = command_bundle(args, config)
assert result == 0
output_path = tmp_path / f"mydevice{BUNDLE_EXTENSION}"
assert output_path.exists()
assert output_path.read_bytes() == b"fake-tar-gz-data"
def test_command_bundle_custom_output(tmp_path: Path) -> None:
"""Test command_bundle with -o custom output path."""
custom_output = tmp_path / "output" / "custom.esphomebundle.tar.gz"
mock_result = BundleResult(
data=b"custom-output-data",
manifest={"manifest_version": 1},
files=[BundleFile(path="mydevice.yaml", source=tmp_path / "mydevice.yaml")],
)
args = MockArgs(output=str(custom_output))
config: dict[str, Any] = {}
mock_creator = MagicMock()
mock_creator.create_bundle.return_value = mock_result
with patch("esphome.bundle.ConfigBundleCreator", return_value=mock_creator):
result = command_bundle(args, config)
assert result == 0
assert custom_output.exists()
assert custom_output.read_bytes() == b"custom-output-data"
def test_command_bundle_creates_parent_dirs(tmp_path: Path) -> None:
"""Test command_bundle creates parent directories for output path."""
nested_output = tmp_path / "deep" / "nested" / "dir" / "out.tar.gz"
mock_result = BundleResult(
data=b"data",
manifest={"manifest_version": 1},
files=[BundleFile(path="mydevice.yaml", source=tmp_path / "mydevice.yaml")],
)
args = MockArgs(output=str(nested_output))
config: dict[str, Any] = {}
mock_creator = MagicMock()
mock_creator.create_bundle.return_value = mock_result
with patch("esphome.bundle.ConfigBundleCreator", return_value=mock_creator):
result = command_bundle(args, config)
assert result == 0
assert nested_output.exists()
def test_command_bundle_logs_info(
tmp_path: Path,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test command_bundle logs bundle creation info."""
CORE.config_path = tmp_path / "mydevice.yaml"
mock_result = BundleResult(
data=b"x" * 2048,
manifest={"manifest_version": 1},
files=[
BundleFile(path="mydevice.yaml", source=tmp_path / "mydevice.yaml"),
BundleFile(path="secrets.yaml", source=tmp_path / "secrets.yaml"),
],
)
args = MockArgs()
config: dict[str, Any] = {}
mock_creator = MagicMock()
mock_creator.create_bundle.return_value = mock_result
with (
patch("esphome.bundle.ConfigBundleCreator", return_value=mock_creator),
caplog.at_level(logging.INFO),
):
result = command_bundle(args, config)
assert result == 0
assert "Bundle created" in caplog.text
assert "2 files" in caplog.text
assert "2.0 KB" in caplog.text
def test_run_esphome_bundle_detection(tmp_path: Path) -> None:
"""Test run_esphome detects .esphomebundle.tar.gz and extracts it."""
bundle_path = tmp_path / f"device{BUNDLE_EXTENSION}"
bundle_path.write_bytes(b"fake-bundle")
extracted_yaml = tmp_path / "extracted" / "device.yaml"
with (
patch("esphome.bundle.is_bundle_path", return_value=True) as mock_is_bundle,
patch(
"esphome.bundle.prepare_bundle_for_compile",
return_value=extracted_yaml,
) as mock_prepare,
patch("esphome.__main__.read_config", return_value=None),
):
result = run_esphome(["esphome", "compile", str(bundle_path)])
mock_is_bundle.assert_called_once()
mock_prepare.assert_called_once_with(bundle_path)
# read_config returns None → exit code 2
assert result == 2
def test_run_esphome_non_bundle_skips_extraction(tmp_path: Path) -> None:
"""Test run_esphome does not extract for regular .yaml files."""
yaml_file = tmp_path / "device.yaml"
yaml_file.write_text("esphome:\n name: test\n")
with (
patch("esphome.bundle.is_bundle_path", return_value=False) as mock_is_bundle,
patch("esphome.bundle.prepare_bundle_for_compile") as mock_prepare,
patch("esphome.__main__.read_config", return_value=None),
):
result = run_esphome(["esphome", "compile", str(yaml_file)])
mock_is_bundle.assert_called_once()
mock_prepare.assert_not_called()
assert result == 2
def test_get_configured_xtal_freq_reads_sdkconfig(tmp_path: Path) -> None:
"""Test reading XTAL_FREQ from sdkconfig."""
CORE.name = "test-device"
+54
View File
@@ -323,6 +323,60 @@ def test_dump_sort_keys() -> None:
assert sorted_dump.index("a_key:") < sorted_dump.index("z_key:")
# ---------------------------------------------------------------------------
# track_yaml_loads
# ---------------------------------------------------------------------------
def test_track_yaml_loads_records_files(tmp_path: Path) -> None:
"""track_yaml_loads records every file loaded inside the context."""
yaml_file = tmp_path / "test.yaml"
yaml_file.write_text("key: value\n")
with yaml_util.track_yaml_loads() as loaded:
yaml_util.load_yaml(yaml_file)
assert len(loaded) == 1
assert loaded[0] == yaml_file.resolve()
def test_track_yaml_loads_records_includes(tmp_path: Path) -> None:
"""track_yaml_loads records nested !include files."""
inc = tmp_path / "included.yaml"
inc.write_text("included_key: 42\n")
main = tmp_path / "main.yaml"
main.write_text("child: !include included.yaml\n")
with yaml_util.track_yaml_loads() as loaded:
yaml_util.load_yaml(main)
resolved = [p.name for p in loaded]
assert "main.yaml" in resolved
assert "included.yaml" in resolved
def test_track_yaml_loads_empty_outside_context(tmp_path: Path) -> None:
"""Files loaded outside the context are not recorded."""
yaml_file = tmp_path / "test.yaml"
yaml_file.write_text("key: value\n")
with yaml_util.track_yaml_loads() as loaded:
pass # load nothing inside
yaml_util.load_yaml(yaml_file)
assert loaded == []
def test_track_yaml_loads_cleanup_on_exception(tmp_path: Path) -> None:
"""Listener is removed even if the body raises."""
before = len(yaml_util._load_listeners)
with pytest.raises(RuntimeError), yaml_util.track_yaml_loads():
raise RuntimeError("boom")
assert len(yaml_util._load_listeners) == before
@pytest.mark.parametrize(
"data",
[