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_only_unsold_products_returned(self):
        """Check that only products not in sales are returned."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        # Find all product_ids in sales
        self.cursor.execute("SELECT DISTINCT product_id FROM sales")
        sold_ids = {row[0] for row in self.cursor.fetchall()}
        # All returned product_ids must NOT be in sold_ids
        if not result_df.empty:
            not_sold = result_df['product_id'].apply(lambda x: x not in sold_ids)
            condition = not_sold.all()
        else:
            condition = True
        _dynamic_test(
            self,
            condition,
            "Only unsold products are returned.",
            "Some returned products have sales entries."
        )

    def test_all_unsold_products_included(self):
        """Check that all unsold products are included in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Find all product_ids in sales
        self.cursor.execute("SELECT DISTINCT product_id FROM sales")
        sold_ids = {row[0] for row in self.cursor.fetchall()}
        # Find all unsold products
        self.cursor.execute("SELECT * FROM products WHERE product_id NOT IN (SELECT product_id FROM sales)")
        expected_rows = self.cursor.fetchall()
        expected_columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(expected_rows, columns=expected_columns)
        # User result
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Compare
        missing = []
        for _, expected_row in expected_df.iterrows():
            expected_dict = expected_row.to_dict()
            mask = (result_df[list(expected_dict.keys())] == pd.Series(expected_dict)).all(axis=1)
            if not mask.any():
                missing.append(expected_dict)
        condition = len(missing) == 0
        failure_msg = f"Missing unsold products: {missing}" if missing else ""
        _dynamic_test(
            self,
            condition,
            "All unsold products are returned.",
            failure_msg
        )

    def test_all_product_columns_included(self):
        """Check that all columns from products table are included in the output."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        self.cursor.execute("SELECT * FROM products LIMIT 1")
        product_columns = [desc[0] for desc in self.cursor.description]
        # All product columns must be present in output
        missing = [col for col in product_columns if col not in columns]
        condition = len(missing) == 0
        failure_msg = f"Missing columns from products: {missing}" if missing else ""
        _dynamic_test(
            self,
            condition,
            "All columns from products table are included.",
            failure_msg
        )

    def test_no_sold_products_returned(self):
        """Check that no sold products are in the result."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT DISTINCT product_id FROM sales")
        sold_ids = {row[0] for row in self.cursor.fetchall()}
        if not result_df.empty:
            has_sold = result_df['product_id'].apply(lambda x: x in sold_ids)
            condition = not has_sold.any()
        else:
            condition = True
        _dynamic_test(
            self,
            condition,
            "No sold products are included in the result.",
            "Some sold products are present in the output."
        )

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 beginner-friendly course designed to teach SQL from the ground up, focusing on practical data analysis skills. Through engaging explanations, real-world examples, and hands-on challenges, learners will master querying, transforming, and analyzing data using SQL.

Learn the foundational concepts of relational databases and SQL. Explore how to retrieve, filter, and sort data from tables, setting the stage for deeper data analysis.

Dive into SQL's powerful aggregation features. Learn to summarize, group, and analyze data for deeper insights.

Unlock the power of combining data from multiple tables. Learn to perform different types of joins to answer complex business questions.

Master advanced SQL techniques for powerful data analysis, including subqueries, window functions, and data transformation.

Challenge: Unsold Products

Lösung

Challenge: Unsold Products

Lösung