sqlalchemy-pgview/tests/test_dependencies.py

"""Tests for view dependency tracking."""

from sqlalchemy import Table, func, select
from sqlalchemy.engine import Engine

from sqlalchemy_pgview import (
    CreateMaterializedView,
    CreateView,
    DropMaterializedView,
    DropView,
    MaterializedView,
    View,
    get_all_views,
    get_view_definition,
)
from sqlalchemy_pgview.dependencies import (
    ViewDependency,
    ViewInfo,
    get_dependency_order,
    get_reverse_dependencies,
    get_view_dependencies,
)


class TestViewInfo:
    """Tests for ViewInfo dataclass."""

    def test_view_info_fullname_without_schema(self) -> None:
        """Test ViewInfo.fullname without schema."""
        info = ViewInfo(
            name="my_view",
            schema=None,
            definition="SELECT 1",
            is_materialized=False,
        )
        assert info.fullname == "my_view"

    def test_view_info_fullname_with_schema(self) -> None:
        """Test ViewInfo.fullname with schema."""
        info = ViewInfo(
            name="my_view",
            schema="analytics",
            definition="SELECT 1",
            is_materialized=False,
        )
        assert info.fullname == "analytics.my_view"


class TestViewDependency:
    """Tests for ViewDependency dataclass."""

    def test_view_dependency_fullnames(self) -> None:
        """Test ViewDependency fullname properties."""
        dep = ViewDependency(
            dependent_view="child_view",
            dependent_schema="public",
            referenced_view="parent_view",
            referenced_schema="analytics",
        )
        assert dep.dependent_fullname == "public.child_view"
        assert dep.referenced_fullname == "analytics.parent_view"


