test_settings.py

"""Tests for DataJoint settings module."""

from pathlib import Path

import pytest
from pydantic import SecretStr, ValidationError

import datajoint as dj
from datajoint import settings
from datajoint.errors import DataJointError
from datajoint.settings import (
    CONFIG_FILENAME,
    SECRETS_DIRNAME,
    find_config_file,
    find_secrets_dir,
    read_secret_file,
)


class TestConfigFileSearch:
    """Test recursive config file search."""

    def test_find_in_current_directory(self, tmp_path):
        """Config file in current directory is found."""
        config_file = tmp_path / CONFIG_FILENAME
        config_file.write_text("{}")

        found = find_config_file(tmp_path)
        assert found == config_file

    def test_find_in_parent_directory(self, tmp_path):
        """Config file in parent directory is found."""
        subdir = tmp_path / "src" / "pipeline"
        subdir.mkdir(parents=True)
        config_file = tmp_path / CONFIG_FILENAME
        config_file.write_text("{}")

        found = find_config_file(subdir)
        assert found == config_file

    def test_stop_at_git_boundary(self, tmp_path):
        """Search stops at .git directory."""
        (tmp_path / ".git").mkdir()
        subdir = tmp_path / "src"
        subdir.mkdir()
        # No config file - should return None, not search above .git

        found = find_config_file(subdir)
        assert found is None

    def test_stop_at_hg_boundary(self, tmp_path):
        """Search stops at .hg directory."""
        (tmp_path / ".hg").mkdir()
        subdir = tmp_path / "src"
        subdir.mkdir()

        found = find_config_file(subdir)
        assert found is None

    def test_config_found_before_git(self, tmp_path):
        """Config file found before reaching .git boundary."""
        (tmp_path / ".git").mkdir()
        config_file = tmp_path / CONFIG_FILENAME
        config_file.write_text("{}")
        subdir = tmp_path / "src"
        subdir.mkdir()

        found = find_config_file(subdir)
        assert found == config_file

    def test_returns_none_when_not_found(self, tmp_path):
        """Returns None when no config file exists."""
        (tmp_path / ".git").mkdir()  # Create boundary
        subdir = tmp_path / "src"
        subdir.mkdir()

        found = find_config_file(subdir)
        assert found is None


class TestSecretsDirectory:
    """Test secrets directory detection and loading."""

    def test_find_secrets_next_to_config(self, tmp_path):
        """Finds .secrets/ directory next to config file."""
        config_file = tmp_path / CONFIG_FILENAME
        config_file.write_text("{}")
        secrets_dir = tmp_path / SECRETS_DIRNAME
        secrets_dir.mkdir()

        found = find_secrets_dir(config_file)
        assert found == secrets_dir

    def test_no_secrets_dir_returns_none(self, tmp_path):
        """Returns None when no secrets directory exists."""
        config_file = tmp_path / CONFIG_FILENAME
        config_file.write_text("{}")

        found = find_secrets_dir(config_file)
        # May return system secrets dir if it exists, otherwise None
        if found is not None:
            assert found == settings.SYSTEM_SECRETS_DIR

    def test_read_secret_file(self, tmp_path):
        """Reads secret value from file."""
        (tmp_path / "database.password").write_text("my_secret\n")

        value = read_secret_file(tmp_path, "database.password")
        assert value == "my_secret"  # Strips whitespace

    def test_read_missing_secret_returns_none(self, tmp_path):
        """Returns None for missing secret file."""
        value = read_secret_file(tmp_path, "nonexistent")
        assert value is None

    def test_read_secret_from_none_dir(self):
        """Returns None when secrets_dir is None."""
        value = read_secret_file(None, "database.password")
        assert value is None


