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 setUp(self):
        with open("query.sql", "r") as file:
            self.query = file.read()

    def test_all_columns_returned(self):
        """Check that all columns from movies are returned."""
        self.cursor.execute(self.query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_columns = set(["id", "title", "genre", "rating", "release_year", "box_office"])
        _dynamic_test(
            self,
            set(columns) == expected_columns,
            "All columns from movies table are returned.",
            f"Returned columns are incorrect: {columns}"
        )

    def test_sci_fi_after_2010_first(self):
        """Check that only Sci-Fi movies released after 2010 are at the top."""
        self.cursor.execute(self.query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Find all Sci-Fi movies released after 2010
        sci_fi_after_2010 = df[(df['genre'] == 'Sci-Fi') & (df['release_year'] > 2010)]
        # These should all be at the top
        if not sci_fi_after_2010.empty:
            top_n = len(sci_fi_after_2010)
            top_rows = df.head(top_n)
            top_rows_check = (top_rows['genre'] == 'Sci-Fi') & (top_rows['release_year'] > 2010)
            condition = top_rows_check.all()
        else:
            # There are no Sci-Fi movies after 2010 in test data, so pass
            condition = True
        _dynamic_test(
            self,
            condition,
            "All prioritized Sci-Fi movies (after 2010) are at the top.",
            "Not all prioritized Sci-Fi movies are at the top."
        )

    def test_no_non_sci_fi_in_priority_group(self):
        """Check that no non-Sci-Fi/<=2010 movies are mixed in with the prioritized group."""
        self.cursor.execute(self.query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Find all Sci-Fi movies released after 2010
        sci_fi_after_2010 = df[(df['genre'] == 'Sci-Fi') & (df['release_year'] > 2010)]
        top_n = len(sci_fi_after_2010)
        if top_n > 0:
            top_rows = df.head(top_n)
            # Ensure all top rows are ONLY Sci-Fi and release_year > 2010
            valid = ((top_rows['genre'] == 'Sci-Fi') & (top_rows['release_year'] > 2010)).all()
        else:
            valid = True
        _dynamic_test(
            self,
            valid,
            "No non-Sci-Fi or <=2010 movies are in the prioritized group.",
            "There are movies in the prioritized group that do not match the condition."
        )

    def test_rest_ordered_by_release_year_desc(self):
        """Check that all other movies are ordered by release_year descending."""
        self.cursor.execute(self.query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        sci_fi_after_2010 = df[(df['genre'] == 'Sci-Fi') & (df['release_year'] > 2010)]
        top_n = len(sci_fi_after_2010)
        rest = df.iloc[top_n:]
        # The rest should be ordered by release_year descending
        release_years = rest['release_year'].tolist()
        is_desc = all(release_years[i] >= release_years[i+1] for i in range(len(release_years)-1))
        _dynamic_test(
            self,
            is_desc,
            "Remaining movies are ordered by release_year descending.",
            "Remaining movies are not ordered by release_year descending."
        )

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: Release Year Priority

解答