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 get_query_result(self):
        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]
        return pd.DataFrame(rows, columns=columns)

    def test_required_columns_present(self):
        """Check that result includes all required columns."""
        df = self.get_query_result()
        expected_columns = {'title', 'genre', 'rating', 'box_office', 'blockbuster_status'}
        missing = expected_columns - set(df.columns)
        _dynamic_test(
            self,
            not missing,
            "All required columns are present.",
            f"Missing columns: {missing}"
        )

    def test_blockbuster_status_label(self):
        """Check that blockbuster_status is correct for all rows."""
        df = self.get_query_result()
        for idx, row in df.iterrows():
            expected = 'Blockbuster' if row['box_office'] >= 100000000 or row['rating'] >= 8.5 else 'Standard'
            _dynamic_test(
                self,
                row['blockbuster_status'] == expected,
                f"Row {row['title']} correctly labeled as {expected}.",
                f"Row {row['title']} incorrectly labeled as {row['blockbuster_status']}."
            )

    def test_all_movies_included(self):
        """Check that all movies from the table are included in the result."""
        self.cursor.execute("SELECT COUNT(*) FROM movies;")
        total_movies = self.cursor.fetchone()[0]
        df = self.get_query_result()
        _dynamic_test(
            self,
            df.shape[0] == total_movies,
            "All movies are included in the result.",
            f"Expected {total_movies} movies, but got {df.shape[0]}."
        )

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):
    import re
    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: Blockbuster Filter

解答