sqlalchemy-pgview/tests/test_metadata_integration.py

"""Tests for metadata integration and auto-registration."""

from decimal import Decimal

from sqlalchemy import (
    Column,
    ForeignKey,
    Integer,
    MetaData,
    Numeric,
    String,
    Table,
    func,
    insert,
    select,
)
from sqlalchemy.engine import Engine

from sqlalchemy_pgview import (
    MaterializedView,
    View,
    get_materialized_views,
    get_views,
)


class TestAutoRegistration:
    """Tests for auto-registration of views with metadata."""

    def test_view_auto_registers_with_metadata(self) -> None:
        """Test that View is auto-registered when metadata is provided."""
        metadata = MetaData()

        # Dummy table for selectable
        users = Table("users", metadata, Column("id", Integer, primary_key=True))

        # View should be auto-registered
        user_view = View(
            "user_view",
            select(users.c.id),
            metadata=metadata,
        )

        registered = get_views(metadata)
        assert "user_view" in registered
        assert registered["user_view"] is user_view

    def test_materialized_view_auto_registers_with_metadata(self) -> None:
        """Test that MaterializedView is auto-registered when metadata is provided."""
        metadata = MetaData()

        users = Table("users", metadata, Column("id", Integer, primary_key=True))

        mv = MaterializedView(
            "user_mv",
            select(users.c.id),
            metadata=metadata,
            with_data=True,
        )

        registered = get_materialized_views(metadata)
        assert "user_mv" in registered
        assert registered["user_mv"] is mv

    def test_view_with_schema_registers_correctly(self) -> None:
        """Test that views with schema are registered with correct key."""
        metadata = MetaData()

        users = Table("users", metadata, Column("id", Integer, primary_key=True))

        view = View(
            "stats_view",
            select(users.c.id),
            schema="analytics",
            metadata=metadata,
        )

        registered = get_views(metadata)
        assert "analytics.stats_view" in registered
        assert registered["analytics.stats_view"] is view

    def test_view_without_metadata_not_registered(self) -> None:
        """Test that views without metadata are not auto-registered."""
        metadata = MetaData()

        users = Table("users", metadata, Column("id", Integer, primary_key=True))

        # No metadata provided - should not be registered
        View("orphan_view", select(users.c.id))

        registered = get_views(metadata)
        assert "orphan_view" not in registered

    def test_multiple_views_registered(self) -> None:
        """Test that multiple views can be registered."""
        metadata = MetaData()

        users = Table("users", metadata, Column("id", Integer, primary_key=True))

        View("view1", select(users.c.id), metadata=metadata)
        View("view2", select(users.c.id), metadata=metadata)
        MaterializedView("mv1", select(users.c.id), metadata=metadata)

        views = get_views(metadata)
        mviews = get_materialized_views(metadata)

        assert len(views) == 2
        assert "view1" in views
        assert "view2" in views
        assert len(mviews) == 1
        assert "mv1" in mviews


class TestMetadataCreateAll:
    """Tests for metadata.create_all() with views."""

    def test_create_all_creates_views(self, pg_engine: Engine) -> None:
        """Test that metadata.create_all() creates registered views."""
        metadata = MetaData()

        users = Table(
            "test_users_ca",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(100)),
        )

        orders = Table(
            "test_orders_ca",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("user_id", Integer, ForeignKey("test_users_ca.id")),
            Column("total", Numeric(10, 2)),
        )

        # Define view - auto-registered
        user_stats = View(
            "test_user_stats_ca",
            select(
                users.c.id,
                users.c.name,
                func.count(orders.c.id).label("order_count"),
            )
            .select_from(users.outerjoin(orders, users.c.id == orders.c.user_id))
            .group_by(users.c.id, users.c.name),
            metadata=metadata,
        )

        try:
            # Create all - should create tables AND views
            metadata.create_all(pg_engine)

            # Insert test data
            with pg_engine.begin() as conn:
                conn.execute(insert(users).values(id=1, name="Alice"))
                conn.execute(insert(users).values(id=2, name="Bob"))
                conn.execute(insert(orders).values(id=1, user_id=1, total=100))
                conn.execute(insert(orders).values(id=2, user_id=1, total=200))

            # Query the view
            with pg_engine.connect() as conn:
                result = conn.execute(
                    select(user_stats.as_table()).order_by(user_stats.as_table().c.name)
                ).fetchall()

                assert len(result) == 2
                assert result[0].name == "Alice"
                assert result[0].order_count == 2
                assert result[1].name == "Bob"
                assert result[1].order_count == 0

        finally:
            metadata.drop_all(pg_engine)

    def test_create_all_creates_materialized_views(self, pg_engine: Engine) -> None:
        """Test that metadata.create_all() creates materialized views."""
        metadata = MetaData()

        orders = Table(
            "test_orders_mv",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("total", Numeric(10, 2)),
        )

        # Define materialized view - auto-registered
        order_summary = MaterializedView(
            "test_order_summary_mv",
            select(
                func.count(orders.c.id).label("order_count"),
                func.sum(orders.c.total).label("total_revenue"),
            ),
            metadata=metadata,
            with_data=True,
        )

        try:
            metadata.create_all(pg_engine)

            # Insert test data
            with pg_engine.begin() as conn:
                conn.execute(insert(orders).values(id=1, total=100))
                conn.execute(insert(orders).values(id=2, total=200))

            # Query the materialized view (shows data at creation time = 0)
            with pg_engine.connect() as conn:
                result = conn.execute(select(order_summary.as_table())).fetchone()
                # MV was created before data, so shows 0
                assert result.order_count == 0

                # Refresh to see actual data
                order_summary.refresh(conn)

                result = conn.execute(select(order_summary.as_table())).fetchone()
                assert result.order_count == 2
                assert result.total_revenue == Decimal("300")

        finally:
            metadata.drop_all(pg_engine)

    def test_drop_all_drops_views(self, pg_engine: Engine) -> None:
        """Test that metadata.drop_all() drops registered views."""
        from sqlalchemy import text

        metadata = MetaData()

        users = Table(
            "test_users_drop",
            metadata,
            Column("id", Integer, primary_key=True),
        )

        View(
            "test_view_drop",
            select(users.c.id),
            metadata=metadata,
        )

        MaterializedView(
            "test_mv_drop",
            select(users.c.id),
            metadata=metadata,
        )

        # Create all
        metadata.create_all(pg_engine)

        # Verify views exist
        with pg_engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT COUNT(*) FROM pg_class WHERE relname IN ('test_view_drop', 'test_mv_drop')"
                )
            ).scalar()
            assert result == 2

        # Drop all
        metadata.drop_all(pg_engine)

        # Verify views are gone
        with pg_engine.connect() as conn:
            result = conn.execute(
                text(
                    "SELECT COUNT(*) FROM pg_class WHERE relname IN ('test_view_drop', 'test_mv_drop')"
                )
            ).scalar()
            assert result == 0

    def test_views_created_after_tables(self, pg_engine: Engine) -> None:
        """Test that views are created after tables (dependencies work)."""
        metadata = MetaData()

        # Table
        users = Table(
            "test_users_order",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(100)),
        )

        # View that depends on table
        View(
            "test_users_view_order",
            select(users.c.id, users.c.name),
            metadata=metadata,
        )

        # This should not raise - tables created before views
        metadata.create_all(pg_engine)
        metadata.drop_all(pg_engine)