class TestGetAllViews:
    """Tests for get_all_views function."""

    def test_get_all_views(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting all views from database."""
        authors = pg_one_to_many_tables["authors"]
        books = pg_one_to_many_tables["books"]

        view1 = View(
            "test_view_deps_1",
            select(authors.c.id, authors.c.name),
        )

        mview1 = MaterializedView(
            "test_mview_deps_1",
            select(func.count(books.c.id).label("count")),
            with_data=True,
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(view1, or_replace=True))
            conn.execute(CreateMaterializedView(mview1, if_not_exists=True))

            views = get_all_views(conn)
            view_names = [v.name for v in views]

            assert "test_view_deps_1" in view_names
            assert "test_mview_deps_1" in view_names

            # Check materialized flag
            mview = next(v for v in views if v.name == "test_mview_deps_1")
            assert mview.is_materialized is True

            regular = next(v for v in views if v.name == "test_view_deps_1")
            assert regular.is_materialized is False

            conn.execute(DropView(view1, if_exists=True))
            conn.execute(DropMaterializedView(mview1, if_exists=True))

    def test_get_all_views_with_schema_filter(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting views filtered by schema."""
        authors = pg_one_to_many_tables["authors"]

        view = View(
            "test_view_schema_filter",
            select(authors.c.id),
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(view, or_replace=True))

            # Filter by public schema
            views = get_all_views(conn, schema="public")
            view_names = [v.name for v in views]

            assert "test_view_schema_filter" in view_names

            # Filter by non-existent schema
            views = get_all_views(conn, schema="nonexistent")
            assert len(views) == 0

            conn.execute(DropView(view, if_exists=True))

    def test_get_all_views_exclude_materialized(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting views excluding materialized views."""
        authors = pg_one_to_many_tables["authors"]

        view = View("test_view_excl", select(authors.c.id))
        mview = MaterializedView(
            "test_mview_excl",
            select(func.count(authors.c.id).label("cnt")),
            with_data=True,
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(view, or_replace=True))
            conn.execute(CreateMaterializedView(mview, if_not_exists=True))

            # Include materialized
            views = get_all_views(conn, include_materialized=True)
            names = [v.name for v in views]
            assert "test_view_excl" in names
            assert "test_mview_excl" in names

            # Exclude materialized
            views = get_all_views(conn, include_materialized=False)
            names = [v.name for v in views]
            assert "test_view_excl" in names
            assert "test_mview_excl" not in names

            conn.execute(DropView(view, if_exists=True))
            conn.execute(DropMaterializedView(mview, if_exists=True))


class TestGetViewDefinition:
    """Tests for get_view_definition function."""

    def test_get_view_definition(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting view definition."""
        authors = pg_one_to_many_tables["authors"]

        view = View(
            "test_view_def",
            select(authors.c.id, authors.c.name),
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(view, or_replace=True))

            definition = get_view_definition(conn, "test_view_def")

            assert definition is not None
            assert "authors" in definition.lower()

            conn.execute(DropView(view, if_exists=True))

    def test_get_view_definition_not_found(self, pg_engine: Engine) -> None:
        """Test getting definition for non-existent view."""
        with pg_engine.connect() as conn:
            definition = get_view_definition(conn, "nonexistent_view_xyz")
            assert definition is None


class TestGetViewDependencies:
    """Tests for get_view_dependencies function."""

    def test_get_view_dependencies(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting view dependencies."""
        authors = pg_one_to_many_tables["authors"]

        # Create base view
        base_view = View(
            "test_base_view_deps",
            select(authors.c.id, authors.c.name),
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(base_view, or_replace=True))

            # Create dependent view that references base view
            dependent_view = View(
                "test_dependent_view_deps",
                select(base_view.as_table().c.id),
            )
            conn.execute(CreateView(dependent_view, or_replace=True))

            # Get dependencies
            deps = get_view_dependencies(conn)

            # Find our dependency
            our_deps = [
                d for d in deps if d.dependent_view == "test_dependent_view_deps"
            ]

            assert len(our_deps) > 0
            # Should reference the base view
            ref_names = [d.referenced_view for d in our_deps]
            assert "test_base_view_deps" in ref_names

            conn.execute(DropView(dependent_view, if_exists=True, cascade=True))
            conn.execute(DropView(base_view, if_exists=True, cascade=True))


class TestGetDependencyOrder:
    """Tests for get_dependency_order function."""

    def test_get_dependency_order(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting views in dependency order."""
        authors = pg_one_to_many_tables["authors"]

        # Create views with dependencies
        view_a = View(
            "test_order_a",
            select(authors.c.id, authors.c.name),
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(view_a, or_replace=True))

            view_b = View(
                "test_order_b",
                select(view_a.as_table().c.id),
            )
            conn.execute(CreateView(view_b, or_replace=True))

            # Get dependency order
            ordered = get_dependency_order(conn, schema="public")
            names = [v.name for v in ordered]

            # view_a should come before view_b (dependencies first)
            if "test_order_a" in names and "test_order_b" in names:
                idx_a = names.index("test_order_a")
                idx_b = names.index("test_order_b")
                assert idx_a < idx_b

            conn.execute(DropView(view_b, if_exists=True, cascade=True))
            conn.execute(DropView(view_a, if_exists=True, cascade=True))


class TestGetReverseDependencies:
    """Tests for get_reverse_dependencies function."""

    def test_get_reverse_dependencies(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting reverse dependencies (views that depend on a view)."""
        authors = pg_one_to_many_tables["authors"]

        # Create base view
        base_view = View(
            "test_reverse_base",
            select(authors.c.id, authors.c.name),
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(base_view, or_replace=True))

            # Create dependent view
            dep_view = View(
                "test_reverse_dep",
                select(base_view.as_table().c.id),
            )
            conn.execute(CreateView(dep_view, or_replace=True))

            # Get reverse dependencies of base view
            dependents = get_reverse_dependencies(conn, "test_reverse_base")
            dep_names = [v.name for v in dependents]

            assert "test_reverse_dep" in dep_names

            conn.execute(DropView(dep_view, if_exists=True, cascade=True))
            conn.execute(DropView(base_view, if_exists=True, cascade=True))

    def test_get_reverse_dependencies_none(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test getting reverse dependencies when there are none."""
        authors = pg_one_to_many_tables["authors"]

        # Create standalone view with no dependents
        standalone = View(
            "test_standalone_view",
            select(authors.c.id),
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateView(standalone, or_replace=True))

            dependents = get_reverse_dependencies(conn, "test_standalone_view")
            assert len(dependents) == 0

            conn.execute(DropView(standalone, if_exists=True))