import json import os import stat import pytest from config_store import ConfigError, ConfigStore def test_round_trip_preserves_comments_and_applies(store): store.apply({"quarantine": {"default_limit": 500}}, actor="t") txt = store.path.read_text() assert "# UI default row count" in txt # comment survived assert "default_limit = 500" in txt # value written assert store.snapshot().quarantine["default_limit"] == 500 def test_atomic_write_perms_and_backup(store): store.apply({"quarantine": {"chunk_size": 40}}, actor="t") mode = stat.S_IMODE(os.stat(store.path).st_mode) assert mode == 0o600 bak = store.path.with_name(store.path.name + ".bak") assert bak.exists() # no temp files left behind leftovers = list(store.path.parent.glob(".config.*.tmp")) assert not leftovers def test_redacted_never_contains_secrets(store): # Set a distinctive password value that isn't a substring of any key name. store.apply({"pps": {"password": "ZZTOPSECRET42"}}, actor="t") red = store.redacted() assert "ZZTOPSECRET42" not in json.dumps(red) assert "password" not in red["pps"] assert red["pps"]["password_set"] is True def test_blank_password_leaves_value_unchanged(store): store.apply({"pps": {"password": "keepme"}}, actor="t") store.apply({"pps": {"password": ""}}, actor="t") assert ConfigStore(store.path).snapshot().pps["password"] == "keepme" @pytest.mark.parametrize("patch,field", [ ({"quarantine": {"default_folder": "Nope"}}, "quarantine.default_folder"), ({"quarantine": {"folders": ["a,b"]}}, "quarantine.folders"), ({"pps": {"base_url": "ftp://x"}}, "pps.base_url"), ({"quarantine": {"report_release": {"steps": ["move", "release"]}}}, "quarantine.report_release.steps"), ({"quarantine": {"report_release": {"steps": ["bogus"]}}}, "quarantine.report_release.steps"), ({"pps": {"timeout": 0}}, "pps.timeout"), ]) def test_validation_rejects(store, patch, field): with pytest.raises(ConfigError) as exc: store.apply(patch, actor="t") assert any(e["field"] == field for e in exc.value.errors) def test_restart_vs_live_classification(store): r = store.apply({"app": {"log_level": "DEBUG"}}, actor="t") assert r.restart_required == [] # log_level is live r = store.apply({"quarantine": {"chunk_size": 30}}, actor="t") assert r.restart_required == [] def test_non_editable_key_rejected(store): with pytest.raises(ConfigError) as exc: store.apply({"app": {"port": 9999}}, actor="t") # port not admin-editable assert exc.value.errors[0]["field"] == "app.port" def test_version_increments_and_snapshot_immutable(store): v1 = store.snapshot().version snap1 = store.snapshot() store.apply({"quarantine": {"default_limit": 300}}, actor="t") assert store.snapshot().version > v1 # old snapshot unchanged assert snap1.quarantine["default_limit"] != 300 or v1 != store.snapshot().version def test_pps_client_rebuilt_only_on_connection_change(store): c0 = store.pps() store.apply({"quarantine": {"default_limit": 250}}, actor="t") # not a conn key assert store.pps() is c0 store.apply({"pps": {"timeout": 45}}, actor="t") # conn key assert store.pps() is not c0 def test_migration_report_release_folder(tmp_path): cfg = tmp_path / "config.toml" cfg.write_text( '[pps]\nbase_url="http://x:10000"\nusername="u"\npassword="p"\n' '[quarantine]\nfolders=["Quarantine","Rep"]\ndefault_folder="Quarantine"\n' 'deleted_folder="Quarantine"\nlist_query="from=*"\nreport_release_folder="Rep"\n' 'default_sort_field="subject"\ndefault_sort_dir="asc"\n' '[app]\nsecret_key="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"\n' '[auth]\nmode="static"\n' ) snap = ConfigStore(cfg).snapshot() rr = snap.quarantine["report_release"] assert list(rr["steps"]) == ["release", "move"] assert rr["move_target"] == "Rep" def test_ppsq_config_guard_under_pytest(monkeypatch): monkeypatch.delenv("PPSQ_CONFIG", raising=False) monkeypatch.setenv("PYTEST_CURRENT_TEST", "x") with pytest.raises(RuntimeError): ConfigStore() # no explicit path, no env -> refuse