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_action_genre_rating(self):
        """Check that all 'Action' genre movies with rating >= 7 are included, and none below 7 are present."""
        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)
        action_rows = df[df['genre'] == 'Action']
        all_above_7 = (action_rows['rating'] >= 7).all() if not action_rows.empty else True
        # Check that all Action movies in table with rating >= 7 are present
        self.cursor.execute("SELECT * FROM movies WHERE genre = 'Action' AND rating >= 7")
        expected = self.cursor.fetchall()
        cond = all_above_7 and len(action_rows) == len(expected)
        _dynamic_test(self, cond, "All 'Action' movies with rating >= 7 are included, and none below 7 are present.", "Action genre filtering failed.")

    def test_comedy_genre_rating(self):
        """Check that all 'Comedy' genre movies with rating >= 6 are included, and none below 6 are present."""
        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)
        comedy_rows = df[df['genre'] == 'Comedy']
        all_above_6 = (comedy_rows['rating'] >= 6).all() if not comedy_rows.empty else True
        # Check that all Comedy movies in table with rating >= 6 are present
        self.cursor.execute("SELECT * FROM movies WHERE genre = 'Comedy' AND rating >= 6")
        expected = self.cursor.fetchall()
        cond = all_above_6 and len(comedy_rows) == len(expected)
        _dynamic_test(self, cond, "All 'Comedy' movies with rating >= 6 are included, and none below 6 are present.", "Comedy genre filtering failed.")

    def test_other_genre_rating(self):
        """Check that all non-Action, non-Comedy genre movies with rating >= 8 are included, and none below 8 are present."""
        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)
        other_rows = df[~df['genre'].isin(['Action', 'Comedy'])]
        all_above_8 = (other_rows['rating'] >= 8).all() if not other_rows.empty else True
        # Check that all such movies in table with rating >= 8 are present
        self.cursor.execute("SELECT * FROM movies WHERE genre NOT IN ('Action', 'Comedy') AND rating >= 8")
        expected = self.cursor.fetchall()
        cond = all_above_8 and len(other_rows) == len(expected)
        _dynamic_test(self, cond, "All non-Action, non-Comedy movies with rating >= 8 are included, and none below 8 are present.", "Other genre filtering failed.")

    def test_excluded_movies(self):
        """Check that no movies that fail the genre-specific criteria are included."""
        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)
        # Action < 7
        action_fail = df[(df['genre'] == 'Action') & (df['rating'] < 7)]
        # Comedy < 6
        comedy_fail = df[(df['genre'] == 'Comedy') & (df['rating'] < 6)]
        # Others < 8
        other_fail = df[(~df['genre'].isin(['Action', 'Comedy'])) & (df['rating'] < 8)]
        cond = action_fail.empty and comedy_fail.empty and other_fail.empty
        _dynamic_test(self, cond, "No movies included that fail the genre-specific rating criteria.", "One or more movies included that do not meet the genre-specific criteria.")

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: Genre-Specific Movie Selection

解答