class TestSecretStr:
    """Test SecretStr handling for sensitive fields."""

    def test_password_is_secret_str(self):
        """Password field uses SecretStr type."""
        dj.config.database.password = "test_password"
        assert isinstance(dj.config.database.password, SecretStr)
        dj.config.database.password = None

    def test_secret_str_masked_in_repr(self):
        """SecretStr values are masked in repr."""
        dj.config.database.password = "super_secret"
        repr_str = repr(dj.config.database.password)
        assert "super_secret" not in repr_str
        assert "**" in repr_str
        dj.config.database.password = None

    def test_dict_access_unwraps_secret(self):
        """Dict-style access returns plain string for secrets."""
        dj.config.database.password = "unwrapped_secret"
        value = dj.config["database.password"]
        assert value == "unwrapped_secret"
        assert isinstance(value, str)
        assert not isinstance(value, SecretStr)
        dj.config.database.password = None

    def test_store_secret_key_is_secret_str(self):
        """Store secret key uses SecretStr type when set."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {"secret_key": "aws_secret"}
            # SecretStr is handled by pydantic if defined, but stores dict doesn't enforce it
            assert dj.config.stores["test_store"]["secret_key"] == "aws_secret"
        finally:
            dj.config.stores = original_stores


class TestSettingsAccess:
    """Test accessing settings via different methods."""

    def test_attribute_access(self):
        """Test accessing settings via attributes."""
        # Host can be localhost or db (docker), just verify it's a string
        assert isinstance(dj.config.database.host, str)
        assert len(dj.config.database.host) > 0
        # Port may be 3306 (default) or a random port (testcontainers)
        assert isinstance(dj.config.database.port, int)
        assert 1 <= dj.config.database.port <= 65535
        # safemode may be modified by conftest fixtures
        assert isinstance(dj.config.safemode, bool)

    def test_dict_style_access(self):
        """Test accessing settings via dict-style notation."""
        # Host can be localhost or db (docker), just verify it's a string
        assert isinstance(dj.config["database.host"], str)
        assert len(dj.config["database.host"]) > 0
        # Port may be 3306 (default) or a random port (testcontainers)
        assert isinstance(dj.config["database.port"], int)
        assert 1 <= dj.config["database.port"] <= 65535
        # safemode may be modified by conftest fixtures
        assert isinstance(dj.config["safemode"], bool)

    def test_get_with_default(self):
        """Test get() method with default values."""
        # Host can be localhost or db (docker), just verify it exists
        assert dj.config.get("database.host") is not None
        assert dj.config.get("nonexistent.key", "default") == "default"
        assert dj.config.get("nonexistent.key") is None


class TestSettingsModification:
    """Test modifying settings."""

    def test_attribute_assignment(self):
        """Test setting values via attribute assignment."""
        original = dj.config.database.host
        try:
            dj.config.database.host = "testhost"
            assert dj.config.database.host == "testhost"
        finally:
            dj.config.database.host = original

    def test_dict_style_assignment(self):
        """Test setting values via dict-style notation."""
        original = dj.config["database.host"]
        try:
            dj.config["database.host"] = "testhost2"
            assert dj.config["database.host"] == "testhost2"
        finally:
            dj.config["database.host"] = original


class TestTypeValidation:
    """Test pydantic type validation."""

    def test_port_must_be_integer(self):
        """Test that port must be an integer."""
        with pytest.raises(ValidationError):
            dj.config.database.port = "not_an_integer"

    def test_loglevel_validation(self):
        """Test that loglevel must be a valid level."""
        with pytest.raises(ValidationError):
            dj.config.loglevel = "INVALID_LEVEL"

    def test_fetch_format_validation(self):
        """Test that fetch_format must be array or frame."""
        with pytest.raises(ValidationError):
            dj.config.fetch_format = "invalid"


class TestContextManager:
    """Test the override context manager."""

    def test_override_simple_value(self):
        """Test overriding a simple value."""
        original = dj.config.safemode
        with dj.config.override(safemode=False):
            assert dj.config.safemode is False
        assert dj.config.safemode == original

    def test_override_nested_value(self):
        """Test overriding nested values with double underscore."""
        original = dj.config.database.host
        with dj.config.override(database__host="override_host"):
            assert dj.config.database.host == "override_host"
        assert dj.config.database.host == original

    def test_override_restores_on_exception(self):
        """Test that override restores values even when exception occurs."""
        original = dj.config.safemode
        try:
            with dj.config.override(safemode=False):
                assert dj.config.safemode is False
                raise ValueError("test exception")
        except ValueError:
            pass
        assert dj.config.safemode == original


class TestLoad:
    """Test loading configuration."""

    def test_load_config_file(self, tmp_path, monkeypatch):
        """Test loading configuration from file.

        Note: Environment variables take precedence over config file values.
        We need to clear DJ_HOST to test file loading.
        """
        filename = tmp_path / "test_config.json"
        filename.write_text('{"database": {"host": "loaded_host"}}')
        original_host = dj.config.database.host

        # Clear env var so file value takes effect
        monkeypatch.delenv("DJ_HOST", raising=False)

        try:
            dj.config.load(filename)
            assert dj.config.database.host == "loaded_host"
        finally:
            dj.config.database.host = original_host

    def test_env_var_overrides_config_file(self, tmp_path, monkeypatch):
        """Test that environment variables take precedence over config file.

        When DJ_HOST is set, loading a config file should NOT override the value.
        The env var value should be preserved.
        """
        filename = tmp_path / "test_config.json"
        filename.write_text('{"database": {"host": "file_host"}}')
        original_host = dj.config.database.host

        # Set env var - it should take precedence over file
        monkeypatch.setenv("DJ_HOST", "env_host")
        # Reset config to pick up new env var
        dj.config.database.host = "env_host"

        try:
            dj.config.load(filename)
            # File value should be skipped because DJ_HOST is set
            # The env var value should be preserved
            assert dj.config.database.host == "env_host"
        finally:
            dj.config.database.host = original_host

    def test_load_nonexistent_file(self):
        """Test loading nonexistent file raises FileNotFoundError."""
        with pytest.raises(FileNotFoundError):
            dj.config.load("/nonexistent/path/config.json")


class TestStoreSpec:
    """Test external store configuration."""

    def test_get_store_spec_not_configured(self):
        """Test getting unconfigured store raises error."""
        with pytest.raises(DataJointError, match="not configured"):
            dj.config.get_store_spec("nonexistent_store")

    def test_get_store_spec_file_protocol(self):
        """Test file protocol store spec validation."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_file"] = {
                "protocol": "file",
                "location": "/tmp/test",
            }
            spec = dj.config.get_store_spec("test_file")
            assert spec["protocol"] == "file"
            assert spec["location"] == "/tmp/test"
            # Default is now None (no subfolding) instead of DEFAULT_SUBFOLDING
            assert spec["subfolding"] is None
            assert spec["partition_pattern"] is None
            assert spec["token_length"] == 8
        finally:
            dj.config.stores = original_stores

    def test_get_store_spec_missing_required(self):
        """Test missing required keys raises error."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["bad_store"] = {
                "protocol": "file",
                # missing location
            }
            with pytest.raises(DataJointError, match="missing"):
                dj.config.get_store_spec("bad_store")
        finally:
            dj.config.stores = original_stores

    def test_get_store_spec_default_store(self):
        """Test getting default store when store=None."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["default"] = "my_default"
            dj.config.stores["my_default"] = {
                "protocol": "file",
                "location": "/tmp/default",
            }
            # Calling with None should use stores.default
            spec = dj.config.get_store_spec(None)
            assert spec["protocol"] == "file"
            assert spec["location"] == "/tmp/default"
        finally:
            dj.config.stores = original_stores

    def test_get_store_spec_no_default_configured(self):
        """Test error when stores.default is not configured."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores = {}  # Clear stores
            with pytest.raises(DataJointError, match="stores.default is not configured"):
                dj.config.get_store_spec(None)
        finally:
            dj.config.stores = original_stores

    def test_get_store_spec_filepath_default(self):
        """Test filepath_default for filepath references (not part of OAS)."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["default"] = "integrated"
            dj.config.stores["filepath_default"] = "raw_data"
            dj.config.stores["integrated"] = {
                "protocol": "s3",
                "endpoint": "s3.amazonaws.com",
                "bucket": "my-bucket",
                "location": "processed",
                "access_key": "xxx",
                "secret_key": "yyy",
            }
            dj.config.stores["raw_data"] = {
                "protocol": "file",
                "location": "/data/acquisition",
            }

            # Regular default for integrated storage
            spec = dj.config.get_store_spec(None, use_filepath_default=False)
            assert spec["protocol"] == "s3"
            assert spec["location"] == "processed"

            # Filepath default for filepath references
            spec = dj.config.get_store_spec(None, use_filepath_default=True)
            assert spec["protocol"] == "file"
            assert spec["location"] == "/data/acquisition"
        finally:
            dj.config.stores = original_stores

    def test_get_store_spec_no_filepath_default(self):
        """Test error when filepath_default not configured but requested."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["default"] = "integrated"
            dj.config.stores["integrated"] = {
                "protocol": "file",
                "location": "/data/store",
            }
            # No filepath_default configured

            with pytest.raises(DataJointError, match="stores.filepath_default is not configured"):
                dj.config.get_store_spec(None, use_filepath_default=True)
        finally:
            dj.config.stores = original_stores

    def test_get_store_spec_explicit_store_ignores_defaults(self):
        """Test that explicit store name bypasses both defaults."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["default"] = "store_a"
            dj.config.stores["filepath_default"] = "store_b"
            dj.config.stores["store_a"] = {"protocol": "file", "location": "/a"}
            dj.config.stores["store_b"] = {"protocol": "file", "location": "/b"}
            dj.config.stores["store_c"] = {"protocol": "file", "location": "/c"}

            # Explicitly naming store_c should work regardless of use_filepath_default
            spec = dj.config.get_store_spec("store_c", use_filepath_default=False)
            assert spec["location"] == "/c"

            spec = dj.config.get_store_spec("store_c", use_filepath_default=True)
            assert spec["location"] == "/c"
        finally:
            dj.config.stores = original_stores


