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_only_blockbusters_selected(self):
        """Check only movies with box_office >= 100000000 or rating >= 8.5 are included."""
        df = self.get_query_result()
        non_blockbusters = df[(df['box_office'] < 100000000) & (df['rating'] < 8.5)]
        _dynamic_test(
            self,
            non_blockbusters.empty,
            "All returned movies meet the blockbuster criteria.",
            f"Found movies in result that do not meet blockbuster criteria: {non_blockbusters['title'].tolist()}"
        )

    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 'Blockbuster' for all returned rows."""
        df = self.get_query_result()
        not_blockbuster = df[df['blockbuster_status'] != 'Blockbuster']
        _dynamic_test(
            self,
            not_blockbuster.empty if df.shape[0] > 0 else True,
            "All blockbuster_status values are 'Blockbuster'.",
            f"Non-'Blockbuster' status found in result: {not_blockbuster['title'].tolist()}"
        )

    def test_no_extra_movies(self):
        """Check that movies not meeting blockbuster criteria are not included."""
        df = self.get_query_result()
        # Find all movies that should NOT be present
        self.cursor.execute("SELECT title FROM movies WHERE box_office < 100000000 AND rating < 8.5;")
        not_expected = {row[0] for row in self.cursor.fetchall()}
        result_titles = set(df['title'])
        intersection = not_expected & result_titles
        _dynamic_test(
            self,
            len(intersection) == 0,
            "No non-blockbuster movies are included in the result.",
            f"Non-blockbuster movies included: {intersection}"
        )

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

SQLにおけるCASE WHEN文を基礎的な条件分岐からデータ分析・レポーティングでの高度な活用例まで習得できる包括的なモジュールです。理論と実践的な課題を通じて、複雑なビジネス要件に対応する表現力豊かなクエリの作成方法を学びます。

CASE WHEN文の基本、構文、およびSQLクエリでの単純な条件付きロジックへの活用方法を学習します。

チャレンジ：ブロックバスター・フィルター

解答