202 lines
7.2 KiB
Python
202 lines
7.2 KiB
Python
"""Config loading + validation tests (IMPLEMENTATION_PLAN §4.1)."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from opcda_diagnozer.config import ConfigError, load_config
|
|
|
|
EXAMPLE = Path(__file__).resolve().parents[1] / "config.example.yaml"
|
|
|
|
|
|
def _write(tmp_path: Path, text: str) -> Path:
|
|
p = tmp_path / "config.yaml"
|
|
p.write_text(text, encoding="utf-8")
|
|
return p
|
|
|
|
|
|
def test_example_config_loads() -> None:
|
|
config = load_config(EXAMPLE)
|
|
assert config.opc.server == "Matrikon.OPC.Simulation.1"
|
|
assert config.opc.host is None # host commented out -> local server
|
|
assert len(config.tags) == 2
|
|
assert config.tags[0].name == "Channel1.Device1.Counter"
|
|
assert config.tags[0].expect_change is True
|
|
assert config.tags[1].expect_change is False
|
|
# D16 device-link indicators (spec 6.1); normal reading configurable per device
|
|
assert [d.name for d in config.devices] == ["Conveyor 3", "Conveyor 4", "Pump A", "Boiler 2"]
|
|
assert config.devices[0].tag == "Channel1.Device3.Connected"
|
|
assert config.devices[0].ok_values == ("true",) # default: boolean connected-flag
|
|
assert config.devices[1].ok_values == ("true", "1")
|
|
assert config.devices[2].ok_values == ("false", "0") # inverted fault flag: 0 = ok
|
|
assert config.devices[3].ok_values == ("RUN",) # string state
|
|
assert config.poll.interval_s == 1.0
|
|
assert config.poll.timeout_ms == 5000
|
|
assert config.latency.warn_ms == 1000
|
|
assert config.report.eventlog.min_severity == "WARN"
|
|
assert config.report.jsonl.dir == "C:/ProgramData/OPCDA-Diagnozer/logs"
|
|
|
|
|
|
def test_missing_file_is_actionable() -> None:
|
|
with pytest.raises(ConfigError, match="not found"):
|
|
load_config("no-such-config.yaml")
|
|
|
|
|
|
def test_missing_opc_section_is_actionable(tmp_path: Path) -> None:
|
|
p = _write(tmp_path, "tags:\n - name: t\n")
|
|
with pytest.raises(ConfigError, match=r"opc: missing required key 'server'"):
|
|
load_config(p)
|
|
|
|
|
|
def test_unknown_key_rejected(tmp_path: Path) -> None:
|
|
p = _write(tmp_path, "opc:\n server: X\n typo_key: 1\ntags:\n - name: t\n")
|
|
with pytest.raises(ConfigError, match="typo_key"):
|
|
load_config(p)
|
|
|
|
|
|
def test_unknown_top_level_section_rejected(tmp_path: Path) -> None:
|
|
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\nnonsense: {}\n")
|
|
with pytest.raises(ConfigError, match="nonsense"):
|
|
load_config(p)
|
|
|
|
|
|
def test_both_tags_and_devices_empty_rejected(tmp_path: Path) -> None:
|
|
"""tags: and devices: are each optional, but their union must not be empty."""
|
|
p = _write(tmp_path, "opc:\n server: X\ntags: []\ndevices: []\n")
|
|
with pytest.raises(ConfigError, match="nothing to monitor"):
|
|
load_config(p)
|
|
|
|
|
|
def test_no_tags_section_with_devices_is_allowed(tmp_path: Path) -> None:
|
|
"""Devices-only config: link monitoring without extra surveillance tags."""
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\n"
|
|
"devices:\n - { name: 'Pump A', tag: Ind.Tag }\n",
|
|
)
|
|
config = load_config(p)
|
|
assert config.tags == ()
|
|
assert [d.name for d in config.devices] == ["Pump A"]
|
|
|
|
|
|
def test_tags_only_still_allowed(tmp_path: Path) -> None:
|
|
"""The classic setup (no device indicators) stays valid."""
|
|
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\n")
|
|
config = load_config(p)
|
|
assert len(config.tags) == 1
|
|
assert config.devices == ()
|
|
|
|
|
|
def test_duplicate_tags_rejected(tmp_path: Path) -> None:
|
|
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: a\n - name: a\n")
|
|
with pytest.raises(ConfigError, match="duplicate"):
|
|
load_config(p)
|
|
|
|
|
|
def test_devices_missing_keys_rejected(tmp_path: Path) -> None:
|
|
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\ndevices:\n - name: Device\n")
|
|
with pytest.raises(ConfigError, match="'tag' is required"):
|
|
load_config(p)
|
|
|
|
|
|
def test_devices_blank_name_rejected(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\ndevices:\n - { name: '', tag: Ind.Tag }\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="non-empty"):
|
|
load_config(p)
|
|
|
|
|
|
def test_devices_duplicate_indicator_rejected(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\n"
|
|
"devices:\n - { name: A, tag: Ind.Tag }\n - { name: B, tag: Ind.Tag }\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="one device per indicator"):
|
|
load_config(p)
|
|
|
|
|
|
def test_devices_indicator_in_tags_rejected(tmp_path: Path) -> None:
|
|
"""An indicator is polled automatically; listing it in tags: double-reports."""
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\n"
|
|
"tags:\n - name: Ind.Tag\ndevices:\n - { name: A, tag: Ind.Tag }\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="remove them from"):
|
|
load_config(p)
|
|
|
|
|
|
def test_devices_empty_is_allowed(tmp_path: Path) -> None:
|
|
"""devices: is optional (no link monitoring) and may be an empty list."""
|
|
p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\ndevices: []\n")
|
|
assert load_config(p).devices == ()
|
|
|
|
|
|
def test_devices_ok_values_default_is_true(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\ndevices:\n - { name: A, tag: I.Tag }\n",
|
|
)
|
|
device = load_config(p).devices[0]
|
|
assert device.ok_values == ("true",)
|
|
|
|
|
|
def test_devices_ok_values_normalized(tmp_path: Path) -> None:
|
|
"""Booleans/ints/strings all render to comparison strings; scalar = shorthand."""
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\n"
|
|
"devices:\n"
|
|
" - { name: A, tag: I.One, ok_values: [true, 1] }\n"
|
|
" - { name: B, tag: I.Two, ok_values: [false, 0] }\n"
|
|
" - { name: C, tag: I.Three, ok_values: [\"RUN\"] }\n"
|
|
" - { name: D, tag: I.Four, ok_values: 1 }\n",
|
|
)
|
|
devices = {d.name: d for d in load_config(p).devices}
|
|
assert devices["A"].ok_values == ("true", "1")
|
|
assert devices["B"].ok_values == ("false", "0")
|
|
assert devices["C"].ok_values == ("RUN",)
|
|
assert devices["D"].ok_values == ("1",) # single scalar shorthand
|
|
|
|
|
|
def test_devices_ok_values_empty_rejected(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\n"
|
|
"devices:\n - { name: A, tag: I.Tag, ok_values: [] }\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="must not be empty"):
|
|
load_config(p)
|
|
|
|
|
|
def test_devices_ok_values_non_scalar_rejected(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\n"
|
|
"devices:\n - { name: A, tag: I.Tag, ok_values: [[1]] }\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="scalars"):
|
|
load_config(p)
|
|
|
|
|
|
def test_latency_threshold_order_rejected(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\n"
|
|
"latency:\n warn_ms: 5000\n critical_ms: 1000\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="warn_ms"):
|
|
load_config(p)
|
|
|
|
|
|
def test_hard_hang_must_exceed_timeout(tmp_path: Path) -> None:
|
|
p = _write(
|
|
tmp_path,
|
|
"opc:\n server: X\ntags:\n - name: t\n"
|
|
"poll:\n timeout_ms: 5000\n hard_hang_s: 4\n",
|
|
)
|
|
with pytest.raises(ConfigError, match="hard_hang_s"):
|
|
load_config(p) |