"""Tests for view behavior after INSERT/UPDATE/DELETE on underlying tables."""

from decimal import Decimal

from sqlalchemy import Table, delete, func, insert, select, update
from sqlalchemy.engine import Engine

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


class TestRegularViewUpdates:
    """Tests that regular views reflect changes immediately."""

    def test_view_reflects_insert(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test that view shows new rows after INSERT."""
        authors = pg_one_to_many_tables["authors"]
        books = pg_one_to_many_tables["books"]

        author_book_count = View(
            "author_book_count",
            select(
                authors.c.id,
                authors.c.name,
                func.count(books.c.id).label("book_count"),
            )
            .select_from(authors.outerjoin(books, authors.c.id == books.c.author_id))
            .group_by(authors.c.id, authors.c.name),
        )

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

            # Check initial count for Orwell (has 2 books)
            result = conn.execute(
                select(author_book_count.as_table()).where(
                    author_book_count.as_table().c.name == "George Orwell"
                )
            ).fetchone()
            assert result.book_count == 2

            # INSERT a new book for Orwell
            conn.execute(
                insert(books).values(id=100, title="Homage to Catalonia", author_id=1, price=13.99)
            )

            # View should immediately show 3 books
            result = conn.execute(
                select(author_book_count.as_table()).where(
                    author_book_count.as_table().c.name == "George Orwell"
                )
            ).fetchone()
            assert result.book_count == 3

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

    def test_view_reflects_update(
        self, pg_engine: Engine, pg_many_to_many_tables: dict[str, Table]
    ) -> None:
        """Test that view shows updated values after UPDATE."""
        students = pg_many_to_many_tables["students"]
        student_courses = pg_many_to_many_tables["student_courses"]

        student_gpa = View(
            "student_gpa",
            select(
                students.c.id,
                students.c.name,
                func.round(func.avg(student_courses.c.grade), 2).label("gpa"),
            )
            .select_from(
                students.join(student_courses, students.c.id == student_courses.c.student_id)
            )
            .group_by(students.c.id, students.c.name),
        )

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

            # Check Alice's initial GPA
            result = conn.execute(
                select(student_gpa.as_table()).where(student_gpa.as_table().c.name == "Alice")
            ).fetchone()
            initial_gpa = result.gpa

            # UPDATE Alice's grade in Database Systems from 3.8 to 4.0
            conn.execute(
                update(student_courses)
                .where(student_courses.c.student_id == 1)
                .where(student_courses.c.course_id == 1)
                .values(grade=Decimal("4.0"))
            )

            # View should show updated GPA
            result = conn.execute(
                select(student_gpa.as_table()).where(student_gpa.as_table().c.name == "Alice")
            ).fetchone()
            assert result.gpa > initial_gpa

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

    def test_view_reflects_delete(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test that view excludes deleted rows after DELETE."""
        books = pg_one_to_many_tables["books"]
        reviews = pg_one_to_many_tables["reviews"]

        book_review_count = View(
            "book_review_count",
            select(
                books.c.id,
                books.c.title,
                func.count(reviews.c.id).label("review_count"),
            )
            .select_from(books.outerjoin(reviews, books.c.id == reviews.c.book_id))
            .group_by(books.c.id, books.c.title),
        )

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

            # Check initial review count for "1984" (has 3 reviews)
            result = conn.execute(
                select(book_review_count.as_table()).where(
                    book_review_count.as_table().c.title == "1984"
                )
            ).fetchone()
            assert result.review_count == 3

            # DELETE one review for "1984"
            conn.execute(delete(reviews).where(reviews.c.id == 1))

            # View should immediately show 2 reviews
            result = conn.execute(
                select(book_review_count.as_table()).where(
                    book_review_count.as_table().c.title == "1984"
                )
            ).fetchone()
            assert result.review_count == 2

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


class TestMaterializedViewUpdates:
    """Tests that materialized views require explicit refresh."""

    def test_materialized_view_stale_after_insert(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test that materialized view shows stale data after INSERT until refreshed."""
        authors = pg_one_to_many_tables["authors"]
        books = pg_one_to_many_tables["books"]

        author_book_count_mv = MaterializedView(
            "author_book_count_mv",
            select(
                authors.c.id,
                authors.c.name,
                func.count(books.c.id).label("book_count"),
            )
            .select_from(authors.outerjoin(books, authors.c.id == books.c.author_id))
            .group_by(authors.c.id, authors.c.name),
            with_data=True,
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateMaterializedView(author_book_count_mv))

            # Check initial count for Orwell (has 2 books)
            result = conn.execute(
                select(author_book_count_mv.as_table()).where(
                    author_book_count_mv.as_table().c.name == "George Orwell"
                )
            ).fetchone()
            assert result.book_count == 2

            # INSERT a new book for Orwell
            conn.execute(
                insert(books).values(id=101, title="Down and Out in Paris", author_id=1, price=11.99)
            )

            # Materialized view still shows OLD count (stale data)
            result = conn.execute(
                select(author_book_count_mv.as_table()).where(
                    author_book_count_mv.as_table().c.name == "George Orwell"
                )
            ).fetchone()
            assert result.book_count == 2  # Still 2, not 3!

            # REFRESH the materialized view
            conn.execute(RefreshMaterializedView(author_book_count_mv))

            # Now it shows the updated count
            result = conn.execute(
                select(author_book_count_mv.as_table()).where(
                    author_book_count_mv.as_table().c.name == "George Orwell"
                )
            ).fetchone()
            assert result.book_count == 3  # Now shows 3

            conn.execute(DropMaterializedView(author_book_count_mv, if_exists=True))

    def test_materialized_view_stale_after_update(
        self, pg_engine: Engine, pg_many_to_many_tables: dict[str, Table]
    ) -> None:
        """Test that materialized view shows stale data after UPDATE until refreshed."""
        students = pg_many_to_many_tables["students"]
        student_courses = pg_many_to_many_tables["student_courses"]

        student_gpa_mv = MaterializedView(
            "student_gpa_mv_test",
            select(
                students.c.id,
                students.c.name,
                func.round(func.avg(student_courses.c.grade), 2).label("gpa"),
            )
            .select_from(
                students.join(student_courses, students.c.id == student_courses.c.student_id)
            )
            .group_by(students.c.id, students.c.name),
            with_data=True,
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateMaterializedView(student_gpa_mv))

            # Get Bob's initial GPA
            result = conn.execute(
                select(student_gpa_mv.as_table()).where(student_gpa_mv.as_table().c.name == "Bob")
            ).fetchone()
            initial_gpa = result.gpa

            # UPDATE Bob's grades to all 4.0
            conn.execute(
                update(student_courses)
                .where(student_courses.c.student_id == 2)
                .values(grade=Decimal("4.0"))
            )

            # Materialized view still shows old GPA
            result = conn.execute(
                select(student_gpa_mv.as_table()).where(student_gpa_mv.as_table().c.name == "Bob")
            ).fetchone()
            assert result.gpa == initial_gpa  # Unchanged

            # REFRESH
            conn.execute(RefreshMaterializedView(student_gpa_mv))

            # Now shows 4.0
            result = conn.execute(
                select(student_gpa_mv.as_table()).where(student_gpa_mv.as_table().c.name == "Bob")
            ).fetchone()
            assert result.gpa == Decimal("4.00")

            conn.execute(DropMaterializedView(student_gpa_mv, if_exists=True))

    def test_materialized_view_stale_after_delete(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test that materialized view shows stale data after DELETE until refreshed."""
        authors = pg_one_to_many_tables["authors"]
        books = pg_one_to_many_tables["books"]
        reviews = pg_one_to_many_tables["reviews"]

        author_count_mv = MaterializedView(
            "author_count_mv",
            select(func.count(authors.c.id).label("total_authors")),
            with_data=True,
        )

        with pg_engine.begin() as conn:
            conn.execute(CreateMaterializedView(author_count_mv))

            # Initial count
            result = conn.execute(select(author_count_mv.as_table())).fetchone()
            assert result.total_authors == 3

            # DELETE an author (need to delete reviews -> books -> author due to FK)
            # Author 2 has book 3 which has review 5
            conn.execute(delete(reviews).where(reviews.c.book_id == 3))
            conn.execute(delete(books).where(books.c.author_id == 2))
            conn.execute(delete(authors).where(authors.c.id == 2))

            # Materialized view still shows 3
            result = conn.execute(select(author_count_mv.as_table())).fetchone()
            assert result.total_authors == 3  # Stale!

            # REFRESH
            conn.execute(RefreshMaterializedView(author_count_mv))

            # Now shows 2
            result = conn.execute(select(author_count_mv.as_table())).fetchone()
            assert result.total_authors == 2

            conn.execute(DropMaterializedView(author_count_mv, if_exists=True))

    def test_materialized_view_refresh_via_method(
        self, pg_engine: Engine, pg_one_to_many_tables: dict[str, Table]
    ) -> None:
        """Test refreshing materialized view using the .refresh() method."""
        books = pg_one_to_many_tables["books"]

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

        with pg_engine.begin() as conn:
            conn.execute(CreateMaterializedView(book_count_mv))

            result = conn.execute(select(book_count_mv.as_table())).fetchone()
            assert result.total_books == 5

            # Add a new book
            conn.execute(
                insert(books).values(id=102, title="New Book", author_id=1, price=9.99)
            )

            # Still shows 5
            result = conn.execute(select(book_count_mv.as_table())).fetchone()
            assert result.total_books == 5

            # Use the .refresh() method
            book_count_mv.refresh(conn)

            # Now shows 6
            result = conn.execute(select(book_count_mv.as_table())).fetchone()
            assert result.total_books == 6

            conn.execute(DropMaterializedView(book_count_mv, if_exists=True))