class TestStoreSecrets:
    """Test loading store credentials from secrets directory."""

    def test_load_store_credentials_from_secrets(self, tmp_path):
        """Test loading per-store credentials from .secrets/ directory."""
        # Create secrets directory with store credentials
        secrets_dir = tmp_path / SECRETS_DIRNAME
        secrets_dir.mkdir()
        (secrets_dir / "stores.main.access_key").write_text("test_access_key")
        (secrets_dir / "stores.main.secret_key").write_text("test_secret_key")

        # Create a fresh config instance
        cfg = settings.Config()
        original_stores = cfg.stores.copy()
        try:
            # Load secrets
            cfg._load_secrets(secrets_dir)

            # Verify credentials were loaded
            assert "main" in cfg.stores
            assert cfg.stores["main"]["access_key"] == "test_access_key"
            assert cfg.stores["main"]["secret_key"] == "test_secret_key"
        finally:
            cfg.stores = original_stores

    def test_secrets_do_not_override_existing(self, tmp_path):
        """Test that secrets don't override already-configured store settings."""
        secrets_dir = tmp_path / SECRETS_DIRNAME
        secrets_dir.mkdir()
        (secrets_dir / "stores.main.access_key").write_text("secret_key")

        cfg = settings.Config()
        original_stores = cfg.stores.copy()
        try:
            # Pre-configure the store with a key
            cfg.stores["main"] = {"access_key": "existing_key"}

            # Load secrets - should not override
            cfg._load_secrets(secrets_dir)

            # Existing key should be preserved
            assert cfg.stores["main"]["access_key"] == "existing_key"
        finally:
            cfg.stores = original_stores


