import unittest
import psycopg2
import pandas as pd

class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()

    def test_all_rows_returned(self):
        """Check if all rows from the movies table are returned."""
        self.cursor.execute("SELECT COUNT(*) FROM movies;")
        expected_count = self.cursor.fetchone()[0]
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            f"Returned all {expected_count} rows from the movies table.",
            f"Returned {len(rows)} rows; expected {expected_count}."
        )

    def test_action_high_box_office_first(self):
        """Check that all 'Action' movies with box_office >= 50000000 appear before any other movie."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        df["is_action_high"] = (df["genre"] == "Action") & (df["box_office"] >= 50000000)
        # Index of first non-action-high
        non_action_high_indices = df.index[~df["is_action_high"]].tolist()
        first_non_action_high_idx = non_action_high_indices[0] if non_action_high_indices else len(df)
        # Check if any action-high appears after
        action_high_after = df.loc[first_non_action_high_idx:]["is_action_high"].any() if first_non_action_high_idx < len(df) else False
        _dynamic_test(
            self,
            not action_high_after,
            "All 'Action' movies with box_office >= 50000000 appear first.",
            "Some 'Action' movies with box_office >= 50000000 appear after other movies."
        )

    def test_other_movies_rating_desc(self):
        """Check that all non-'Action' high box office movies are ordered by rating descending."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        df["is_action_high"] = (df["genre"] == "Action") & (df["box_office"] >= 50000000)
        non_action_high = df[~df["is_action_high"]].copy()
        ratings = non_action_high["rating"].tolist()
        sorted_ratings = sorted(ratings, reverse=True)
        _dynamic_test(
            self,
            ratings == sorted_ratings,
            "All other movies are ordered by rating descending.",
            "Non-'Action' high box office movies are not ordered by rating descending."
        )

    def test_action_high_ordered_by_rating(self):
        """Check that within the 'Action' high box office group, movies are ordered by rating descending."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        df["is_action_high"] = (df["genre"] == "Action") & (df["box_office"] >= 50000000)
        action_high = df[df["is_action_high"]].copy()
        ratings = action_high["rating"].tolist()
        sorted_ratings = sorted(ratings, reverse=True)
        _dynamic_test(
            self,
            ratings == sorted_ratings,
            "'Action' movies with high box office are ordered by rating descending.",
            "'Action' high box office movies are not ordered by rating descending."
        )

    def test_no_action_low_box_office_first(self):
        """Check that no 'Action' movie with box_office < 50000000 appears before any non-'Action' movie."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        df["is_action_low"] = (df["genre"] == "Action") & (df["box_office"] < 50000000)
        df["is_non_action_high"] = ~((df["genre"] == "Action") & (df["box_office"] >= 50000000))
        action_low_idxs = df.index[df["is_action_low"]].tolist()
        non_action_high_idxs = df.index[df["is_non_action_high"]].tolist()
        if action_low_idxs and non_action_high_idxs:
            first_non_action_high_idx = non_action_high_idxs[0]
            action_low_before_non_action_high = any(idx < first_non_action_high_idx for idx in action_low_idxs)
        else:
            action_low_before_non_action_high = False
        _dynamic_test(
            self,
            not action_low_before_non_action_high,
            "No 'Action' movie with box_office < 50000000 appears before any non-'Action' movie.",
            "An 'Action' movie with box_office < 50000000 appears before a non-'Action' movie."
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

if __name__ == "__main__":
    unittest.main()


test_solution.py

A comprehensive course designed to help you master the CASE WHEN statement in SQL, from basic conditional logic to advanced use cases in data analysis and reporting. Through engaging theory and real-world tasks, you'll learn to write expressive, powerful queries that adapt to complex business requirements.

Learn the basics of the CASE WHEN statement, its syntax, and how to use it for simple conditional logic in SQL queries.

Delve deeper into complex CASE WHEN expressions, including nested conditions and multiple criteria.

Apply CASE WHEN in WHERE and ORDER BY clauses to filter and sort data conditionally.

Combine CASE WHEN with aggregate functions for advanced reporting and analytics.

Challenge: Custom Movie Ranking

解答