PacketStation/tests/server/test_settings.py

from __future__ import annotations

from pathlib import Path

import pytest
from pydantic import ValidationError

from packetstation import logs
from packetstation.settings import SecretFileError, Settings, read_secret


def test_defaults_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("STACK", "TEST")
    monkeypatch.setenv("PG_PORT", "5433")
    monkeypatch.setenv("LDAP_URL", "ldaps://dc1.pesco.com ldaps://dc2.pesco.com")
    monkeypatch.setenv("LDAP_CA_BUNDLE", "")
    s = Settings()
    assert s.stack == "test"
    assert s.pg_port == 5433
    assert s.site_tz == "America/Denver"
    assert s.pg_app_user == "ps_app" and s.pg_migrate_user == "ps_migrate"
    assert s.pg_app_password_file == Path("/run/secrets/postgres_app_password")
    assert s.ldap_urls == ["ldaps://dc1.pesco.com", "ldaps://dc2.pesco.com"]
    assert s.ldap_ca_bundle is None
    assert s.ldap_user_filter == "(|(sAMAccountName={0})(userPrincipalName={0}))"
    assert s.smtp_starttls is True and s.notify_dry_run is False
    assert s.ps_build == "dev"


def test_stack_required(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("STACK")
    with pytest.raises(ValidationError):
        Settings()


def test_stack_rejects_unknown(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("STACK", "staging")
    with pytest.raises(ValidationError):
        Settings()


def test_ps_env_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    env = tmp_path / "dev.env"
    env.write_text("STACK=dev\nSITE_ID=TESTSITE\n")
    monkeypatch.delenv("STACK")
    monkeypatch.setenv("PS_ENV_FILE", str(env))
    s = Settings()
    assert s.stack == "dev" and s.site_id == "TESTSITE"


def test_read_secret_strips_crlf(tmp_path: Path) -> None:
    f = tmp_path / "pw"
    f.write_bytes(b"s3cr3t value\r\n\r\n")
    assert read_secret(f) == "s3cr3t value"


def test_read_secret_keeps_inner_whitespace(tmp_path: Path) -> None:
    f = tmp_path / "pw"
    f.write_bytes(b" a b \n")
    assert read_secret(f) == " a b "


def test_read_secret_empty_is_error(tmp_path: Path) -> None:
    f = tmp_path / "pw"
    f.write_bytes(b"\r\n")
    with pytest.raises(SecretFileError, match="empty"):
        read_secret(f)


def test_read_secret_missing_names_file(tmp_path: Path) -> None:
    with pytest.raises(SecretFileError, match="nope"):
        read_secret(tmp_path / "nope", "PG_APP_PASSWORD_FILE")


def test_pg_password_by_role(tmp_path: Path) -> None:
    (tmp_path / "app").write_text("apppw\n")
    (tmp_path / "mig").write_text("migpw\n")
    s = Settings(
        _env_file=None, stack="ci", pg_app_password_file=tmp_path / "app", pg_migrate_password_file=tmp_path / "mig"
    )
    assert s.pg_password("app") == "apppw"
    assert s.pg_password("migrate") == "migpw"


def test_db_connect_kwargs(tmp_path: Path) -> None:
    from packetstation import db

    (tmp_path / "app").write_text("apppw\n")
    s = Settings(_env_file=None, stack="ci", pg_app_password_file=tmp_path / "app")
    k = db.connect_kwargs("worker", s)
    assert k["user"] == "ps_app" and k["application_name"] == "ps-worker" and k["password"] == "apppw"
    assert db.connect_kwargs("app", s)["application_name"] == "ps-app"


def test_ldap_insecure_refused_on_prod() -> None:
    with pytest.raises(ValidationError, match="LDAP_ALLOW_INSECURE"):
        Settings(_env_file=None, stack="prod", ldap_allow_insecure=True)
    assert Settings(_env_file=None, stack="test", ldap_allow_insecure=True).ldap_allow_insecure


def test_log_level_validated() -> None:
    assert Settings(_env_file=None, stack="ci", log_level="debug").log_level == "DEBUG"
    with pytest.raises(ValidationError):
        Settings(_env_file=None, stack="ci", log_level="loud")


@pytest.mark.parametrize("text", ["password=hunter2", "PWD=abc;x", "connect secret=xyz now", "user=a Password=p@ss"])
def test_redaction(text: str) -> None:
    out = logs.redact(text)
    for word in ("hunter2", "abc;x", "xyz", "p@ss"):
        assert word not in out
    assert "***" in out


def test_log_lines_are_redacted_json(capsys: pytest.CaptureFixture[str]) -> None:
    import json
    import logging

    logs.setup("INFO")
    logging.getLogger("t").info("dsn host=x password=%s", "topsecret")
    line = capsys.readouterr().out.strip().splitlines()[-1]
    rec = json.loads(line)
    assert rec["level"] == "INFO" and "topsecret" not in line and "password=***" in rec["msg"]