argus-netconf-exporter/tests/test_config.py

149 lines
4.1 KiB
Python

from pathlib import Path
import base64
import pytest
from exporter.config import Config
VALID_FERNET_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=" # 44 chars base64 -> 32 bytes
def test_config_loads_global_fields(tmp_path: Path):
yaml_content = f"""
global:
http_listen: "0.0.0.0:9100"
scrape_interval_seconds: 60
netconf_port: 830
connect_timeout_seconds: 5
rpc_timeout_seconds: 10
max_workers: 10
api_token: "changeme"
runtime_db_path: "./devices.db"
password_secret: "{VALID_FERNET_KEY}"
ssh_keepalive_seconds: 30
failure_threshold: 3
max_backoff_factor: 8
shutdown_timeout_seconds: 30
log_level: INFO
log_to_stdout: true
log_file: ""
log_file_max_bytes: 10485760
log_file_backup_count: 5
"""
cfg_file = tmp_path / "config.yaml"
cfg_file.write_text(yaml_content)
cfg = Config.from_file(cfg_file)
g = cfg.global_
assert g.http_listen == "0.0.0.0:9100"
assert g.scrape_interval_seconds == 60
assert g.rpc_timeout_seconds == 10
assert g.runtime_db_path == "./devices.db"
assert g.password_secret == VALID_FERNET_KEY
assert g.ssh_keepalive_seconds == 30
assert g.failure_threshold == 3
assert g.max_backoff_factor == 8
assert g.shutdown_timeout_seconds == 30
def test_config_invalid_fernet_key_raises():
data = {
"global": {
"runtime_db_path": "./devices.db",
"password_secret": "too-short",
}
}
with pytest.raises(ValueError, match="Invalid Fernet key"):
Config.from_dict(data)
def test_config_missing_password_secret_raises():
data = {
"global": {
"runtime_db_path": "./devices.db",
# password_secret 缺失
}
}
with pytest.raises(ValueError, match="global.password_secret must be configured"):
Config.from_dict(data)
def test_config_invalid_fernet_key_length_raises():
# 构造一个合法 base64 但长度不是 32 字节的 key
bad_key_bytes = b"too-short"
bad_key = base64.urlsafe_b64encode(bad_key_bytes).decode()
data = {
"global": {
"runtime_db_path": "./devices.db",
"password_secret": bad_key,
}
}
with pytest.raises(ValueError, match="Invalid Fernet key length"):
Config.from_dict(data)
def test_shutdown_timeout_too_small_warns():
data = {
"global": {
"http_listen": "0.0.0.0:9100",
"scrape_interval_seconds": 60,
"netconf_port": 830,
"connect_timeout_seconds": 5,
"rpc_timeout_seconds": 20,
"max_workers": 10,
"api_token": "changeme",
"runtime_db_path": "./devices.db",
"password_secret": VALID_FERNET_KEY,
"ssh_keepalive_seconds": 30,
"failure_threshold": 3,
"max_backoff_factor": 8,
"shutdown_timeout_seconds": 10,
"log_level": "INFO",
"log_to_stdout": True,
"log_file": "",
"log_file_max_bytes": 10485760,
"log_file_backup_count": 5,
}
}
with pytest.warns(UserWarning):
Config.from_dict(data)
def test_deviceconfig_vendor_parsed_and_normalized_from_yaml(tmp_path: Path):
yaml_content = f"""
global:
runtime_db_path: "./devices.db"
password_secret: "{VALID_FERNET_KEY}"
devices:
- name: rj-1
host: 192.0.2.10
port: 830
username: u
password: p
enabled: true
supports_xpath: false
vendor: " Ruijie "
- name: h3c-1
host: 198.51.100.10
port: 830
username: u2
password: p2
enabled: true
supports_xpath: false
"""
cfg_file = tmp_path / "config_vendor.yaml"
cfg_file.write_text(yaml_content)
cfg = Config.from_file(cfg_file)
assert len(cfg.devices) == 2
rj = next(d for d in cfg.devices if d.name == "rj-1")
h3c = next(d for d in cfg.devices if d.name == "h3c-1")
# vendor 显式配置应被 strip + lower
assert rj.vendor == "ruijie"
# 未配置 vendor 时应为 None
assert h3c.vendor is None