proofpoint-quarantine-portal/tests/test_config_store.py

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": ["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"]) == ["move", "release"]
    assert rr["move_target"] == "Rep"


def test_migration_reorders_pre_flip_steps(store):
    """Configs written when release ran before move are re-ordered, not rejected."""
    store.apply({"quarantine": {"report_release": {"steps": ["release", "move"]}}},
                actor="t")
    rr = store.snapshot().quarantine["report_release"]
    assert list(rr["steps"]) == ["move", "release"]


def test_migration_drops_the_retired_delete_step(store):
    """Report & Release no longer deletes; an old config with the step still boots."""
    store.apply({"quarantine": {"report_release": {"steps": ["release", "move", "delete"]}}},
                actor="t")
    rr = store.snapshot().quarantine["report_release"]
    assert list(rr["steps"]) == ["move", "release"]


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