# -*- coding: utf-8 -*-
"""Location: ./tests/unit/mcpgateway/test_postgresql_schema_config.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

Unit tests for PostgreSQL schema configuration support (Issue #1535).

Tests verify that the DATABASE_URL options parameter is correctly parsed
and applied to PostgreSQL connections for custom schema support.
"""

import pytest
from sqlalchemy import make_url


class TestPostgreSQLSchemaConfiguration:
    """Test suite for PostgreSQL schema configuration via options parameter."""

    def test_url_parsing_with_options(self):
        """Test that options parameter is correctly extracted from DATABASE_URL."""
        # Test URL with search_path option
        url_string = "postgresql://user:pass@host:5432/db?options=-c%20search_path=mcp_gateway"
        url = make_url(url_string)

        # Verify options parameter is present
        assert "options" in url.query
        assert url.query["options"] == "-c search_path=mcp_gateway"

    def test_url_parsing_without_options(self):
        """Test that URLs without options parameter work correctly."""
        url_string = "postgresql://user:pass@host:5432/db"
        url = make_url(url_string)

        # Verify no options parameter
        assert "options" not in url.query

    def test_url_parsing_multiple_schemas(self):
        """Test URL with multiple schemas in search_path."""
        url_string = "postgresql://user:pass@host:5432/db?options=-c%20search_path=mcp_gateway,public"
        url = make_url(url_string)

        assert "options" in url.query
        assert "mcp_gateway" in url.query["options"]
        assert "public" in url.query["options"]

    def test_sqlite_url_ignores_options(self):
        """Test that SQLite URLs with options don't cause errors."""
        url_string = "sqlite:///./test.db?options=-c%20search_path=test"
        url = make_url(url_string)

        # SQLite backend should be detected
        assert url.get_backend_name() == "sqlite"
        # Options may be present but should be ignored by SQLite

    def test_connect_args_generation_postgresql(self):
        """Test that connect_args includes options for PostgreSQL."""
        # Simulate the logic from db.py - must use postgresql+psycopg:// for psycopg3
        url_string = "postgresql+psycopg://user:pass@host:5432/db?options=-c%20search_path=mcp_gateway"
        url = make_url(url_string)
        backend = url.get_backend_name()
        driver = url.get_driver_name() or "default"

        connect_args = {}

        if backend == "postgresql" and driver in ("psycopg", "default", ""):
            connect_args.update(
                keepalives=1,
                keepalives_idle=30,
                keepalives_interval=5,
                keepalives_count=5,
            )

            url_options = url.query.get("options")
            if url_options:
                connect_args["options"] = url_options

        # Verify connect_args includes both keepalives and options
        assert "keepalives" in connect_args
        assert "options" in connect_args
        assert connect_args["options"] == "-c search_path=mcp_gateway"

    def test_connect_args_generation_sqlite(self):
        """Test that connect_args for SQLite doesn't include PostgreSQL options."""
        url_string = "sqlite:///./test.db"
        url = make_url(url_string)
        backend = url.get_backend_name()

        connect_args = {}

        if backend == "sqlite":
            connect_args["check_same_thread"] = False

        # Verify SQLite-specific args only
        assert "check_same_thread" in connect_args
        assert "options" not in connect_args
        assert "keepalives" not in connect_args

    def test_url_encoding_variations(self):
        """Test various URL encoding formats for options parameter."""
        test_cases = [
            # URL encoded space
            "postgresql://user:pass@host/db?options=-c%20search_path=test",
            # Plus sign for space (alternative encoding)
            "postgresql://user:pass@host/db?options=-c+search_path=test",
            # No encoding (may work in some contexts)
            "postgresql://user:pass@host/db?options=-c search_path=test",
        ]

        for url_string in test_cases:
            url = make_url(url_string)
            assert "options" in url.query
            assert "search_path" in url.query["options"]

    def test_multiple_query_parameters(self):
        """Test URL with multiple query parameters including options."""
        url_string = "postgresql://user:pass@host/db?sslmode=require&options=-c%20search_path=mcp_gateway&connect_timeout=10"
        url = make_url(url_string)

        # Verify all parameters are present
        assert "sslmode" in url.query
        assert "options" in url.query
        assert "connect_timeout" in url.query
        assert url.query["options"] == "-c search_path=mcp_gateway"

    @pytest.mark.parametrize(
        "schema_name",
        [
            "mcp_gateway",
            "custom_schema",
            "app_schema_v2",
            "schema_123",
        ],
    )
    def test_various_schema_names(self, schema_name):
        """Test that various valid schema names work correctly."""
        url_string = f"postgresql://user:pass@host/db?options=-c%20search_path={schema_name}"
        url = make_url(url_string)

        assert "options" in url.query
        assert schema_name in url.query["options"]

    def test_backward_compatibility_no_options(self):
        """Test that existing deployments without options continue to work."""
        # Standard PostgreSQL URL without options - must use postgresql+psycopg:// for psycopg3
        url_string = "postgresql+psycopg://user:pass@host:5432/db"
        url = make_url(url_string)
        backend = url.get_backend_name()
        driver = url.get_driver_name() or "default"

        connect_args = {}

        if backend == "postgresql" and driver in ("psycopg", "default", ""):
            connect_args.update(
                keepalives=1,
                keepalives_idle=30,
                keepalives_interval=5,
                keepalives_count=5,
            )

            url_options = url.query.get("options")
            if url_options:
                connect_args["options"] = url_options

        # Verify keepalives are present but options are not
        assert "keepalives" in connect_args
        assert "options" not in connect_args

    def test_complex_options_parameter(self):
        """Test options parameter with multiple PostgreSQL settings."""
        url_string = "postgresql://user:pass@host/db?options=-c%20search_path=mcp_gateway%20-c%20statement_timeout=30000"
        url = make_url(url_string)

        assert "options" in url.query
        options = url.query["options"]
        assert "search_path=mcp_gateway" in options
        assert "statement_timeout" in options


class TestSchemaConfigurationIntegration:
    """Integration tests for schema configuration (require PostgreSQL)."""

    @pytest.mark.integration
    @pytest.mark.postgresql
    def test_connection_with_custom_schema(self):
        """Test actual database connection with custom schema (requires PostgreSQL)."""
        # This test would require a real PostgreSQL instance
        # Mark as integration test to skip in unit test runs
        pytest.skip("Requires PostgreSQL instance - run with integration tests")

    @pytest.mark.integration
    @pytest.mark.postgresql
    def test_table_creation_in_custom_schema(self):
        """Test that tables are created in the specified schema (requires PostgreSQL)."""
        # This test would verify tables are created in correct schema
        pytest.skip("Requires PostgreSQL instance - run with integration tests")


if __name__ == "__main__":
    pytest.main([__file__, "-v"])