class TestDisplaySettings:
    """Test display-related settings."""

    def test_display_limit(self):
        """Test display limit setting."""
        original = dj.config.display.limit
        try:
            dj.config.display.limit = 50
            assert dj.config.display.limit == 50
        finally:
            dj.config.display.limit = original


class TestCachePaths:
    """Test cache path settings."""

    def test_query_cache_path_string(self):
        """Test setting query_cache path as string."""
        original = dj.config.query_cache
        try:
            dj.config.query_cache = "/tmp/cache"
            assert dj.config.query_cache == Path("/tmp/cache")
        finally:
            dj.config.query_cache = original

    def test_query_cache_path_none(self):
        """Test query_cache path can be None."""
        original = dj.config.query_cache
        try:
            dj.config.query_cache = None
            assert dj.config.query_cache is None
        finally:
            dj.config.query_cache = original


class TestSaveTemplate:
    """Test save_template method for creating configuration templates."""

    def test_save_minimal_template(self, tmp_path):
        """Test creating a minimal template."""
        config_path = tmp_path / "datajoint.json"
        result = dj.config.save_template(config_path, minimal=True, create_secrets_dir=False)

        assert result == config_path.absolute()
        assert config_path.exists()

        import json

        with open(config_path) as f:
            content = json.load(f)

        assert "database" in content
        assert content["database"]["host"] == "localhost"
        assert content["database"]["port"] == 3306
        # Minimal template should not have credentials
        assert "password" not in content["database"]
        assert "user" not in content["database"]

    def test_save_full_template(self, tmp_path):
        """Test creating a full template."""
        config_path = tmp_path / "datajoint.json"
        result = dj.config.save_template(config_path, minimal=False, create_secrets_dir=False)

        assert result == config_path.absolute()
        assert config_path.exists()

        import json

        with open(config_path) as f:
            content = json.load(f)

        # Full template should have all settings groups
        assert "database" in content
        assert "connection" in content
        assert "display" in content
        assert "stores" in content
        assert "loglevel" in content
        assert "safemode" in content
        # Verify stores structure
        assert "default" in content["stores"]
        assert "main" in content["stores"]
        assert content["stores"]["default"] == "main"
        assert content["stores"]["main"]["protocol"] == "file"
        # But still no credentials
        assert "password" not in content["database"]
        assert "user" not in content["database"]

    def test_save_template_creates_secrets_dir(self, tmp_path):
        """Test that save_template creates .secrets/ directory."""
        config_path = tmp_path / "datajoint.json"
        dj.config.save_template(config_path, create_secrets_dir=True)

        secrets_dir = tmp_path / SECRETS_DIRNAME
        assert secrets_dir.exists()
        assert secrets_dir.is_dir()

        # Check placeholder files created
        assert (secrets_dir / "database.user").exists()
        assert (secrets_dir / "database.password").exists()

        # Check .gitignore created
        gitignore = secrets_dir / ".gitignore"
        assert gitignore.exists()
        assert "*" in gitignore.read_text()

    def test_save_template_refuses_overwrite(self, tmp_path):
        """Test that save_template won't overwrite existing file."""
        config_path = tmp_path / "datajoint.json"
        config_path.write_text("{}")

        with pytest.raises(FileExistsError, match="already exists"):
            dj.config.save_template(config_path)

    def test_save_template_secrets_dir_idempotent(self, tmp_path):
        """Test that creating secrets dir doesn't overwrite existing secrets."""
        config_path = tmp_path / "datajoint.json"
        secrets_dir = tmp_path / SECRETS_DIRNAME
        secrets_dir.mkdir()

        # Pre-populate a secret
        password_file = secrets_dir / "database.password"
        password_file.write_text("existing_password")

        dj.config.save_template(config_path, create_secrets_dir=True)

        # Original password should be preserved
        assert password_file.read_text() == "existing_password"


