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

A comprehensive module 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.

Challenge: Blockbuster Filter

Рішення