"""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_empty_tags_rejected(tmp_path: Path) -> None: p = _write(tmp_path, "opc:\n server: X\ntags: []\n") with pytest.raises(ConfigError, match="at least one tag"): load_config(p) 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)