import pytest
def test_worker_bootstrap_loads_config_from_switches(tmp_path, monkeypatch):
"""worker_bootstrap 从 SWITCHES.config_path 加载 yaml(含 TLS)——根因测试。"""
import ttk.config.loader as loader
from ttk.utilities.classes import SWITCHES
import ttk.core_modules.tbe_multiprocessing.pool as pool
monkeypatch.setattr(pool, "default_logging_config", lambda **kw: None)
yaml_path = tmp_path / "worker-test.yaml"
yaml_path.write_text(
'remote:\n'
' tls_ca: "/fake/ca.crt"\n'
' endpoints:\n'
' - host: 10.0.0.1\n'
' port: 9090\n'
)
saved = loader._config
loader._config = None
try:
sw = SWITCHES()
sw.config_path = str(yaml_path)
sw.logging_to_file = False
from ttk.core_modules.tbe_multiprocessing.pool import worker_bootstrap
worker_bootstrap(sw)
rc = loader.get_remote_config()
assert rc is not None
assert rc.tls_ca == "/fake/ca.crt"
assert len(rc.endpoints) == 1
finally:
loader._config = saved
loader.load_config()
def test_worker_bootstrap_without_config_path_loads_defaults(monkeypatch):
"""config_path=None 时 worker_bootstrap 走标准路径(不 crash)。"""
import ttk.config.loader as loader
from ttk.utilities.classes import SWITCHES
import ttk.core_modules.tbe_multiprocessing.pool as pool
monkeypatch.setattr(pool, "default_logging_config", lambda **kw: None)
saved = loader._config
loader._config = None
try:
sw = SWITCHES()
sw.config_path = None
sw.logging_to_file = False
from ttk.core_modules.tbe_multiprocessing.pool import worker_bootstrap
worker_bootstrap(sw)
cfg = loader.get_config()
assert "remote" in cfg
finally:
loader._config = saved
loader.load_config()