class TestStorePrefixes:
    """Tests for storage section prefix configuration and validation."""

    def test_default_prefixes(self):
        """Test that default prefixes are set correctly."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
            }

            spec = dj.config.get_store_spec("test_store")
            assert spec["hash_prefix"] == "_hash"
            assert spec["schema_prefix"] == "_schema"
            assert spec["filepath_prefix"] is None
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)

    def test_custom_prefixes(self):
        """Test configuring custom prefixes."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
                "hash_prefix": "content_addressed",
                "schema_prefix": "structured_data",
                "filepath_prefix": "user_files",
            }

            spec = dj.config.get_store_spec("test_store")
            assert spec["hash_prefix"] == "content_addressed"
            assert spec["schema_prefix"] == "structured_data"
            assert spec["filepath_prefix"] == "user_files"
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)

    def test_prefix_overlap_hash_and_schema(self):
        """Test that overlapping hash and schema prefixes are rejected."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
                "hash_prefix": "managed",
                "schema_prefix": "managed/schema",  # Nested under hash
            }

            with pytest.raises(DataJointError, match=r"overlap.*mutually exclusive"):
                dj.config.get_store_spec("test_store")
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)

    def test_prefix_overlap_schema_and_filepath(self):
        """Test that overlapping schema and filepath prefixes are rejected."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
                "schema_prefix": "data",
                "filepath_prefix": "data/files",  # Nested under schema
            }

            with pytest.raises(DataJointError, match=r"overlap.*mutually exclusive"):
                dj.config.get_store_spec("test_store")
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)

    def test_prefix_overlap_reverse_nesting(self):
        """Test that parent-child relationship is detected in either direction."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
                "hash_prefix": "dj/managed/hash",  # Child
                "schema_prefix": "dj/managed",  # Parent
            }

            with pytest.raises(DataJointError, match=r"overlap.*mutually exclusive"):
                dj.config.get_store_spec("test_store")
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)

    def test_non_overlapping_prefixes_accepted(self):
        """Test that non-overlapping prefixes are accepted."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
                "hash_prefix": "hash_store",
                "schema_prefix": "schema_store",
                "filepath_prefix": "user_files",
            }

            # Should not raise
            spec = dj.config.get_store_spec("test_store")
            assert spec["hash_prefix"] == "hash_store"
            assert spec["schema_prefix"] == "schema_store"
            assert spec["filepath_prefix"] == "user_files"
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)

    def test_similar_prefix_names_allowed(self):
        """Test that prefixes with similar names but no nesting are allowed."""
        original_stores = dj.config.stores.copy()
        try:
            dj.config.stores["test_store"] = {
                "protocol": "file",
                "location": "/tmp/test",
                "hash_prefix": "managed_hash",
                "schema_prefix": "managed_schema",  # Similar name, but separate
                "filepath_prefix": None,
            }

            # Should not raise - these are separate paths
            spec = dj.config.get_store_spec("test_store")
            assert spec["hash_prefix"] == "managed_hash"
            assert spec["schema_prefix"] == "managed_schema"
        finally:
            dj.config.stores.clear()
            dj.config.stores.update(original_stores)


