sqlalchemy-pgview/tests/test_ddl.py

"""Tests for DDL operations."""

from sqlalchemy import select
from sqlalchemy.engine import Engine

from sqlalchemy_pgview import (
    CreateMaterializedView,
    CreateView,
    DropMaterializedView,
    DropView,
    MaterializedView,
    RefreshMaterializedView,
    View,
)


class TestCreateView:
    """Tests for CreateView DDL."""

    def test_create_view_sql(self, pg_engine: Engine, sample_tables: tuple) -> None:
        """Test CREATE VIEW SQL generation."""
        users, _ = sample_tables
        view = View(
            "active_users",
            select(users.c.id, users.c.name),
        )

        stmt = CreateView(view)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "CREATE VIEW active_users AS" in sql
        assert "users.id" in sql
        assert "users.name" in sql

    def test_create_or_replace_view_sql(
        self, pg_engine: Engine, sample_tables: tuple
    ) -> None:
        """Test CREATE OR REPLACE VIEW SQL generation."""
        users, _ = sample_tables
        view = View(
            "active_users",
            select(users.c.id),
        )

        stmt = CreateView(view, or_replace=True)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "CREATE OR REPLACE VIEW" in sql

    def test_create_view_with_schema(
        self, pg_engine: Engine, sample_tables: tuple
    ) -> None:
        """Test CREATE VIEW with schema."""
        users, _ = sample_tables
        view = View(
            "active_users",
            select(users.c.id),
            schema="analytics",
        )

        stmt = CreateView(view)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "analytics.active_users" in sql


class TestDropView:
    """Tests for DropView DDL."""

    def test_drop_view_sql(self, pg_engine: Engine) -> None:
        """Test DROP VIEW SQL generation."""
        stmt = DropView("test_view")
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert sql == "DROP VIEW test_view"

    def test_drop_view_if_exists(self, pg_engine: Engine) -> None:
        """Test DROP VIEW IF EXISTS."""
        stmt = DropView("test_view", if_exists=True)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert sql == "DROP VIEW IF EXISTS test_view"

    def test_drop_view_cascade(self, pg_engine: Engine) -> None:
        """Test DROP VIEW CASCADE."""
        stmt = DropView("test_view", cascade=True)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert sql == "DROP VIEW test_view CASCADE"

    def test_drop_view_with_schema(self, pg_engine: Engine) -> None:
        """Test DROP VIEW with schema."""
        stmt = DropView("test_view", schema="analytics", if_exists=True)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert sql == "DROP VIEW IF EXISTS analytics.test_view"


class TestCreateMaterializedView:
    """Tests for CreateMaterializedView DDL."""

    def test_create_materialized_view_sql(
        self, pg_engine: Engine, sample_tables: tuple
    ) -> None:
        """Test CREATE MATERIALIZED VIEW SQL generation."""
        users, _ = sample_tables
        mview = MaterializedView(
            "user_cache",
            select(users.c.id, users.c.name),
            with_data=True,
        )

        stmt = CreateMaterializedView(mview)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "CREATE MATERIALIZED VIEW user_cache AS" in sql
        assert "WITH DATA" in sql

    def test_create_materialized_view_without_data(
        self, pg_engine: Engine, sample_tables: tuple
    ) -> None:
        """Test CREATE MATERIALIZED VIEW WITH NO DATA."""
        users, _ = sample_tables
        mview = MaterializedView(
            "user_cache",
            select(users.c.id),
            with_data=False,
        )

        stmt = CreateMaterializedView(mview)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "WITH NO DATA" in sql


class TestDropMaterializedView:
    """Tests for DropMaterializedView DDL."""

    def test_drop_materialized_view_sql(self, pg_engine: Engine) -> None:
        """Test DROP MATERIALIZED VIEW SQL generation."""
        stmt = DropMaterializedView("test_mview", if_exists=True, cascade=True)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert sql == "DROP MATERIALIZED VIEW IF EXISTS test_mview CASCADE"


class TestRefreshMaterializedView:
    """Tests for RefreshMaterializedView DDL."""

    def test_refresh_materialized_view_sql(self, pg_engine: Engine) -> None:
        """Test REFRESH MATERIALIZED VIEW SQL generation."""
        stmt = RefreshMaterializedView("test_mview")
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert sql == "REFRESH MATERIALIZED VIEW test_mview WITH DATA"

    def test_refresh_concurrently(self, pg_engine: Engine) -> None:
        """Test REFRESH MATERIALIZED VIEW CONCURRENTLY."""
        stmt = RefreshMaterializedView("test_mview", concurrently=True)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "CONCURRENTLY" in sql

    def test_refresh_without_data(self, pg_engine: Engine) -> None:
        """Test REFRESH MATERIALIZED VIEW WITH NO DATA."""
        stmt = RefreshMaterializedView("test_mview", with_data=False)
        compiled = stmt.compile(dialect=pg_engine.dialect)
        sql = str(compiled)

        assert "WITH NO DATA" in sql