test_config_store.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. import json
  2. import os
  3. import stat
  4. import pytest
  5. from config_store import ConfigError, ConfigStore
  6. def test_round_trip_preserves_comments_and_applies(store):
  7. store.apply({"quarantine": {"default_limit": 500}}, actor="t")
  8. txt = store.path.read_text()
  9. assert "# UI default row count" in txt # comment survived
  10. assert "default_limit = 500" in txt # value written
  11. assert store.snapshot().quarantine["default_limit"] == 500
  12. def test_atomic_write_perms_and_backup(store):
  13. store.apply({"quarantine": {"chunk_size": 40}}, actor="t")
  14. mode = stat.S_IMODE(os.stat(store.path).st_mode)
  15. assert mode == 0o600
  16. bak = store.path.with_name(store.path.name + ".bak")
  17. assert bak.exists()
  18. # no temp files left behind
  19. leftovers = list(store.path.parent.glob(".config.*.tmp"))
  20. assert not leftovers
  21. def test_redacted_never_contains_secrets(store):
  22. # Set a distinctive password value that isn't a substring of any key name.
  23. store.apply({"pps": {"password": "ZZTOPSECRET42"}}, actor="t")
  24. red = store.redacted()
  25. assert "ZZTOPSECRET42" not in json.dumps(red)
  26. assert "password" not in red["pps"]
  27. assert red["pps"]["password_set"] is True
  28. def test_blank_password_leaves_value_unchanged(store):
  29. store.apply({"pps": {"password": "keepme"}}, actor="t")
  30. store.apply({"pps": {"password": ""}}, actor="t")
  31. assert ConfigStore(store.path).snapshot().pps["password"] == "keepme"
  32. @pytest.mark.parametrize("patch,field", [
  33. ({"quarantine": {"default_folder": "Nope"}}, "quarantine.default_folder"),
  34. ({"quarantine": {"folders": ["a,b"]}}, "quarantine.folders"),
  35. ({"pps": {"base_url": "ftp://x"}}, "pps.base_url"),
  36. ({"quarantine": {"report_release": {"steps": ["move", "release"]}}},
  37. "quarantine.report_release.steps"),
  38. ({"quarantine": {"report_release": {"steps": ["bogus"]}}},
  39. "quarantine.report_release.steps"),
  40. ({"pps": {"timeout": 0}}, "pps.timeout"),
  41. ])
  42. def test_validation_rejects(store, patch, field):
  43. with pytest.raises(ConfigError) as exc:
  44. store.apply(patch, actor="t")
  45. assert any(e["field"] == field for e in exc.value.errors)
  46. def test_restart_vs_live_classification(store):
  47. r = store.apply({"app": {"log_level": "DEBUG"}}, actor="t")
  48. assert r.restart_required == [] # log_level is live
  49. r = store.apply({"quarantine": {"chunk_size": 30}}, actor="t")
  50. assert r.restart_required == []
  51. def test_non_editable_key_rejected(store):
  52. with pytest.raises(ConfigError) as exc:
  53. store.apply({"app": {"port": 9999}}, actor="t") # port not admin-editable
  54. assert exc.value.errors[0]["field"] == "app.port"
  55. def test_version_increments_and_snapshot_immutable(store):
  56. v1 = store.snapshot().version
  57. snap1 = store.snapshot()
  58. store.apply({"quarantine": {"default_limit": 300}}, actor="t")
  59. assert store.snapshot().version > v1
  60. # old snapshot unchanged
  61. assert snap1.quarantine["default_limit"] != 300 or v1 != store.snapshot().version
  62. def test_pps_client_rebuilt_only_on_connection_change(store):
  63. c0 = store.pps()
  64. store.apply({"quarantine": {"default_limit": 250}}, actor="t") # not a conn key
  65. assert store.pps() is c0
  66. store.apply({"pps": {"timeout": 45}}, actor="t") # conn key
  67. assert store.pps() is not c0
  68. def test_migration_report_release_folder(tmp_path):
  69. cfg = tmp_path / "config.toml"
  70. cfg.write_text(
  71. '[pps]\nbase_url="http://x:10000"\nusername="u"\npassword="p"\n'
  72. '[quarantine]\nfolders=["Quarantine","Rep"]\ndefault_folder="Quarantine"\n'
  73. 'deleted_folder="Quarantine"\nlist_query="from=*"\nreport_release_folder="Rep"\n'
  74. 'default_sort_field="subject"\ndefault_sort_dir="asc"\n'
  75. '[app]\nsecret_key="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"\n'
  76. '[auth]\nmode="static"\n'
  77. )
  78. snap = ConfigStore(cfg).snapshot()
  79. rr = snap.quarantine["report_release"]
  80. assert list(rr["steps"]) == ["release", "move"]
  81. assert rr["move_target"] == "Rep"
  82. def test_ppsq_config_guard_under_pytest(monkeypatch):
  83. monkeypatch.delenv("PPSQ_CONFIG", raising=False)
  84. monkeypatch.setenv("PYTEST_CURRENT_TEST", "x")
  85. with pytest.raises(RuntimeError):
  86. ConfigStore() # no explicit path, no env -> refuse