class TestDatabaseNameConfiguration:
    """Test database.name configuration."""

    def test_database_name_default_is_none(self):
        """Database name defaults to None when not configured."""
        from datajoint.settings import DatabaseSettings

        s = DatabaseSettings()
        assert s.name is None

    def test_database_name_env_var(self, monkeypatch):
        """DJ_DATABASE_NAME environment variable sets database name."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_DATABASE_NAME", "my_database")
        s = DatabaseSettings()
        assert s.name == "my_database"

    def test_database_name_from_config_file(self, tmp_path, monkeypatch):
        """Load database name from config file."""
        import json

        from datajoint.settings import Config

        config_file = tmp_path / "test_config.json"
        config_file.write_text(json.dumps({"database": {"name": "custom_db", "host": "localhost"}}))

        monkeypatch.delenv("DJ_DATABASE_NAME", raising=False)
        monkeypatch.delenv("DJ_HOST", raising=False)

        cfg = Config()
        cfg.load(config_file)
        assert cfg.database.name == "custom_db"

    def test_database_name_dict_access(self):
        """Dict-style access reads and writes database name."""
        original = dj.config.database.name
        try:
            dj.config.database.name = "test_db"
            assert dj.config["database.name"] == "test_db"
        finally:
            dj.config.database.name = original

    def test_database_name_override_context_manager(self):
        """Override context manager temporarily sets database name."""
        original = dj.config.database.name
        with dj.config.override(database__name="override_db"):
            assert dj.config.database.name == "override_db"
        assert dj.config.database.name == original

    def test_database_prefix_empty_no_warning(self):
        """Empty database_prefix does not emit DeprecationWarning at config load."""
        import warnings

        from datajoint.settings import DatabaseSettings

        with warnings.catch_warnings(record=True) as w:
            warnings.simplefilter("always")
            DatabaseSettings()
            deprecation_warnings = [
                x for x in w if issubclass(x.category, DeprecationWarning) and "database_prefix" in str(x.message)
            ]
            assert len(deprecation_warnings) == 0


class TestBackendConfiguration:
    """Test database backend configuration and port auto-detection."""

    def test_backend_default(self):
        """Test default backend is mysql."""
        from datajoint.settings import DatabaseSettings

        settings = DatabaseSettings()
        assert settings.backend == "mysql"
        assert settings.port == 3306

    def test_backend_postgresql(self, monkeypatch):
        """Test PostgreSQL backend with auto port."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_BACKEND", "postgresql")
        settings = DatabaseSettings()
        assert settings.backend == "postgresql"
        assert settings.port == 5432

    def test_backend_explicit_port_overrides(self, monkeypatch):
        """Test explicit port overrides auto-detection."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_BACKEND", "postgresql")
        monkeypatch.setenv("DJ_PORT", "9999")
        settings = DatabaseSettings()
        assert settings.backend == "postgresql"
        assert settings.port == 9999

    def test_backend_env_var(self, monkeypatch):
        """Test DJ_BACKEND environment variable."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_BACKEND", "postgresql")
        settings = DatabaseSettings()
        assert settings.backend == "postgresql"
        assert settings.port == 5432

    def test_port_env_var_overrides_backend_default(self, monkeypatch):
        """Test DJ_PORT overrides backend auto-detection."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_BACKEND", "postgresql")
        monkeypatch.setenv("DJ_PORT", "8888")
        settings = DatabaseSettings()
        assert settings.backend == "postgresql"
        assert settings.port == 8888

    def test_invalid_backend(self, monkeypatch):
        """Test invalid backend raises validation error."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_BACKEND", "sqlite")
        with pytest.raises(ValidationError, match="Input should be 'mysql' or 'postgresql'"):
            DatabaseSettings()

    def test_config_file_backend(self, tmp_path, monkeypatch):
        """Test loading backend from config file."""
        import json

        from datajoint.settings import Config

        # Include port in config since auto-detection only happens during initialization
        config_file = tmp_path / "test_config.json"
        config_file.write_text(json.dumps({"database": {"backend": "postgresql", "host": "db.example.com", "port": 5432}}))

        # Clear env vars so file values take effect
        monkeypatch.delenv("DJ_BACKEND", raising=False)
        monkeypatch.delenv("DJ_HOST", raising=False)
        monkeypatch.delenv("DJ_PORT", raising=False)

        cfg = Config()
        cfg.load(config_file)
        assert cfg.database.backend == "postgresql"
        assert cfg.database.port == 5432
        assert cfg.database.host == "db.example.com"

    def test_global_config_backend(self):
        """Test global config has backend configuration."""
        # Global config should have backend field with default mysql
        assert hasattr(dj.config.database, "backend")
        # Backend should be one of the valid values
        assert dj.config.database.backend in ["mysql", "postgresql"]
        # Port should be set (either 3306 or 5432 or custom)
        assert isinstance(dj.config.database.port, int)
        assert 1 <= dj.config.database.port <= 65535

    def test_port_auto_detection_on_initialization(self):
        """Test port auto-detects only during initialization, not on live updates."""
        from datajoint.settings import DatabaseSettings

        # Start with MySQL (default)
        settings = DatabaseSettings()
        assert settings.port == 3306

        # Change backend on live config - port won't auto-update
        settings.backend = "postgresql"
        # Port remains at previous value (this is expected behavior)
        # Users should set port explicitly when changing backend on live config
        assert settings.port == 3306  # Didn't auto-update

    def test_mysql_backend_with_explicit_port(self, monkeypatch):
        """Test MySQL backend with explicit non-default port."""
        from datajoint.settings import DatabaseSettings

        monkeypatch.setenv("DJ_BACKEND", "mysql")
        monkeypatch.setenv("DJ_PORT", "3307")
        settings = DatabaseSettings()
        assert settings.backend == "mysql"
        assert settings.port == 3307

    def test_backend_field_in_env_var_mapping(self):
        """Test that backend is mapped to DJ_BACKEND in ENV_VAR_MAPPING."""
        from datajoint.settings import ENV_VAR_MAPPING

        assert "database.backend" in ENV_VAR_MAPPING
        assert ENV_VAR_MAPPING["database.backend"] == "DJ_BACKEND"