Octop/tests/unit/test_config.py

"""tests/unit/test_config.py"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from octop.config import load_config, parse_database_config


def test_defaults_when_missing(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg = load_config(cfg_path)
    assert cfg.bind_host == "127.0.0.1"
    assert cfg.port == 8088
    assert cfg.access_token_ttl_seconds == 86400
    assert cfg.enable_dashboard is True
    assert cfg.enable_api_docs is False
    assert cfg.require_setup_password is True
    assert cfg.database.driver == "sqlite"
    assert cfg.database.sqlite_path == "octop.db"
    assert cfg.database.is_sqlite
    assert cfg.database_in_file is False
    assert cfg.backup.auto_enabled is False
    assert cfg.backup.schedule == "cron:0 4 * * *"
    assert cfg.backup.retention_count == 7
    assert cfg.backup.include_config is True
    assert cfg.backup.include_workspaces is True
    assert cfg.backup.include_skill_packages is True
    assert cfg.backup.include_plugins is True
    assert cfg.backup.include_knowledge is True
    assert cfg.backup.include_chats is False
    assert cfg.max_upload_mb == 100
    assert cfg.max_upload_bytes == 100 * 1024 * 1024
    assert cfg_path.exists()  # file written with defaults
    written = json.loads(cfg_path.read_text(encoding="utf-8"))
    assert "password" not in written.get("database", {})
    assert written.get("backup", {}).get("auto_enabled") is False
    assert written["max_upload_mb"] == 100


def test_loads_backup_section(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "backup": {
                    "auto_enabled": True,
                    "schedule": "interval:7200",
                    "retention_count": 3,
                    "include_config": False,
                    "include_workspaces": False,
                    "include_skill_packages": False,
                    "include_plugins": False,
                    "include_knowledge": False,
                    "include_chats": True,
                }
            }
        )
    )
    cfg = load_config(cfg_path)
    assert cfg.backup.auto_enabled is True
    assert cfg.backup.schedule == "interval:7200"
    assert cfg.backup.retention_count == 3
    assert cfg.backup.include_config is False
    assert cfg.backup.include_workspaces is False
    assert cfg.backup.include_skill_packages is False
    assert cfg.backup.include_plugins is False
    assert cfg.backup.include_knowledge is False
    assert cfg.backup.include_chats is True


def test_backup_env_overrides(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_BACKUP_AUTO_ENABLED", "true")
    monkeypatch.setenv("OCTOP_BACKUP_SCHEDULE", "cron:30 5 * * *")
    monkeypatch.setenv("OCTOP_BACKUP_RETENTION_COUNT", "5")
    monkeypatch.setenv("OCTOP_BACKUP_INCLUDE_CHATS", "true")
    monkeypatch.setenv("OCTOP_BACKUP_INCLUDE_CONFIG", "false")
    monkeypatch.setenv("OCTOP_BACKUP_INCLUDE_WORKSPACES", "false")
    monkeypatch.setenv("OCTOP_BACKUP_INCLUDE_SKILL_PACKAGES", "false")
    monkeypatch.setenv("OCTOP_BACKUP_INCLUDE_PLUGINS", "false")
    monkeypatch.setenv("OCTOP_BACKUP_INCLUDE_KNOWLEDGE", "false")
    cfg = load_config(tmp_path / "config.json")
    assert cfg.backup.auto_enabled is True
    assert cfg.backup.schedule == "cron:30 5 * * *"
    assert cfg.backup.retention_count == 5
    assert cfg.backup.include_chats is True
    assert cfg.backup.include_config is False
    assert cfg.backup.include_workspaces is False
    assert cfg.backup.include_skill_packages is False
    assert cfg.backup.include_plugins is False
    assert cfg.backup.include_knowledge is False


def test_loads_existing(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"port": 9000, "log_level": "debug"}))
    cfg = load_config(cfg_path)
    assert cfg.port == 9000
    assert cfg.log_level == "debug"
    assert cfg.bind_host == "127.0.0.1"  # default fills
    assert cfg.max_upload_mb == 100  # missing key uses default


def test_loads_max_upload_mb(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"max_upload_mb": 50}))
    cfg = load_config(cfg_path)
    assert cfg.max_upload_mb == 50
    assert cfg.max_upload_bytes == 50 * 1024 * 1024


def test_max_upload_mb_env_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_MAX_UPLOAD_MB", "25")
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"max_upload_mb": 50}))
    cfg = load_config(cfg_path)
    assert cfg.max_upload_mb == 25
    assert cfg.max_upload_bytes == 25 * 1024 * 1024


def test_max_upload_mb_rejects_non_positive(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"max_upload_mb": 0}))
    cfg = load_config(cfg_path)
    assert cfg.max_upload_mb == 100


def test_loads_database_section(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "database": {
                    "driver": "sqlite",
                    "sqlite_path": "data/custom.db",
                }
            }
        )
    )
    cfg = load_config(cfg_path)
    assert cfg.database.sqlite_path == "data/custom.db"
    assert cfg.database_in_file is True
    assert cfg.database.resolve_sqlite_path(tmp_path) == tmp_path / "data" / "custom.db"


def test_env_overrides(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_PORT", "7000")
    monkeypatch.setenv("OCTOP_LOG_LEVEL", "warning")
    monkeypatch.setenv("OCTOP_BROWSER_IDLE_TIMEOUT_MINUTES", "12")
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"port": 8088}))
    cfg = load_config(cfg_path)
    assert cfg.port == 7000
    assert cfg.log_level == "warning"
    assert cfg.browser_idle_timeout_minutes == 12


def test_browser_idle_timeout_defaults_and_can_be_disabled(tmp_path: Path):
    cfg = load_config(tmp_path / "config.json")
    assert cfg.browser_idle_timeout_minutes == 30

    cfg_path = tmp_path / "disabled.json"
    cfg_path.write_text(json.dumps({"browser_idle_timeout_minutes": 0}))
    assert load_config(cfg_path).browser_idle_timeout_minutes == 0


def test_database_env_sqlite_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_DATABASE_SQLITE_PATH", "/tmp/octop-test.db")
    cfg = load_config(tmp_path / "config.json")
    assert cfg.database.sqlite_path == "/tmp/octop-test.db"
    resolved = cfg.database.resolve_sqlite_path(tmp_path)
    assert resolved.is_absolute()
    assert resolved.as_posix().endswith("/tmp/octop-test.db")


def test_database_env_url_postgresql(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_DATABASE_URL", "postgresql://alice:secret@db.example.com:5433/mydb")
    cfg = load_config(tmp_path / "config.json")
    assert cfg.database.is_postgresql
    assert cfg.database.host == "db.example.com"
    assert cfg.database.port == 5433
    assert cfg.database.database == "mydb"
    assert cfg.database.user == "alice"
    assert cfg.database.password == "secret"


def test_database_url_preserves_query_for_conninfo(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv(
        "OCTOP_DATABASE_URL",
        "postgresql://alice:secret@db.example.com:5433/mydb?sslmode=require",
    )
    cfg = load_config(tmp_path / "config.json")
    assert cfg.database.is_postgresql
    assert cfg.database.url == "postgresql://alice:secret@db.example.com:5433/mydb?sslmode=require"
    assert "sslmode=require" in cfg.database.postgresql_conninfo()


def test_postgresql_conninfo_from_discrete_fields(tmp_path: Path):
    (tmp_path / "config.json").write_text(
        json.dumps(
            {
                "database": {
                    "driver": "postgresql",
                    "host": "127.0.0.1",
                    "port": 5432,
                    "database": "octop",
                    "user": "octop",
                    "password": "s3cret",
                }
            }
        ),
        encoding="utf-8",
    )
    cfg = load_config(tmp_path / "config.json")
    info = cfg.database.postgresql_conninfo()
    assert info.startswith("postgresql://")
    assert "octop" in info
    assert "s3cret" in info


def test_database_env_password_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "database": {
                    "driver": "postgresql",
                    "host": "localhost",
                    "database": "octop",
                    "user": "octop",
                }
            }
        )
    )
    monkeypatch.setenv("OCTOP_DATABASE_PASSWORD", "from-env")
    cfg = load_config(cfg_path)
    assert cfg.database.password == "from-env"


def test_invalid_port_falls_back(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_PORT", "not-a-number")
    cfg_path = tmp_path / "config.json"
    cfg = load_config(cfg_path)
    assert cfg.port == 8088


def test_invalid_database_driver_raises(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"database": {"driver": "mysql"}}))
    with pytest.raises(ValueError, match="database.driver"):
        load_config(cfg_path)


def test_postgresql_missing_user_raises(tmp_path: Path):
    with pytest.raises(ValueError, match="database.user"):
        parse_database_config(
            {
                "driver": "postgresql",
                "host": "localhost",
                "database": "octop",
                "user": "",
            }
        )


def test_feature_flags_env_overrides(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_ENABLE_DASHBOARD", "false")
    monkeypatch.setenv("OCTOP_ENABLE_API_DOCS", "true")
    monkeypatch.setenv("OCTOP_REQUIRE_SETUP_PASSWORD", "0")
    cfg = load_config(tmp_path / "config.json")
    assert cfg.enable_dashboard is False
    assert cfg.enable_api_docs is True
    assert cfg.require_setup_password is False


def test_feature_flags_from_file(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "enable_dashboard": False,
                "enable_api_docs": True,
                "require_setup_password": False,
            }
        )
    )
    cfg = load_config(cfg_path)
    assert cfg.enable_dashboard is False
    assert cfg.enable_api_docs is True
    assert cfg.require_setup_password is False


def test_database_section_must_be_object(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"database": "sqlite"}))
    with pytest.raises(ValueError, match="config.database must be an object"):
        load_config(cfg_path)


def test_tls_defaults(tmp_path: Path):
    cfg = load_config(tmp_path / "config.json")
    assert cfg.tls.enabled is False
    assert cfg.tls.domains == []
    assert cfg.tls.http_port == 80


def test_tls_from_file(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "tls": {
                    "enabled": True,
                    "mode": "letsencrypt",
                    "domains": ["a.example.com"],
                    "cert_file": "ssl/fullchain.pem",
                    "key_file": "ssl/privkey.pem",
                    "expires_at": "2030-01-01T00:00:00+00:00",
                }
            }
        )
    )
    cfg = load_config(cfg_path)
    assert cfg.tls.enabled is True
    assert cfg.tls.mode == "letsencrypt"
    assert cfg.tls.domains == ["a.example.com"]


def test_default_timezone_written_on_fresh_config(tmp_path: Path):
    cfg = load_config(tmp_path / "config.json")
    assert cfg.default_timezone == "Asia/Shanghai"
    written = json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
    assert written["default_timezone"] == "Asia/Shanghai"
    assert "cron_timezone" not in written


def test_default_timezone_from_file(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"default_timezone": "America/New_York"}))
    cfg = load_config(cfg_path)
    assert cfg.default_timezone == "America/New_York"


def test_legacy_cron_timezone_key_still_loads(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"cron_timezone": "Europe/Berlin"}))
    cfg = load_config(cfg_path)
    assert cfg.default_timezone == "Europe/Berlin"


def test_default_timezone_prefers_new_key_over_legacy(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "default_timezone": "UTC",
                "cron_timezone": "Europe/Berlin",
            }
        )
    )
    cfg = load_config(cfg_path)
    assert cfg.default_timezone == "UTC"


def test_default_timezone_env_override(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_DEFAULT_TIMEZONE", "Pacific/Auckland")
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"default_timezone": "UTC"}))
    cfg = load_config(cfg_path)
    assert cfg.default_timezone == "Pacific/Auckland"


def test_legacy_cron_timezone_env_still_works(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.delenv("OCTOP_DEFAULT_TIMEZONE", raising=False)
    monkeypatch.setenv("OCTOP_CRON_TIMEZONE", "Asia/Tokyo")
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(json.dumps({"default_timezone": "UTC"}))
    cfg = load_config(cfg_path)
    assert cfg.default_timezone == "Asia/Tokyo"


def test_new_timezone_env_wins_over_legacy_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    monkeypatch.setenv("OCTOP_DEFAULT_TIMEZONE", "UTC")
    monkeypatch.setenv("OCTOP_CRON_TIMEZONE", "Asia/Tokyo")
    cfg = load_config(tmp_path / "config.json")
    assert cfg.default_timezone == "UTC"


def test_loads_mobile_capabilities(tmp_path: Path) -> None:
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text(
        json.dumps(
            {
                "capabilities": {
                    "mobile": {
                        "enabled": True,
                        "backend": "physical",
                        "probed_at": "2026-01-01T00:00:00Z",
                        "reason": "",
                    }
                }
            }
        )
    )
    cfg = load_config(cfg_path)
    assert cfg.capabilities.mobile.enabled is True
    assert cfg.capabilities.mobile.backend == "physical"


def test_corrupt_config_names_file_and_position_without_leaking_contents(tmp_path: Path):
    """issue #730: a bare JSONDecodeError never says *which* config file failed."""
    cfg_path = tmp_path / "config.json"
    original = '{"database": {"password": "s3cret"},}'  # trailing comma
    cfg_path.write_text(original, encoding="utf-8")
    with pytest.raises(ValueError, match=r"line \d+, column \d+") as excinfo:
        load_config(cfg_path)
    msg = str(excinfo.value)
    assert str(cfg_path) in msg
    assert "s3cret" not in msg
    # the file is left exactly as the user wrote it
    assert cfg_path.read_text(encoding="utf-8") == original


def test_non_object_config_raises(tmp_path: Path):
    cfg_path = tmp_path / "config.json"
    cfg_path.write_text("[1, 2]", encoding="utf-8")
    with pytest.raises(ValueError, match="JSON object"):
        load_config(cfg_path)
    assert cfg_path.read_text(encoding="utf-8") == "[1, 2]"